package packs

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

	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

// SQL audit summary — RouteVersion handling in this file
// =======================================================
// This file contains TWO code paths that reference _System_ConnectorEvent.RouteVersion:
//
//  1. Apply (applyConnectorInboundRoutePackSweep): per-candidate queue-reference check
//     inside the row-level loop uses:
//       AND (RouteVersion = ? OR RouteVersion IS NULL)
//     Positional parameter ? = r.version (the candidate route's Version value).
//
//  2. Verify (verifyConnectorInboundRoutePackSweep): per-undeclared-EventKind queue-
//     reference check uses a JOIN condition:
//       AND (r.Version = e.RouteVersion OR e.RouteVersion IS NULL)
//
// BOTH sites explicitly handle the 3-state schema (NULL legacy / matching Version /
// non-matching Version). Strict equality `RouteVersion = ?` alone is FALSE for NULL
// in SQL (NULL comparisons always return UNKNOWN), so the explicit
// `OR RouteVersion IS NULL` clause is required to preserve in-flight legacy events
// that pre-date the RouteVersion column. Do NOT remove or simplify the IS NULL clause.

// connectorInboundRouteOrphanSweeperGroupKey is the distinct ApplyGroupKey for the
// orphan sweeper (pack-level: removes routes no longer declared in the manifest).
// Must differ from connectorInboundRouteIncarnationSweeperGroupKey so both sweepers
// survive InstallPlan.Apply's per-group dedup.
const connectorInboundRouteOrphanSweeperGroupKey = "connector_inbound_route_orphan"

// connectorInboundRouteBatch carries the set of EventKind values declared by a
// single connector pack in the current manifest. It is populated during
// source-backed pack parsing (in source_backed_pack.go buildComponents) and
// passed to the orphan sweeper so apply-side sweep and verify-side assertion
// share the same declared set.
//
// The batch uses the direct-from-component typed-payload pattern (per §6 sketch):
// buildComponents accumulates cc.payload.EventKind directly, avoiding string-split
// on BusinessKey.
type connectorInboundRouteBatch struct {
	declaredEventKinds []string
}

// newConnectorInboundRoutePackSweeperComponent is the synthetic pack-level
// component that guarantees stale _System_ConnectorInboundRoute rows owned by
// this pack (identified by Namespace=packName) get soft-deleted on reapply
// when the new manifest no longer declares them.
//
// Without this sweeper, dropping an eventKind from a connector pack manifest
// would leave the old route row live and the Worker would continue routing
// events of that kind to the (now-dropped) flow. This sweeper fires on EVERY
// pack install regardless of whether the current manifest declares any
// connector_inbound_route components.
//
// Pattern mirrors newConnectorCapabilityPackSweeperComponent
// (connector_capability_sweeper.go).
//
// Synthetic component: NOT exposed via SourceBackedPack.Components() or
// DeclaredComponents() — no payload hash contribution, no manifest business
// key. Folded into InstallPlan/Verify exclusively via lifecycleClosureComponents().
func newConnectorInboundRoutePackSweeperComponent(packName string, declaredEventKinds []string) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindConnectorInboundRoute,
		BusinessKey:   JoinComponentBusinessKey(packName, "@route-sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		applyGroupKey:     connectorInboundRouteOrphanSweeperGroupKey,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applyConnectorInboundRoutePackSweep(ctx, deps.DB, packName, declaredEventKinds)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifyConnectorInboundRoutePackSweep(ctx, deps.DB, packName, declaredEventKinds)
		},
	}
}

// applyConnectorInboundRoutePackSweep soft-deletes _System_ConnectorInboundRoute
// rows (across all Versions) owned by packName (Namespace=packName) whose EventKind
// is not in the declared set. When declaredEventKinds is empty, every undropped row
// for this namespace is a candidate for soft-delete (the pack has dropped all routes).
//
// Because routes are versioned and append-only, dropping an eventKind from the
// manifest must soft-delete ALL version rows for that eventKind — they are all stale
// once the manifest no longer declares the route.  This sweeper handles GC for DROPPED
// eventKinds only.
//
// # Queue-reference guard (B6 — codex R8 BLOCK)
//
// A candidate row is skipped (NOT soft-deleted) if any non-terminal
// _System_ConnectorEvent queue row still references its
// (InstalledPackId, EventKind, Version) triple. The queue row must drain to a
// terminal status (processed or dead) before the route row can be reclaimed.
// Version-level GC for OLD versions of LIVE eventKinds uses the same guard
// and will be addressed in a follow-up once the sweeper covers that dimension too.
//
// # Legacy NULL-RouteVersion guard (codex R10 BLOCK)
//
// Pre-upgrade _System_ConnectorEvent rows created before the RouteVersion
// column existed carry RouteVersion=NULL. Strict equality `RouteVersion = ?`
// is FALSE for these rows in SQL (NULL comparisons always return UNKNOWN), so
// without an explicit NULL clause the sweeper would delete the route row while
// a pending legacy event still needs it for its fallback dispatch lookup.
//
// The guard therefore also counts non-terminal events with RouteVersion IS NULL
// for the same (InstalledPackId, EventKind) pair. If any such legacy event is
// pending, the route row is preserved regardless of its Version value.
// Once all legacy events terminate, the next sweeper run can reclaim the row.
func applyConnectorInboundRoutePackSweep(ctx context.Context, db *sql.DB, packName string, declaredEventKinds []string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	declaredSet := connectorInboundRouteEventKindSet(declaredEventKinds)
	query := `
SELECT Id, InstalledPackId, EventKind, Version
  FROM _System_ConnectorInboundRoute
 WHERE Namespace = ?
   AND COALESCE(IsDeleted, 0) = 0`
	args := []any{packName}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		// A stopped replacement owns a new incarnation. Manifest-difference
		// cleanup is candidate-local until activation; touching the still-active
		// incarnation would make a failed candidate mutate runtime authority.
		query += " AND InstalledPackId = ?"
		args = append(args, candidateInstallID)
	}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("query connector inbound routes for sweep apply (pack %s): %w", packName, err)
	}
	defer rows.Close()

	type candidateRow struct {
		id              string
		installedPackID string
		eventKind       string
		version         int64
	}
	var candidates []candidateRow
	for rows.Next() {
		var r candidateRow
		if err := rows.Scan(&r.id, &r.installedPackID, &r.eventKind, &r.version); err != nil {
			return fmt.Errorf("scan connector inbound route row for sweep apply: %w", err)
		}
		if _, ok := declaredSet[strings.ToLower(strings.TrimSpace(r.eventKind))]; ok {
			continue
		}
		candidates = append(candidates, r)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate connector inbound route rows for sweep apply: %w", err)
	}
	if len(candidates) == 0 {
		return nil
	}

	// Sort by Id for deterministic processing order.
	sort.Slice(candidates, func(i, j int) bool { return candidates[i].id < candidates[j].id })

	for _, r := range candidates {
		// Skip rows still referenced by non-terminal queue events.
		// The OR RouteVersion IS NULL arm handles legacy pre-upgrade events whose
		// RouteVersion column was NULL at insert time; strict equality is FALSE for
		// NULL in SQL, so without this guard the route row would be swept while the
		// legacy event is still pending. NULL events match at (InstalledPackId,
		// EventKind) level — all version rows of that key are preserved.
		var queueCount int
		err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?
   AND LOWER(TRIM(EventKind)) = ?
   AND (RouteVersion = ? OR RouteVersion IS NULL /* LEGACY NULL-AWARE */)
   AND Status IN ('pending', 'processing', 'failed')`, r.installedPackID, strings.ToLower(strings.TrimSpace(r.eventKind)), r.version).Scan(&queueCount)
		if err != nil {
			return fmt.Errorf("check queue references for connector inbound route %s (pack %s): %w", r.id, packName, err)
		}
		if queueCount > 0 {
			// Queue still draining: preserve row for backlog (IsDeleted=0 so
			// ResolveFlowByVersion can still find it by exact Version), but clear
			// IsCurrent so ResolveFlow's live-traffic filter (AND IsCurrent=1)
			// stops routing new webhook traffic to this dropped-eventKind row
			// (B19 fix — codex R16 BLOCK).
			if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsCurrent = 0, LastModifiedDate = NOW()
 WHERE Id = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND IsCurrent = 1`, r.id); err != nil {
				return fmt.Errorf("clear IsCurrent for preserved connector inbound route %s (pack %s): %w", r.id, packName, err)
			}
			continue
		}
		if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsDeleted = 1, IsCurrent = 0, LastModifiedDate = NOW()
 WHERE Id = ?
   AND COALESCE(IsDeleted, 0) = 0`, r.id); err != nil {
			return fmt.Errorf("soft-delete orphan connector inbound route %s for pack %s: %w", r.id, packName, err)
		}
	}
	return nil
}

// verifyConnectorInboundRoutePackSweep asserts that no live (non-deleted)
// _System_ConnectorInboundRoute row exists for packName whose EventKind is not
// in the declared set, excluding rows intentionally preserved by the queue-
// reference guard (i.e., undeclared EventKind rows that still have non-terminal
// queue events referencing them).
//
// This mirrors the apply-side queue-reference guard: a route row for a dropped
// EventKind is legitimately live while its queue events drain. Only rows with
// no queue references are counted as leftover errors.
func verifyConnectorInboundRoutePackSweep(ctx context.Context, db *sql.DB, packName string, declaredEventKinds []string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	declaredSet := connectorInboundRouteEventKindSet(declaredEventKinds)
	query := `
SELECT DISTINCT EventKind
  FROM _System_ConnectorInboundRoute
 WHERE Namespace = ?
   AND COALESCE(IsDeleted, 0) = 0`
	args := []any{packName}
	candidateInstallID, stoppedCandidate := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName)
	if stoppedCandidate {
		query += " AND InstalledPackId = ?"
		args = append(args, candidateInstallID)
	}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("query connector inbound routes for sweep verify (pack %s): %w", packName, err)
	}
	defer rows.Close()

	var undeclared []string
	for rows.Next() {
		var eventKind string
		if err := rows.Scan(&eventKind); err != nil {
			return fmt.Errorf("scan connector inbound route row for sweep verify: %w", err)
		}
		if _, ok := declaredSet[strings.ToLower(strings.TrimSpace(eventKind))]; ok {
			continue
		}
		undeclared = append(undeclared, eventKind)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate connector inbound route rows for sweep verify: %w", err)
	}
	if len(undeclared) == 0 {
		return nil
	}

	// For each undeclared EventKind, skip if still referenced by non-terminal queue events.
	// The OR e.RouteVersion IS NULL arm handles legacy pre-upgrade events whose
	// RouteVersion was NULL at insert; strict equality misses them and would cause
	// the verify to incorrectly flag the route as orphaned while a legacy event is
	// still pending. NULL events match at (InstalledPackId, EventKind) level —
	// any version row for that key is a valid preserving reference.
	var leftover []string
	for _, ek := range undeclared {
		var queueCount int
		queueQuery := `
SELECT COUNT(*)
  FROM _System_ConnectorEvent e
  JOIN _System_ConnectorInboundRoute r
    ON r.InstalledPackId = e.InstalledPackId
   AND r.EventKind       = LOWER(TRIM(e.EventKind))
   AND (r.Version = e.RouteVersion OR e.RouteVersion IS NULL /* LEGACY NULL-AWARE */)
 WHERE r.Namespace = ?
   AND r.EventKind = ?
   AND COALESCE(r.IsDeleted, 0) = 0
	   AND e.Status IN ('pending', 'processing', 'failed')`
		queueArgs := []any{packName, strings.ToLower(strings.TrimSpace(ek))}
		if stoppedCandidate {
			queueQuery += " AND r.InstalledPackId = ?"
			queueArgs = append(queueArgs, candidateInstallID)
		}
		if err := db.QueryRowContext(ctx, queueQuery, queueArgs...).Scan(&queueCount); err != nil {
			return fmt.Errorf("check queue references for undeclared eventKind %s (pack %s): %w", ek, packName, err)
		}
		if queueCount > 0 {
			continue // queue still draining — apply-side correctly preserved this row
		}
		leftover = append(leftover, ek)
	}

	if len(leftover) > 0 {
		sort.Strings(leftover)
		return fmt.Errorf("pack %s has %d undeclared connector inbound route(s) still live (not queue-referenced): %s",
			packName, len(leftover), strings.Join(leftover, ", "))
	}
	return nil
}

func connectorInboundRouteEventKindSet(declaredEventKinds []string) map[string]struct{} {
	set := make(map[string]struct{}, len(declaredEventKinds))
	for _, ek := range declaredEventKinds {
		set[strings.ToLower(strings.TrimSpace(ek))] = struct{}{}
	}
	return set
}