package packs

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

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

// SQL audit summary — RouteVersion handling in this file
// =======================================================
// This file contains TWO SQL queries that reference _System_ConnectorEvent.RouteVersion:
//
//  1. Apply (applyConnectorInboundRouteIncarnationSweep): the NOT EXISTS subquery
//     inside the UPDATE statement uses:
//       AND (e.RouteVersion = _System_ConnectorInboundRoute.Version OR e.RouteVersion IS NULL)
//
//  2. Verify (verifyConnectorInboundRouteIncarnationSweep): the NOT EXISTS subquery
//     inside the SELECT COUNT(*) uses:
//       AND (e.RouteVersion = r.Version OR e.RouteVersion IS NULL)
//
// BOTH sites explicitly handle the 3-state schema (NULL legacy / matching Version /
// non-matching Version). Strict equality `e.RouteVersion = r.Version` alone is FALSE
// for NULL in SQL (NULL comparisons always return UNKNOWN), so the explicit
// `OR e.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.

// connectorInboundRouteIncarnationSweeperGroupKey is the distinct ApplyGroupKey for the
// incarnation sweeper (cross-incarnation: removes routes belonging to old PackInstallIds).
// Must differ from connectorInboundRouteOrphanSweeperGroupKey so both sweepers
// survive InstallPlan.Apply's per-group dedup.
const connectorInboundRouteIncarnationSweeperGroupKey = "connector_inbound_route_incarnation"

// newConnectorInboundRouteIncarnationSweeper returns a synthetic pack-level
// install step that soft-deletes _System_ConnectorInboundRoute rows whose
// InstalledPackId is NOT present as a live (active, installing, or removing)
// PackInstallId in _System_InstalledPack for this pack.
//
// # Why this sweeper exists (Skeptic HIGH #3)
//
// When a connector pack is uninstalled and then reinstalled, a fresh PackInstallId
// is minted for the new incarnation. The old incarnation's route rows remain in
// _System_ConnectorInboundRoute unless explicitly soft-deleted. Without this sweeper,
// loadConnectorInboundRouteFlowName's JOIN might not match (the old InstalledPackId
// no longer matches any active pack row), but the stale rows consume space and
// confuse LiveProjection verification.
//
// # Status='installing' inclusion (Challenger HIGH #1)
//
// The install lifecycle in source_pack_ops.go (markInstalledPackInstalling →
// individual component Apply → sweepers → markInstalledPackActive) means the
// CURRENT pack's row has Status='installing' when this sweeper runs. Excluding
// 'installing' from the subquery would cause the sweeper to soft-delete the rows
// just written by connectorInboundRouteComponent.Apply. The status set
// ('active', 'installing', 'removing') matches the full in-flight lifecycle.
//
// The MANDATORY regression test is:
//   - TestIncarnationSweeper_DoesNotWipeCurrentInstall_DuringInstallingStatus
//
// Both sweepers are needed:
//   - The pack sweeper (connector_inbound_route_pack_sweeper.go) handles "manifest
//     dropped some routes" within the current incarnation.
//   - This sweeper handles "pack was reborn / old incarnations remain in DB."
func newConnectorInboundRouteIncarnationSweeper(packName string) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindConnectorInboundRoute,
		BusinessKey:   JoinComponentBusinessKey(packName, "@incarnation-sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		applyGroupKey:     connectorInboundRouteIncarnationSweeperGroupKey,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applyConnectorInboundRouteIncarnationSweep(ctx, deps.DB, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifyConnectorInboundRouteIncarnationSweep(ctx, deps.DB, packName)
		},
	}
}

// applyConnectorInboundRouteIncarnationSweep soft-deletes all
// _System_ConnectorInboundRoute rows owned by packName whose InstalledPackId
// is not in the set of live (active, installing, or removing) PackInstallIds
// for this pack, AND that are not still referenced by a non-terminal
// _System_ConnectorEvent queue row.
//
// # Queue-reference guard (B6 — codex R8 BLOCK)
//
// A versioned route row must not be soft-deleted while any queue event
// with (InstalledPackId, EventKind, RouteVersion) = (r.InstalledPackId,
// r.EventKind, r.Version) is still in a non-terminal status (pending,
// processing, or failed-and-retryable). Deleting the route before the
// queue drains causes the dispatcher to receive ErrRouteVersionSwept for
// previously-accepted webhook events, making backlog unrecoverable across
// normal pack upgrades.
//
// # Legacy NULL-RouteVersion guard (codex R10 BLOCK)
//
// Pre-upgrade _System_ConnectorEvent rows created before the RouteVersion
// column existed carry RouteVersion=NULL. Strict equality `e.RouteVersion =
// r.Version` is FALSE for these rows, so without an explicit NULL clause the
// sweeper would delete the only live route row a pending legacy event needs
// for its live-ResolveFlow fallback lookup at dispatch time.
//
// The guard therefore treats any non-terminal event with RouteVersion IS NULL
// as a reference at the (InstalledPackId, EventKind) level — all version rows
// of that key are preserved while at least one legacy NULL event is pending.
// Once the legacy event terminates (processed/dead) or is backfilled to a
// concrete RouteVersion, the next sweeper run can reclaim the row.
//
// Once all referencing queue rows reach a terminal status (processed or
// dead), the orphan route row is eligible for GC on the next sweeper run.
func applyConnectorInboundRouteIncarnationSweep(ctx context.Context, db *sql.DB, packName string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	// A stale incarnation may still be referenced by non-terminal queue work and
	// therefore must remain readable by its exact (install, kind, version) key.
	// It must never remain eligible for new live traffic, however.  Demote every
	// stale incarnation before the queue-reference-aware delete below; preserving
	// a backlog row and preserving IsCurrent are different facts.
	candidateGuard := ""
	demoteArgs := []any{packName, packName}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		candidateGuard = " AND InstalledPackId <> ?"
		demoteArgs = append(demoteArgs, candidateInstallID)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsCurrent = 0, LastModifiedDate = NOW()
 WHERE Namespace = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND COALESCE(IsCurrent, 0) = 1
   AND InstalledPackId NOT IN (
       SELECT PackInstallId
         FROM _System_InstalledPack
        WHERE Name = ?
          AND COALESCE(IsDeleted, 0) = 0
          AND Status IN ('active', 'installing', 'removing')
          AND COALESCE(PackInstallId, '') <> ''
   )`+candidateGuard, demoteArgs...); err != nil {
		return fmt.Errorf("demote stale connector inbound route incarnation for pack %s: %w", packName, err)
	}
	// NOT EXISTS guard ensures only rows with no live queue references are
	// soft-deleted here.  Rows still referenced by non-terminal queue events are
	// NOT touched by this UPDATE — they are preserved by the NOT EXISTS clause.
	// IsCurrent is also cleared to 0 on soft-delete for consistency (these rows
	// are stale incarnation rows; they should not appear as live routes in
	// ResolveFlow even if IsDeleted somehow gets reset — B19 defence-in-depth).
	//
	// Note: incarnation sweeper does NOT need a "queue-referenced → IsCurrent=0
	// only" path because the NOT EXISTS clause already excludes those rows from
	// this UPDATE entirely.  The pack sweeper handles the IsCurrent=0 preservation
	// path for drop-eventKind rows within the current incarnation.
	deleteArgs := []any{packName, packName}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		deleteArgs = append(deleteArgs, candidateInstallID)
	}
	_, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsDeleted = 1, IsCurrent = 0, LastModifiedDate = NOW()
 WHERE Namespace = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND InstalledPackId NOT IN (
       SELECT PackInstallId
         FROM _System_InstalledPack
        WHERE Name = ?
          AND COALESCE(IsDeleted, 0) = 0
          AND Status IN ('active', 'installing', 'removing')
          AND COALESCE(PackInstallId, '') <> ''
   )`+candidateGuard+`
   AND NOT EXISTS (
       SELECT 1
         FROM _System_ConnectorEvent e
        WHERE e.InstalledPackId = _System_ConnectorInboundRoute.InstalledPackId
          AND LOWER(TRIM(e.EventKind)) = _System_ConnectorInboundRoute.EventKind
          AND (e.RouteVersion  = _System_ConnectorInboundRoute.Version OR e.RouteVersion IS NULL /* LEGACY NULL-AWARE */)
          AND e.Status IN ('pending', 'processing', 'failed')
   )`,
		deleteArgs...,
	)
	if err != nil {
		return fmt.Errorf("connector inbound route incarnation sweep for pack %s: %w", packName, err)
	}
	return nil
}

// verifyConnectorInboundRouteIncarnationSweep asserts that no live
// _System_ConnectorInboundRoute row exists for packName whose InstalledPackId
// is not associated with a live (active, installing, or removing) pack
// installation, excluding rows still referenced by non-terminal queue events
// (those are intentionally preserved by the apply-side queue-reference guard).
func verifyConnectorInboundRouteIncarnationSweep(ctx context.Context, db *sql.DB, packName string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	candidateGuard := ""
	args := []any{packName, packName}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		candidateGuard = " AND r.InstalledPackId <> ?"
		args = append(args, candidateInstallID)
	}
	var count int
	err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorInboundRoute r
 WHERE r.Namespace = ?
   AND COALESCE(r.IsDeleted, 0) = 0
   AND r.InstalledPackId NOT IN (
       SELECT PackInstallId
         FROM _System_InstalledPack
        WHERE Name = ?
          AND COALESCE(IsDeleted, 0) = 0
          AND Status IN ('active', 'installing', 'removing')
          AND COALESCE(PackInstallId, '') <> ''
   )`+candidateGuard+`
   AND NOT EXISTS (
       SELECT 1
         FROM _System_ConnectorEvent e
        WHERE e.InstalledPackId = r.InstalledPackId
          AND LOWER(TRIM(e.EventKind)) = r.EventKind
          AND (e.RouteVersion  = r.Version OR e.RouteVersion IS NULL /* LEGACY NULL-AWARE */)
          AND e.Status IN ('pending', 'processing', 'failed')
   )`,
		args...,
	).Scan(&count)
	if err != nil {
		return fmt.Errorf("verify connector inbound route incarnation sweep for pack %s: %w", packName, err)
	}
	if count > 0 {
		return fmt.Errorf("pack %s has %d connector inbound route row(s) belonging to stale incarnation(s) still live (not queue-referenced); incarnation sweeper should have removed them", packName, count)
	}
	return nil
}