package packs

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strings"
)

// connectorRateLimitPackBatch carries the set of EndpointClass values declared
// by a single pack in the current manifest. It is populated during source-backed
// pack parsing and passed to the sweeper so the apply-side sweep and verify-side
// assertion share the same declared set.
type connectorRateLimitPackBatch struct {
	// declaredEndpointClasses is the set of EndpointClass values declared by the pack.
	declaredEndpointClasses []string
}

// newConnectorRateLimitPackSweeperComponent returns a synthetic pack-level install
// step that garbage-collects _System_ConnectorRateLimit rows owned by packName
// whose EndpointClass is no longer declared in the current manifest.
//
// Without this sweeper, a pack that drops a connector_rate_limit component
// leaves a stale row live: the throttler would still honor the old policy for
// that endpointClass even though the pack no longer declares it.
//
// The synthetic is folded into InstallPlan() and Verify() exclusively via
// SourceBackedPack.lifecycleClosureComponents() — it has no manifest business key,
// no payload, and does not participate in payload-hash computation.
func newConnectorRateLimitPackSweeperComponent(packName string, batch *connectorRateLimitPackBatch) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindConnectorRateLimit,
		BusinessKey:   JoinComponentBusinessKey(packName, "@sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applyConnectorRateLimitPackSweeper(ctx, deps.DB, batch, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifyConnectorRateLimitPackSweeper(ctx, deps.DB, batch, packName)
		},
	}
}

// applyConnectorRateLimitPackSweeper soft-deletes _System_ConnectorRateLimit
// rows owned by ownerPack whose EndpointClass is not in the declared set.
// There is no separate value table — the bucket state is in-process only.
func applyConnectorRateLimitPackSweeper(ctx context.Context, db *sql.DB, batch *connectorRateLimitPackBatch, ownerPack string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}

	orphans, err := queryOrphanConnectorRateLimitRows(ctx, db, batch, ownerPack)
	if err != nil {
		return err
	}
	if len(orphans) == 0 {
		return nil
	}

	for _, orphanID := range orphans {
		if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorRateLimit
SET IsDeleted = 1, LastModifiedDate = NOW()
WHERE Id = ? AND COALESCE(IsDeleted, 0) = 0`, orphanID); err != nil {
			return fmt.Errorf("soft-delete orphan connector rate-limit %s for pack %s: %w", orphanID, ownerPack, err)
		}
	}
	return nil
}

// verifyConnectorRateLimitPackSweeper asserts that no live _System_ConnectorRateLimit
// row exists for ownerPack whose EndpointClass is not in the declared set.
func verifyConnectorRateLimitPackSweeper(ctx context.Context, db *sql.DB, batch *connectorRateLimitPackBatch, ownerPack string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}

	orphans, err := queryOrphanConnectorRateLimitRows(ctx, db, batch, ownerPack)
	if err != nil {
		return err
	}
	if len(orphans) == 0 {
		return nil
	}
	sort.Strings(orphans)
	return fmt.Errorf("pack %s has %d undeclared connector_rate_limit row(s) still live: %s",
		ownerPack, len(orphans), strings.Join(orphans, ", "))
}

// queryOrphanConnectorRateLimitRows returns the Ids of _System_ConnectorRateLimit rows
// owned by ownerPack that are live (IsDeleted=0) but whose EndpointClass is NOT in
// the declared set of the current manifest.
func queryOrphanConnectorRateLimitRows(ctx context.Context, db *sql.DB, batch *connectorRateLimitPackBatch, ownerPack string) ([]string, error) {
	// _System_ConnectorRateLimit.OwnerPack is stored lower-trimmed by
	// connector_rate_limit_component.go. Normalize the lookup key the same way
	// so a mixed-case connector pack Name (e.g. "Acme-Connector") matches its
	// rows stored as "acme-connector". Without this, reapply after dropping a
	// connector_rate_limit component would miss the orphan row on the apply-side
	// soft-delete AND the verify-side drift assertion (codex Review PR #1537).
	// This is the single choke point both applyConnectorRateLimitPackSweeper and
	// verifyConnectorRateLimitPackSweeper funnel through.
	ownerPack = strings.ToLower(strings.TrimSpace(ownerPack))
	declaredSet := make(map[string]struct{})
	if batch != nil {
		for _, ec := range batch.declaredEndpointClasses {
			declaredSet[strings.ToLower(strings.TrimSpace(ec))] = struct{}{}
		}
	}

	rows, err := db.QueryContext(ctx, `
SELECT Id, EndpointClass
FROM _System_ConnectorRateLimit
WHERE OwnerPack = ? AND COALESCE(IsDeleted, 0) = 0`, ownerPack)
	if err != nil {
		return nil, fmt.Errorf("query live connector_rate_limit rows for pack sweep %s: %w", ownerPack, err)
	}
	defer rows.Close()

	var orphanIDs []string
	for rows.Next() {
		var id, endpointClass string
		if err := rows.Scan(&id, &endpointClass); err != nil {
			return nil, fmt.Errorf("scan connector_rate_limit row for pack sweep %s: %w", ownerPack, err)
		}
		if _, ok := declaredSet[strings.ToLower(strings.TrimSpace(endpointClass))]; ok {
			continue
		}
		orphanIDs = append(orphanIDs, id)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate connector_rate_limit rows for pack sweep %s: %w", ownerPack, err)
	}
	return orphanIDs, nil
}