package packs

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strings"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

// connectorInboundRoutePayload is the V2 single-route JSON payload shape for a
// connector_inbound_route component. Each component declares EXACTLY ONE route
// (ONE component = ONE route, per V2 fix from done-sweep agent D).
//
// Namespace MUST equal the containing pack's Name (Challenger MED #4 — drop
// optional-with-default; mirrors connector_capability_component.go:18-21).
// EventKind is canonicalised lowercase+trim at parse time (Challenger MED #5).
type connectorInboundRoutePayload struct {
	Namespace             string                                        `json:"namespace"`         // REQUIRED; must equal packName
	EventKind             string                                        `json:"eventKind"`         // REQUIRED non-empty; canonicalised lowercase+trim
	FlowDeveloperName     string                                        `json:"flowDeveloperName"` // REQUIRED non-empty
	InvocationMode        string                                        `json:"invocationMode,omitempty"`
	BatchInputName        string                                        `json:"batchInputName,omitempty"`
	BatchResultOutputName string                                        `json:"batchResultOutputName,omitempty"`
	MaxBatchSize          int                                           `json:"maxBatchSize,omitempty"`
	QueueClass            string                                        `json:"queueClass,omitempty"`
	FlowExecutionFilter   *connector_inbound_routing.PayloadValueFilter `json:"flowExecutionFilter,omitempty"`
}

// connectorInboundRouteRow is the DB row shape for _System_ConnectorInboundRoute.
// Used by LiveProjection and by the VerifyComponent implementation.
type connectorInboundRouteRow struct {
	ID                    string
	Namespace             string
	InstalledPackId       string
	EventKind             string
	Version               int64
	FlowDeveloperName     string
	InvocationMode        string
	BatchInputName        string
	BatchResultOutputName string
	MaxBatchSize          int
	QueueClass            string
	FlowExecutionFilter   string
	IsCurrent             int
	IsDeleted             int
}

// connectorInboundRouteComponent is the PackComponent for a single declared
// connector inbound route. Each component holds exactly one (EventKind →
// FlowDeveloperName) mapping; the pack manifest declares N such components for N routes.
//
// V2 fix (done-sweep Agent D): replaced the old multi-route Routes[] shape with
// a per-component single-route shape, mirroring connector_rate_limit's per-
// endpointClass and connector_capability's per-(namespace,kind) shapes.
type connectorInboundRouteComponent struct {
	packName          string
	manifestComponent ManifestComponent
	payload           connectorInboundRoutePayload
	rawPayload        []byte // verbatim; returned by ExpectedPayload (per capability precedent, avoids re-marshal key-order drift — Challenger LOW #7)
}

// Compile-time check that connectorInboundRouteComponent implements the
// optional componentCustomVerifier interface used for drift detection when
// ExpectedProjection deliberately omits Id+InstalledPackId (Agent A Gap 4 fix).
var _ componentCustomVerifier = (*connectorInboundRouteComponent)(nil)

// newConnectorInboundRouteComponent parses the JSON payload and constructs a
// connectorInboundRouteComponent. It validates required fields and canonicalises
// EventKind (lowercase+trim) so the stored column form matches the Id computation,
// preventing case-skew between Id and column (Challenger MED #5).
func newConnectorInboundRouteComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	var p connectorInboundRoutePayload
	dec := json.NewDecoder(bytes.NewReader(payload))
	dec.DisallowUnknownFields()
	if err := dec.Decode(&p); err != nil {
		return nil, fmt.Errorf("parse connector_inbound_route payload %s: %w", componentPayloadLabel(component), err)
	}
	if err := ensureJSONEOF(dec); err != nil {
		return nil, fmt.Errorf("connector_inbound_route payload %s has trailing data: %w", componentPayloadLabel(component), err)
	}
	// Canonicalise fields at parse time so they are stored canonical from day 0.
	p.Namespace = strings.TrimSpace(p.Namespace)
	p.EventKind = strings.ToLower(strings.TrimSpace(p.EventKind))
	p.FlowDeveloperName = strings.TrimSpace(p.FlowDeveloperName)
	spec, err := connectorInboundRoutePayloadSpec(p)
	if err != nil {
		return nil, fmt.Errorf("connector_inbound_route payload %s has invalid execution spec: %w", componentPayloadLabel(component), err)
	}
	p.InvocationMode = spec.InvocationMode
	p.BatchInputName = spec.BatchInputName
	p.BatchResultOutputName = spec.BatchResultOutputName
	p.MaxBatchSize = spec.MaxBatchSize
	p.QueueClass = spec.QueueClass
	p.FlowExecutionFilter = spec.FlowExecutionFilter
	return &connectorInboundRouteComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           p,
		rawPayload:        append([]byte(nil), payload...),
	}, nil
}

func connectorInboundRoutePayloadSpec(p connectorInboundRoutePayload) (connector_inbound_routing.RouteExecutionSpec, error) {
	return connector_inbound_routing.NormalizeRouteExecutionSpec(connector_inbound_routing.RouteExecutionSpec{
		FlowDeveloperName:     p.FlowDeveloperName,
		InvocationMode:        p.InvocationMode,
		BatchInputName:        p.BatchInputName,
		BatchResultOutputName: p.BatchResultOutputName,
		MaxBatchSize:          p.MaxBatchSize,
		QueueClass:            p.QueueClass,
		FlowExecutionFilter:   p.FlowExecutionFilter,
	})
}

func (c *connectorInboundRouteComponent) Kind() ComponentKind {
	return ComponentKindConnectorInboundRoute
}

func (c *connectorInboundRouteComponent) PackName() string {
	return c.packName
}

// BusinessKey is JoinComponentBusinessKey(packName, eventKind) so each route
// component has a unique identity key within the manifest (V2 fix per Agent D /
// Agent A Gap 1). The (kind, businessKey) identity check at pack_source.go:880
// then catches duplicate eventKind declarations across components.
func (c *connectorInboundRouteComponent) BusinessKey() string {
	return JoinComponentBusinessKey(c.packName, c.payload.EventKind)
}

func (c *connectorInboundRouteComponent) Manageability() Manageability {
	return ManageabilityLocked
}

// DependsOn returns the component references this route declares as
// dependencies in its pack manifest. Returns a detached copy of the original
// ManifestComponent slice so the route→flow integrity edge (per #1606 Phase=
// Automation per S3) is preserved across DeclaredComponents() and
// PayloadHashV1() projections.
//
// Mirrors connector_capability_component.go's manifestComponent field pattern
// (connector_capability_component.go:40-46, 56-62).
func (c *connectorInboundRouteComponent) DependsOn() []ComponentReference {
	return cloneSourceSlice(c.manifestComponent.DependsOn)
}

// ExpectedPayload returns the verbatim rawPayload captured at parse time.
// This matches connectorCapabilityComponent's precedent (connector_capability_component.go
// line 81-83): returning the raw bytes avoids drift-detection false positives
// from key-order changes in re-marshalled JSON (Challenger LOW #7).
func (c *connectorInboundRouteComponent) ExpectedPayload() ([]byte, error) {
	return append([]byte(nil), c.rawPayload...), nil
}

// ExpectedProjection returns the single row that Apply will write, deliberately
// omitting Id, InstalledPackId, and Version because they are resolved at Apply time
// from DB (not known until install). VerifyComponent handles Id+InstalledPackId+Version
// drift separately (Agent A Gap 4 fix — mirrors connector_capability_component.go:92).
func (c *connectorInboundRouteComponent) ExpectedProjection() (any, error) {
	return connectorInboundRouteRow{
		// Id, InstalledPackId, and Version are resolved at Apply time; omitted here.
		// VerifyComponent compares these fields separately.
		Namespace:             c.packName,
		EventKind:             c.payload.EventKind,
		FlowDeveloperName:     c.payload.FlowDeveloperName,
		InvocationMode:        c.payload.InvocationMode,
		BatchInputName:        c.payload.BatchInputName,
		BatchResultOutputName: c.payload.BatchResultOutputName,
		MaxBatchSize:          c.payload.MaxBatchSize,
		QueueClass:            c.payload.QueueClass,
		FlowExecutionFilter:   mustConnectorInboundRouteFilterJSON(c.payload.FlowExecutionFilter),
		IsDeleted:             0,
	}, nil
}

// LiveProjection reads the latest (highest Version) non-deleted row for this
// component's (packName, eventKind) from _System_ConnectorInboundRoute.
func (c *connectorInboundRouteComponent) LiveProjection(ctx context.Context, deps VerifyDeps) (any, error) {
	if deps.DB == nil {
		return nil, fmt.Errorf("database is required")
	}
	var row connectorInboundRouteRow
	err := deps.DB.QueryRowContext(ctx, `
SELECT Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName,
       COALESCE(InvocationMode, ''), COALESCE(BatchInputName, ''), COALESCE(BatchResultOutputName, ''), COALESCE(MaxBatchSize, 0),
       COALESCE(QueueClass, ''), COALESCE(FlowExecutionFilter, ''),
       COALESCE(IsCurrent, 0), COALESCE(IsDeleted, 0)
  FROM _System_ConnectorInboundRoute
 WHERE Namespace = ?
   AND EventKind = ?
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY Version DESC
 LIMIT 1`,
		c.packName, c.payload.EventKind,
	).Scan(&row.ID, &row.Namespace, &row.InstalledPackId, &row.EventKind, &row.Version, &row.FlowDeveloperName, &row.InvocationMode, &row.BatchInputName, &row.BatchResultOutputName, &row.MaxBatchSize, &row.QueueClass, &row.FlowExecutionFilter, &row.IsCurrent, &row.IsDeleted)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("query connector inbound route for pack %s / eventKind %s: %w", c.packName, c.payload.EventKind, err)
	}
	return row, nil
}

// VerifyComponent implements componentCustomVerifier. It finds the latest
// (highest Version) non-deleted row for (packInstallID, eventKind) and compares
// (Namespace, EventKind, FlowDeveloperName, IsDeleted).  Id is verified to equal
// connectorInboundRouteID(packInstallID, eventKind, latestVersion).
//
// ExpectedProjection omits Id, InstalledPackId, and Version because they are not
// known at design time; this method handles the fields that require DB resolution
// (per Agent A Gap 4 — mirrors connector_capability_component.go:VerifyComponent).
func (c *connectorInboundRouteComponent) VerifyComponent(ctx context.Context, deps VerifyDeps) error {
	if deps.DB == nil {
		return fmt.Errorf("database is required")
	}
	// Resolve the current PackInstallId for this pack.
	packInstallID, err := resolveConnectorInboundRoutePackInstallID(ctx, deps.DB, c.packName)
	if err != nil {
		return fmt.Errorf("VerifyComponent: %w", err)
	}

	var row connectorInboundRouteRow
	err = deps.DB.QueryRowContext(ctx, `
SELECT Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName,
       COALESCE(InvocationMode, ''), COALESCE(BatchInputName, ''), COALESCE(BatchResultOutputName, ''), COALESCE(MaxBatchSize, 0),
       COALESCE(QueueClass, ''), COALESCE(FlowExecutionFilter, ''),
       COALESCE(IsCurrent, 0), COALESCE(IsDeleted, 0)
  FROM _System_ConnectorInboundRoute
 WHERE Namespace = ?
   AND EventKind = ?
   AND InstalledPackId = ?
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY Version DESC
 LIMIT 1`,
		c.packName, c.payload.EventKind, packInstallID,
	).Scan(&row.ID, &row.Namespace, &row.InstalledPackId, &row.EventKind, &row.Version, &row.FlowDeveloperName, &row.InvocationMode, &row.BatchInputName, &row.BatchResultOutputName, &row.MaxBatchSize, &row.QueueClass, &row.FlowExecutionFilter, &row.IsCurrent, &row.IsDeleted)
	if err == sql.ErrNoRows {
		return fmt.Errorf("connector inbound route not found: pack=%s eventKind=%s packInstallId=%s", c.packName, c.payload.EventKind, packInstallID)
	}
	if err != nil {
		return fmt.Errorf("verify connector inbound route pack=%s eventKind=%s: %w", c.packName, c.payload.EventKind, err)
	}

	// Verify scalar fields.
	if row.Namespace != c.packName {
		return fmt.Errorf("connector inbound route Namespace mismatch: got %q want %q", row.Namespace, c.packName)
	}
	if row.EventKind != c.payload.EventKind {
		return fmt.Errorf("connector inbound route EventKind mismatch: got %q want %q", row.EventKind, c.payload.EventKind)
	}
	if row.FlowDeveloperName != c.payload.FlowDeveloperName {
		return fmt.Errorf("connector inbound route FlowDeveloperName mismatch for eventKind %s: got %q want %q", c.payload.EventKind, row.FlowDeveloperName, c.payload.FlowDeveloperName)
	}
	if row.InvocationMode != c.payload.InvocationMode {
		return fmt.Errorf("connector inbound route InvocationMode mismatch for eventKind %s: got %q want %q", c.payload.EventKind, row.InvocationMode, c.payload.InvocationMode)
	}
	if row.BatchInputName != c.payload.BatchInputName {
		return fmt.Errorf("connector inbound route BatchInputName mismatch for eventKind %s: got %q want %q", c.payload.EventKind, row.BatchInputName, c.payload.BatchInputName)
	}
	if row.BatchResultOutputName != c.payload.BatchResultOutputName {
		return fmt.Errorf("connector inbound route BatchResultOutputName mismatch for eventKind %s: got %q want %q", c.payload.EventKind, row.BatchResultOutputName, c.payload.BatchResultOutputName)
	}
	if row.MaxBatchSize != c.payload.MaxBatchSize {
		return fmt.Errorf("connector inbound route MaxBatchSize mismatch for eventKind %s: got %d want %d", c.payload.EventKind, row.MaxBatchSize, c.payload.MaxBatchSize)
	}
	if row.QueueClass != c.payload.QueueClass {
		return fmt.Errorf("connector inbound route QueueClass mismatch for eventKind %s: got %q want %q", c.payload.EventKind, row.QueueClass, c.payload.QueueClass)
	}
	liveFilter, err := connector_inbound_routing.ParsePayloadValueFilterJSON(row.FlowExecutionFilter)
	if err != nil {
		return fmt.Errorf("connector inbound route FlowExecutionFilter invalid for eventKind %s: %w", c.payload.EventKind, err)
	}
	liveFilterJSON, err := connector_inbound_routing.PayloadValueFilterJSON(liveFilter)
	if err != nil {
		return fmt.Errorf("connector inbound route FlowExecutionFilter canonicalization failed for eventKind %s: %w", c.payload.EventKind, err)
	}
	expectedFilterJSON := mustConnectorInboundRouteFilterJSON(c.payload.FlowExecutionFilter)
	if liveFilterJSON != expectedFilterJSON {
		return fmt.Errorf("connector inbound route FlowExecutionFilter mismatch for eventKind %s", c.payload.EventKind)
	}
	if row.IsDeleted != 0 {
		return fmt.Errorf("connector inbound route row for eventKind %s is soft-deleted (IsDeleted=%d)", c.payload.EventKind, row.IsDeleted)
	}
	// B22: the currently-declared route row must have IsCurrent=1. A row with
	// IsCurrent=0 means the sweeper preserved it for backlog drain but it is no
	// longer the live-declared route — verify must report mismatch so the installer
	// knows to re-apply.
	if row.IsCurrent != 1 {
		return fmt.Errorf("connector inbound route row for eventKind %s has IsCurrent=%d (want 1); row preserved for backlog drain but not the declared live route", c.payload.EventKind, row.IsCurrent)
	}
	if row.Version < 1 {
		return fmt.Errorf("connector inbound route row for eventKind %s has invalid Version=%d (must be >= 1)", c.payload.EventKind, row.Version)
	}
	// Verify deterministic Id for the latest version.
	expectedID := connectorInboundRouteID(packInstallID, c.payload.EventKind, row.Version)
	if row.ID != expectedID {
		return fmt.Errorf("connector inbound route Id mismatch for eventKind %s version %d: got %q want %q", c.payload.EventKind, row.Version, row.ID, expectedID)
	}
	return nil
}

// Apply ensures ONE live _System_ConnectorInboundRoute row exists for this
// component's declared (EventKind → FlowDeveloperName) mapping, using
// versioned append-only semantics:
//
//   - If no live row exists yet: INSERT a new row with Version=1.
//   - If a live row exists with the same FlowDeveloperName: no-op (idempotent
//     re-apply — same version, same flow, nothing to update).
//   - If a live row exists with a DIFFERENT FlowDeveloperName: INSERT a new row
//     with Version=max+1, leaving the old version row intact for any queue rows
//     that already snapshotted it.
//
// All three cases run inside a transaction to make "read max version, decide
// action, insert" atomic.  Old version rows are preserved for backlog events;
// version-level GC is handled by the incarnation and pack sweepers, which
// preserve route rows referenced by non-terminal queue events and reclaim them
// once the queue drains.
//
// # Legacy v1 ID rewrite (B7 — codex R8 BLOCK)
//
// Pre-#1618 _System_ConnectorInboundRoute rows have Id=legacyConnectorInboundRouteID
// (hash of packInstallId+eventKind, no version component). The migration
// backfilled Version=1 on those rows but did NOT rewrite Id. VerifyComponent
// now expects Id=connectorInboundRouteID(packInstallId,eventKind,version),
// which includes the version component. Without intervention, setup verify
// fails after rollout for orgs with pre-existing routes, and the idempotent
// no-op path cannot self-heal because the live row is not found by the new Id.
//
// Apply detects legacy v1 rows (Version=1 at the legacy Id shape) BEFORE the
// idempotent early-return and rewrites Id in-place. The rewrite is a single
// UPDATE and runs at most once per row (subsequent Apply calls find the row at
// the new Id).
//
// Namespace is written from c.packName (pack identity), NOT from payload.Namespace —
// payload.Namespace is already validated to equal packName by the validator.
func (c *connectorInboundRouteComponent) Apply(ctx context.Context, deps ApplyDeps) error {
	if deps.DB == nil {
		return fmt.Errorf("database is required")
	}
	// TODO(security): install-time same-pack route→flow validation (defense in depth).
	// The runtime already enforces that a connector-inbound execution may only run
	// a protected standard flow whose DefinitionOwnerPack matches the routing
	// connector's namespace (ConnectorInboundExecutionAuthorizesFlow). A complementary
	// install-time gate would reject the route if FlowDeveloperName resolves to a
	// FlowDefinition whose OwnerPack != c.packName, failing closed at pack install
	// rather than at runtime. Deferred: requires a cross-component DB query against
	// FlowDefinition during Apply, which needs careful scoping (FlowDefinition may
	// not be installed yet in a partial pack apply; ordering is needed). Track as a
	// successor issue once the runtime guard is confirmed to be effective.
	packInstallID, err := resolveConnectorInboundRoutePackInstallID(ctx, deps.DB, c.packName)
	if err != nil {
		return err
	}

	tx, err := deps.DB.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx for connector inbound route %s/%s: %w", c.packName, c.payload.EventKind, err)
	}
	defer func() {
		if err != nil {
			_ = tx.Rollback()
		}
	}()

	// B7: Detect and rewrite legacy v1 rows that still have the old 2-arg Id shape.
	// Pre-#1618 rows: Id = legacyConnectorInboundRouteID(packInstallId, eventKind).
	// Migration backfilled Version=1 but did not rewrite Id.
	// New shape: Id = connectorInboundRouteID(packInstallId, eventKind, 1).
	legacyID := legacyConnectorInboundRouteID(packInstallID, c.payload.EventKind)
	newV1ID := connectorInboundRouteID(packInstallID, c.payload.EventKind, 1)
	if legacyID != newV1ID {
		var existsLegacy bool
		err = tx.QueryRowContext(ctx, `
SELECT EXISTS(
  SELECT 1 FROM _System_ConnectorInboundRoute
   WHERE Id = ? AND Version = 1 AND COALESCE(IsDeleted, 0) = 0
)`, legacyID).Scan(&existsLegacy)
		if err != nil {
			err = fmt.Errorf("detect legacy v1 row for connector inbound route %s/%s: %w", c.packName, c.payload.EventKind, err)
			return err
		}
		if existsLegacy {
			_, err = tx.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET Id = ?, LastModifiedDate = NOW()
 WHERE Id = ? AND Version = 1`, newV1ID, legacyID)
			if err != nil {
				err = fmt.Errorf("rewrite legacy v1 Id for connector inbound route %s/%s: %w", c.packName, c.payload.EventKind, err)
				return err
			}
		}
	}

	// Find the current live row (highest non-deleted Version) for this
	// (InstalledPackId, EventKind), including its IsCurrent flag.
	//
	// B22: we also select IsCurrent so we can distinguish:
	//   - IsCurrent=1 + same flow  → true idempotent no-op
	//   - IsCurrent=0 + same flow  → sweeper preserved this row; reactivate (UPDATE IsCurrent=1)
	//   - any IsCurrent  + diff flow → INSERT new version
	var currentVersion int64
	var currentSpec connector_inbound_routing.RouteExecutionSpec
	var currentIsCurrent int
	var currentFilterJSON string
	scanErr := tx.QueryRowContext(ctx, `
SELECT Version, FlowDeveloperName,
       COALESCE(InvocationMode, ''), COALESCE(BatchInputName, ''), COALESCE(BatchResultOutputName, ''), COALESCE(MaxBatchSize, 0),
       COALESCE(QueueClass, ''), COALESCE(FlowExecutionFilter, ''),
       COALESCE(IsCurrent, 0)
  FROM _System_ConnectorInboundRoute
 WHERE InstalledPackId = ?
   AND EventKind = ?
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY Version DESC
 LIMIT 1`,
		packInstallID, c.payload.EventKind,
	).Scan(&currentVersion, &currentSpec.FlowDeveloperName, &currentSpec.InvocationMode, &currentSpec.BatchInputName, &currentSpec.BatchResultOutputName, &currentSpec.MaxBatchSize, &currentSpec.QueueClass, &currentFilterJSON, &currentIsCurrent)
	liveRowExists := true
	if scanErr == sql.ErrNoRows {
		liveRowExists = false
	} else if scanErr != nil {
		err = fmt.Errorf("query current version for connector inbound route %s/%s: %w", c.packName, c.payload.EventKind, scanErr)
		return err
	}
	if liveRowExists {
		currentSpec.FlowExecutionFilter, err = connector_inbound_routing.ParsePayloadValueFilterJSON(currentFilterJSON)
		if err != nil {
			err = fmt.Errorf("parse current connector inbound route flowExecutionFilter %s/%s: %w", c.packName, c.payload.EventKind, err)
			return err
		}
		currentSpec, err = connector_inbound_routing.NormalizeRouteExecutionSpec(currentSpec)
		if err != nil {
			err = fmt.Errorf("normalize current connector inbound route execution spec %s/%s: %w", c.packName, c.payload.EventKind, err)
			return err
		}
	}

	desiredSpec, err := connectorInboundRoutePayloadSpec(c.payload)
	if err != nil {
		err = fmt.Errorf("normalize desired connector inbound route execution spec %s/%s: %w", c.packName, c.payload.EventKind, err)
		return err
	}
	if liveRowExists && connector_inbound_routing.RouteExecutionSpecEqual(currentSpec, desiredSpec) {
		if currentIsCurrent == 1 {
			// True idempotent re-apply: row already live with matching flow.
			return tx.Commit()
		}
		// B22: same flow but IsCurrent=0 — sweeper preserved this row for backlog drain.
		// Reactivate it by setting IsCurrent=1 instead of inserting a new version.
		_, err = tx.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsCurrent = 1, LastModifiedDate = NOW()
 WHERE InstalledPackId = ?
   AND EventKind = ?
   AND Version = ?
   AND COALESCE(IsDeleted, 0) = 0`,
			packInstallID, c.payload.EventKind, currentVersion)
		if err != nil {
			err = fmt.Errorf("reactivate preserved connector inbound route %s/%s version=%d: %w", c.packName, c.payload.EventKind, currentVersion, err)
			return err
		}
		// Clear IsCurrent on all other non-deleted rows for the same (InstalledPackId, EventKind)
		// so only the reactivated version carries IsCurrent=1 after this reactivation.
		// The Version != ? clause ensures we skip the row we just reactivated.
		_, err = tx.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsCurrent = 0, LastModifiedDate = NOW()
 WHERE InstalledPackId = ?
   AND EventKind = ?
   AND Version != ?
   AND COALESCE(IsDeleted, 0) = 0
   AND COALESCE(IsCurrent, 0) = 1`,
			packInstallID, c.payload.EventKind, currentVersion)
		if err != nil {
			err = fmt.Errorf("clear IsCurrent on other versions after reactivation of connector inbound route %s/%s version=%d: %w", c.packName, c.payload.EventKind, currentVersion, err)
			return err
		}
		return tx.Commit()
	}

	// Flow changed (or no live row exists): INSERT a new version row.
	//
	// B21: derive nextVersion from MAX(Version) across ALL rows — including soft-deleted
	// ones — so a drop-then-readd cycle (sweeper soft-deletes V=N, manifest re-adds the
	// same EventKind) never reuses an existing Version number and hits the unique constraint.
	var maxVersionAllRows int64
	scanMaxErr := tx.QueryRowContext(ctx, `
SELECT COALESCE(MAX(Version), 0)
  FROM _System_ConnectorInboundRoute
 WHERE InstalledPackId = ?
   AND EventKind = ?`,
		packInstallID, c.payload.EventKind,
	).Scan(&maxVersionAllRows)
	if scanMaxErr != nil {
		err = fmt.Errorf("query max version (all rows) for connector inbound route %s/%s: %w", c.packName, c.payload.EventKind, scanMaxErr)
		return err
	}
	nextVersion := maxVersionAllRows + 1

	id := connectorInboundRouteID(packInstallID, c.payload.EventKind, nextVersion)
	// IsCurrent=1: this row is declared in the current manifest.
	// Apply is the only writer that mints IsCurrent=1; the sweeper sets it to 0
	// when it preserves a dropped-eventKind row for backlog drain (B19 fix).
	_, err = tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorInboundRoute
  (Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName, InvocationMode, BatchInputName, BatchResultOutputName, MaxBatchSize, QueueClass, FlowExecutionFilter, IsCurrent, IsDeleted, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, 0, NOW(), NOW())`,
		id,
		c.packName, // Namespace = c.packName (pack identity)
		packInstallID,
		c.payload.EventKind,
		nextVersion,
		c.payload.FlowDeveloperName,
		c.payload.InvocationMode,
		c.payload.BatchInputName,
		c.payload.BatchResultOutputName,
		c.payload.MaxBatchSize,
		c.payload.QueueClass,
		nullableConnectorInboundRouteFilter(c.payload.FlowExecutionFilter),
	)
	if err != nil {
		err = fmt.Errorf("insert versioned connector inbound route %s/%s version=%d for pack %s: %w", c.packName, c.payload.EventKind, nextVersion, c.packName, err)
		return err
	}
	// Clear IsCurrent on all other non-deleted rows for the same (InstalledPackId, EventKind)
	// so only the newly inserted version carries IsCurrent=1. This is safe for the
	// first-install case (no rows to match) and mandatory for flow-change versioning —
	// without it, the superseded row keeps IsCurrent=1 and the replay guard becomes
	// ineffective (ReplayDeadEvent's IsCurrent=1 JOIN would still pass for the stale version).
	_, err = tx.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRoute
   SET IsCurrent = 0, LastModifiedDate = NOW()
 WHERE InstalledPackId = ?
   AND EventKind = ?
   AND Version != ?
   AND COALESCE(IsDeleted, 0) = 0
   AND COALESCE(IsCurrent, 0) = 1`,
		packInstallID,
		c.payload.EventKind,
		nextVersion,
	)
	if err != nil {
		err = fmt.Errorf("clear IsCurrent on superseded connector inbound routes %s/%s version<%d: %w", c.packName, c.payload.EventKind, nextVersion, err)
		return err
	}
	return tx.Commit()
}

// legacyConnectorInboundRouteID computes the pre-#1618 row Id shape:
// hash(packInstallId, eventKind) — no version component. Used by Apply to
// detect legacy v1 rows that need Id rewrite (B7 — codex R8 BLOCK).
func legacyConnectorInboundRouteID(packInstallID, eventKind string) string {
	sum := sha256.Sum256([]byte(
		strings.ToLower(strings.TrimSpace(packInstallID)) + "\x00" +
			strings.ToLower(strings.TrimSpace(eventKind)),
	))
	return "cir_" + hex.EncodeToString(sum[:])
}

// resolveConnectorInboundRoutePackInstallID looks up the PackInstallId for the
// in-flight installation of packName. The pack's row will have Status='installing'
// (set by markInstalledPackInstalling in source_pack_ops.go before Apply is
// called) or Status='active' (idempotent re-apply after successful install).
//
// Returns a non-empty PackInstallId string on success. Returns an error if no
// matching row exists or if the found PackInstallId is blank (backfill not applied).
func resolveConnectorInboundRoutePackInstallID(ctx context.Context, db *sql.DB, packName string) (string, error) {
	if packInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		return packInstallID, nil
	}
	var packInstallID string
	err := db.QueryRowContext(ctx, `
    SELECT PackInstallId
      FROM _System_InstalledPack
     WHERE Name = ?
       AND COALESCE(IsDeleted, 0) = 0
       AND Status IN ('installing', 'active')
     ORDER BY LastModifiedDate DESC
     LIMIT 1`, packName).Scan(&packInstallID)
	if err == sql.ErrNoRows {
		return "", fmt.Errorf("connector_inbound_route: pack %q has no in-flight installation record (Apply called outside install lifecycle?)", packName)
	}
	if err != nil {
		return "", fmt.Errorf("connector_inbound_route: lookup PackInstallId for pack %q: %w", packName, err)
	}
	if strings.TrimSpace(packInstallID) == "" {
		return "", fmt.Errorf("connector_inbound_route: pack %q has empty PackInstallId (upgrade backfill not applied)", packName)
	}
	return packInstallID, nil
}

// connectorInboundRouteID produces a deterministic row Id from the
// (packInstallId, eventKind, version) triple. All string inputs are
// canonicalised (lowercase+trim) so the Id form matches the stored column form,
// preventing Id-vs-column case-skew (Challenger MED #5).
//
// Version is included in the hash so each version row has a globally unique Id,
// enabling append-only versioned inserts without Id collisions.
//
// Prefix "cir_" avoids Id collisions with other system table Id spaces.
func connectorInboundRouteID(packInstallID, eventKind string, version int64) string {
	sum := sha256.Sum256([]byte(
		strings.ToLower(strings.TrimSpace(packInstallID)) + "\x00" +
			strings.ToLower(strings.TrimSpace(eventKind)) + "\x00" +
			fmt.Sprintf("%d", version),
	))
	return "cir_" + hex.EncodeToString(sum[:])
}

func mustConnectorInboundRouteFilterJSON(filter *connector_inbound_routing.PayloadValueFilter) string {
	raw, err := connector_inbound_routing.PayloadValueFilterJSON(filter)
	if err != nil {
		panic(err)
	}
	return raw
}

func nullableConnectorInboundRouteFilter(filter *connector_inbound_routing.PayloadValueFilter) any {
	raw := mustConnectorInboundRouteFilterJSON(filter)
	if raw == "" {
		return nil
	}
	return raw
}