package packs

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

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate"
)

const (
	installedPackStatusActive     = "active"
	installedPackStatusRemoving   = "removing"
	installedPackStatusInstalling = "installing"
)

type installedPackStateRow struct {
	status    string
	isDeleted int
}

type installedPackDeclarationRow struct {
	name                  string
	category              string
	status                string
	isDeleted             int
	manifestFormatVersion sql.NullString
	sourceKind            sql.NullString
	sourceURI             sql.NullString
	sourceRevision        sql.NullString
	applied               sql.NullString
	desired               sql.NullString
}

// ConnectorPackRow is the connector-pack-shaped projection of an installed-pack row.
type ConnectorPackRow = packstate.ConnectorPackRow

// ConnectorInboundWorkerPackRow is the startup projection used by the inbound
// worker supervisor. DrainOnly=true means the pack is not active but still has
// non-terminal queue rows that must be finalized before uninstall completes.
type ConnectorInboundWorkerPackRow = packstate.ConnectorInboundWorkerPackRow

// ResolveConnectorPackByNamespace returns the active connector-pack row for the
// given namespace (which is the installed-pack Name for connector packs).
// Returns (nil, false, nil) when the row is missing, soft-deleted, inactive, or
// the Name resolves to a non-connector pack — never silently falls back to a
// non-connector pack with the same Name. Downstream substrate (NCRM-INT-03,
// -04A, -04, -05a) consumes this resolver instead of querying _System_InstalledPack directly.
func ResolveConnectorPackByNamespace(ctx context.Context, db *sql.DB, namespace string) (*ConnectorPackRow, bool, error) {
	return packstate.ResolveConnectorPackByNamespace(ctx, db, namespace)
}

// ListConnectorPacksForInboundWorkers returns the connector-pack incarnations
// that must have inbound workers started at process boot:
//   - active steady-state connector packs that declare the full inbound
//     webhook capability pair for their own namespace, and
//   - drain-only non-active connector packs that still have non-terminal
//     _System_ConnectorEvent rows (pending/failed/processing).
//
// The active-row filter avoids aborting backend startup for valid outbound-only
// connector packs that do not serve inbound webhook traffic. Drain-only rows
// are excluded once the namespace has an active rebirth because pack-owned
// NamedCredentials are namespace-scoped live rows. The same fail-closed rule
// excludes multiple surviving non-active incarnations for one namespace; until
// credentials are install-scoped snapshots, the worker can safely drain only a
// single historical incarnation per namespace.
//
// DrainOnly rows remain eligible only for queue finalization; dispatcher-side
// enqueue stays gated on active-pack resolution.
func ListConnectorPacksForInboundWorkers(ctx context.Context, db *sql.DB) ([]ConnectorInboundWorkerPackRow, error) {
	return packstate.ListConnectorPacksForInboundWorkers(ctx, db)
}

// IsConnectorPackMidApply reports whether a connector pack with the given namespace
// exists but is NOT in a fully-applied steady state. The inbound webhook dispatcher
// uses this to classify retriable-5xx (503) scenarios:
//
//   - pack row is present but Status != "active"
//   - pack row is present but PackInstallId is blank (upgrade backfill not yet applied)
//   - pack row is present but AppliedPayloadHash or DesiredPayloadHash is blank
//   - pack row is present but AppliedPayloadHash != DesiredPayloadHash (hash drift
//     during setup-apply / reapply window)
//
// This is a strict superset of the former "installing/removing status" check: it
// additionally catches the blank-install-id and hash-drift mid-apply states where
// IsCapabilityDeclaredByConnectorPack (whose SQL WHERE requires PackInstallId <> ”
// AND AppliedPayloadHash = DesiredPayloadHash) returns false even though the pack
// will be live once apply converges.
//
// Returns (false, nil) when the row is genuinely absent (ErrNoRows) — the pack has
// never been installed, so the caller should return a terminal 404.
// Returns (false, error) only for DB errors.
func IsConnectorPackMidApply(ctx context.Context, db *sql.DB, namespace string) (bool, error) {
	return packstate.IsConnectorPackMidApply(ctx, db, namespace)
}

func IsPackActive(ctx context.Context, db *sql.DB, packName string) (bool, error) {
	row, ok, err := loadInstalledPackState(ctx, db, packName)
	if err != nil {
		return false, err
	}
	return ok && row.isDeleted == 0 && row.status == installedPackStatusActive, nil
}

func IsPackInRemovalWindow(ctx context.Context, db *sql.DB, packName string) (bool, error) {
	row, ok, err := loadInstalledPackState(ctx, db, packName)
	if err != nil {
		return false, err
	}
	return ok && row.isDeleted == 0 && row.status == installedPackStatusRemoving, nil
}

func IsFieldDeclaredByPack(ctx context.Context, db *sql.DB, packName, objectAPIName, fieldAPIName string) (bool, error) {
	return isFieldDeclaredByPack(ctx, db, packName, objectAPIName, fieldAPIName, nil)
}

// IsFieldDeclaredByPackWithComponentKindRegistrationSet carries one immutable
// distribution registration snapshot through the installed peer source read.
func IsFieldDeclaredByPackWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, packName, objectAPIName, fieldAPIName string, registrations ComponentKindRegistrationSet) (bool, error) {
	return isFieldDeclaredByPack(ctx, db, packName, objectAPIName, fieldAPIName, &registrations)
}

func isFieldDeclaredByPack(ctx context.Context, db *sql.DB, packName, objectAPIName, fieldAPIName string, registrations *ComponentKindRegistrationSet) (bool, error) {
	row, ok, err := loadInstalledPackDeclarationRow(ctx, db, packName)
	if err != nil {
		return false, err
	}
	if !ok || row.isDeleted != 0 || row.status != installedPackStatusActive {
		return false, nil
	}
	sourceKind := strings.TrimSpace(row.sourceKind.String)
	sourceURI := strings.TrimSpace(row.sourceURI.String)
	if sourceKind == "" || sourceURI == "" {
		return false, fmt.Errorf("installed pack %s has no source identity", strings.TrimSpace(row.name))
	}
	pack, err := resolveActiveInstalledSourceBackedPackWithStoredManifestVersionAndComponentKindRegistrationSet(
		ctx,
		db,
		strings.TrimSpace(row.name),
		Category(strings.TrimSpace(row.category)),
		sourceKind,
		sourceURI,
		row.manifestFormatVersion.String,
		registrations,
	)
	if err != nil {
		return false, fmt.Errorf("resolve installed pack %s declaration source: %w", strings.TrimSpace(row.name), err)
	}
	currentHash, err := PackPayloadHash(pack)
	if err != nil {
		return false, fmt.Errorf("compute installed pack %s payload hash: %w", strings.TrimSpace(row.name), err)
	}
	if !row.applied.Valid || strings.TrimSpace(row.applied.String) == "" {
		return false, fmt.Errorf("installed pack %s AppliedPayloadHash is missing; refusing declaration check", strings.TrimSpace(row.name))
	}
	if !row.desired.Valid || strings.TrimSpace(row.desired.String) == "" {
		return false, fmt.Errorf("installed pack %s DesiredPayloadHash is missing; refusing declaration check", strings.TrimSpace(row.name))
	}
	if strings.TrimSpace(row.applied.String) != strings.TrimSpace(row.desired.String) {
		return false, fmt.Errorf("installed pack %s AppliedPayloadHash = %q, DesiredPayloadHash = %q; refusing declaration check", strings.TrimSpace(row.name), strings.TrimSpace(row.applied.String), strings.TrimSpace(row.desired.String))
	}
	if strings.TrimSpace(row.applied.String) != currentHash {
		return false, fmt.Errorf("installed pack %s AppliedPayloadHash = %q, want %q", strings.TrimSpace(row.name), strings.TrimSpace(row.applied.String), currentHash)
	}
	return currentSourceDeclaresField(pack, row.name, objectAPIName, fieldAPIName)
}

func IsFieldDeclaredByCurrentSourceForExplicitRemoval(ctx context.Context, db *sql.DB, packName, objectAPIName, fieldAPIName string) (bool, error) {
	return isFieldDeclaredByCurrentSourceForExplicitRemoval(ctx, db, packName, objectAPIName, fieldAPIName, nil)
}

// IsFieldDeclaredByCurrentSourceForExplicitRemovalWithComponentKindRegistrationSet
// carries one distribution-owned component registration snapshot through the
// active installed-source read used by explicit field removal.
func IsFieldDeclaredByCurrentSourceForExplicitRemovalWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, packName, objectAPIName, fieldAPIName string, registrations ComponentKindRegistrationSet) (bool, error) {
	return isFieldDeclaredByCurrentSourceForExplicitRemoval(ctx, db, packName, objectAPIName, fieldAPIName, &registrations)
}

func isFieldDeclaredByCurrentSourceForExplicitRemoval(ctx context.Context, db *sql.DB, packName, objectAPIName, fieldAPIName string, registrations *ComponentKindRegistrationSet) (bool, error) {
	row, ok, err := loadInstalledPackDeclarationRow(ctx, db, packName)
	if err != nil {
		return false, err
	}
	if !ok || row.isDeleted != 0 {
		return false, fmt.Errorf("installed pack %s is missing", strings.TrimSpace(packName))
	}
	if row.status != installedPackStatusActive {
		return false, fmt.Errorf("installed pack %s status is %q, want active", strings.TrimSpace(row.name), row.status)
	}
	if !row.applied.Valid || strings.TrimSpace(row.applied.String) == "" {
		return false, fmt.Errorf("installed pack %s AppliedPayloadHash is missing; refusing explicit removal check", strings.TrimSpace(row.name))
	}
	if !row.desired.Valid || strings.TrimSpace(row.desired.String) == "" {
		return false, fmt.Errorf("installed pack %s DesiredPayloadHash is missing; refusing explicit removal check", strings.TrimSpace(row.name))
	}
	if strings.TrimSpace(row.applied.String) != strings.TrimSpace(row.desired.String) {
		return false, fmt.Errorf("installed pack %s AppliedPayloadHash = %q, DesiredPayloadHash = %q; refusing explicit removal check", strings.TrimSpace(row.name), strings.TrimSpace(row.applied.String), strings.TrimSpace(row.desired.String))
	}
	if !IsPackPayloadHashV1(strings.TrimSpace(row.applied.String)) {
		return false, fmt.Errorf("installed pack %s AppliedPayloadHash = %q; refusing explicit removal check", strings.TrimSpace(row.name), strings.TrimSpace(row.applied.String))
	}
	pack, err := resolveActiveInstalledPackDeclarationSourceWithComponentKindRegistrationSet(ctx, db, row, registrations)
	if err != nil {
		return false, err
	}
	return currentSourceDeclaresField(pack, row.name, objectAPIName, fieldAPIName)
}

func resolveInstalledPackDeclarationSourceWithOptions(row installedPackDeclarationRow, opts ...SourceBackedPackOption) (*SourceBackedPack, error) {
	sourceKind := strings.TrimSpace(row.sourceKind.String)
	sourceURI := strings.TrimSpace(row.sourceURI.String)
	sourceRevision := strings.TrimSpace(row.sourceRevision.String)
	if sourceKind == "" || sourceURI == "" {
		return nil, fmt.Errorf("installed pack %s has no source identity", strings.TrimSpace(row.name))
	}
	pack, err := resolveInstalledSourceBackedPackWithStoredIdentity(
		strings.TrimSpace(row.name),
		Category(strings.TrimSpace(row.category)),
		sourceKind,
		sourceURI,
		row.manifestFormatVersion.String,
		sourceRevision,
		opts...,
	)
	if err != nil {
		return nil, fmt.Errorf("resolve installed pack %s declaration source: %w", strings.TrimSpace(row.name), err)
	}
	return pack, nil
}

func resolveActiveInstalledPackDeclarationSource(ctx context.Context, db *sql.DB, row installedPackDeclarationRow) (*SourceBackedPack, error) {
	return resolveActiveInstalledPackDeclarationSourceWithComponentKindRegistrationSet(ctx, db, row, nil)
}

func resolveActiveInstalledPackDeclarationSourceWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, row installedPackDeclarationRow, registrations *ComponentKindRegistrationSet) (*SourceBackedPack, error) {
	sourceKind := strings.TrimSpace(row.sourceKind.String)
	sourceURI := strings.TrimSpace(row.sourceURI.String)
	if sourceKind == "" || sourceURI == "" {
		return nil, fmt.Errorf("installed pack %s has no source identity", strings.TrimSpace(row.name))
	}
	pack, err := resolveActiveInstalledSourceBackedPackWithStoredManifestVersionAndComponentKindRegistrationSet(
		ctx,
		db,
		strings.TrimSpace(row.name),
		Category(strings.TrimSpace(row.category)),
		sourceKind,
		sourceURI,
		row.manifestFormatVersion.String,
		registrations,
	)
	if err != nil {
		return nil, fmt.Errorf("resolve active installed pack %s declaration source: %w", strings.TrimSpace(row.name), err)
	}
	return pack, nil
}

// ResolveInstalledPackDeclarationSourceByInstallID resolves the source-backed
// pack for the exact installed-pack incarnation identified by
// (packName, packInstallID), returning the row status alongside the pack.
//
// It returns ok=false when the installed row is absent, soft-deleted, missing
// its applied/desired payload hashes, or when the resolved declaration source no
// longer matches the stored applied payload hash. Active declaration-source
// callers validate against the current source tree at SourceURI. Non-active
// rows with revision-addressable source kinds can reconstruct the stored
// SourceRevision, while local_path rows still read SourceURI because their
// revision is install provenance rather than a stable artifact address.
func ResolveInstalledPackDeclarationSourceByInstallID(ctx context.Context, db *sql.DB, packName, packInstallID string) (*SourceBackedPack, string, bool, error) {
	row, ok, err := loadInstalledPackDeclarationRowByInstallID(ctx, db, packName, packInstallID)
	if err != nil {
		return nil, "", false, err
	}
	if !ok || row.isDeleted != 0 {
		return nil, "", false, nil
	}
	if !row.applied.Valid || strings.TrimSpace(row.applied.String) == "" {
		return nil, row.status, false, nil
	}
	if !row.desired.Valid || strings.TrimSpace(row.desired.String) == "" {
		return nil, row.status, false, nil
	}
	if strings.TrimSpace(row.applied.String) != strings.TrimSpace(row.desired.String) {
		return nil, row.status, false, nil
	}
	var pack *SourceBackedPack
	if row.status == installedPackStatusActive {
		pack, err = resolveActiveInstalledPackDeclarationSource(ctx, db, row)
	} else {
		pack, err = resolveInstalledPackDeclarationSourceWithOptions(row, withLegacyConnectorCalloutBatchIDResponseFieldCompat())
	}
	if err != nil {
		return nil, row.status, false, nil
	}
	if _, err := ValidateStoredPackPayloadHash(pack, row.applied.String); err != nil {
		return nil, row.status, false, nil
	}
	return pack, row.status, true, nil
}

func currentSourceDeclaresField(pack *SourceBackedPack, packName string, objectAPIName, fieldAPIName string) (bool, error) {
	specs, err := ParseObjectSpecsPayloads(pack.RuntimeObjectSpecPayloads(), "runtime-object-spec")
	if err != nil {
		return false, fmt.Errorf("parse installed pack %s object specs: %w", strings.TrimSpace(packName), err)
	}
	wantObject := strings.ToLower(strings.TrimSpace(objectAPIName))
	wantField := strings.ToLower(strings.TrimSpace(fieldAPIName))
	if wantObject == "" || wantField == "" {
		return false, nil
	}
	reservedFieldNames := l0ReservedFieldNameSet()
	for _, spec := range specs {
		if strings.ToLower(strings.TrimSpace(spec.APIName)) != wantObject {
			continue
		}
		if reservedFieldNames[wantField] {
			return true, nil
		}
		for _, field := range spec.Fields {
			if strings.ToLower(strings.TrimSpace(field.APIName)) == wantField {
				return true, nil
			}
		}
		return false, nil
	}
	return false, nil
}

func loadInstalledPackState(ctx context.Context, db *sql.DB, packName string) (installedPackStateRow, bool, error) {
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return installedPackStateRow{}, false, nil
	}
	var row installedPackStateRow
	err := db.QueryRowContext(ctx, `
SELECT Status, COALESCE(IsDeleted, 0)
FROM _System_InstalledPack
WHERE Name = ?
LIMIT 1`, packName).Scan(&row.status, &row.isDeleted)
	if err == sql.ErrNoRows {
		return installedPackStateRow{}, false, nil
	}
	if err != nil {
		return installedPackStateRow{}, false, fmt.Errorf("query installed pack %s state: %w", packName, err)
	}
	row.status = strings.TrimSpace(row.status)
	return row, true, nil
}

func loadInstalledPackDeclarationRow(ctx context.Context, db *sql.DB, packName string) (installedPackDeclarationRow, bool, error) {
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return installedPackDeclarationRow{}, false, nil
	}
	var row installedPackDeclarationRow
	err := db.QueryRowContext(ctx, `
SELECT Name, Category, Status, COALESCE(IsDeleted, 0), ManifestFormatVersion, SourceKind, SourceURI, SourceRevision, AppliedPayloadHash, DesiredPayloadHash
FROM _System_InstalledPack
WHERE Name = ?
LIMIT 1`, packName).Scan(
		&row.name,
		&row.category,
		&row.status,
		&row.isDeleted,
		&row.manifestFormatVersion,
		&row.sourceKind,
		&row.sourceURI,
		&row.sourceRevision,
		&row.applied,
		&row.desired,
	)
	if err == sql.ErrNoRows {
		return installedPackDeclarationRow{}, false, nil
	}
	if err != nil {
		return installedPackDeclarationRow{}, false, fmt.Errorf("query installed pack %s declaration row: %w", packName, err)
	}
	row.name = strings.TrimSpace(row.name)
	row.category = strings.TrimSpace(row.category)
	row.status = strings.TrimSpace(row.status)
	return row, true, nil
}

func loadInstalledPackDeclarationRowByInstallID(ctx context.Context, db *sql.DB, packName, packInstallID string) (installedPackDeclarationRow, bool, error) {
	packName = strings.TrimSpace(packName)
	packInstallID = strings.TrimSpace(packInstallID)
	if packName == "" || packInstallID == "" {
		return installedPackDeclarationRow{}, false, nil
	}
	var row installedPackDeclarationRow
	err := db.QueryRowContext(ctx, `
SELECT Name, Category, Status, COALESCE(IsDeleted, 0), ManifestFormatVersion, SourceKind, SourceURI, SourceRevision, AppliedPayloadHash, DesiredPayloadHash
FROM _System_InstalledPack
WHERE Name = ?
  AND PackInstallId = ?
LIMIT 1`, packName, packInstallID).Scan(
		&row.name,
		&row.category,
		&row.status,
		&row.isDeleted,
		&row.manifestFormatVersion,
		&row.sourceKind,
		&row.sourceURI,
		&row.sourceRevision,
		&row.applied,
		&row.desired,
	)
	if err == sql.ErrNoRows {
		return installedPackDeclarationRow{}, false, nil
	}
	if err != nil {
		return installedPackDeclarationRow{}, false, fmt.Errorf("query installed pack %s (%s) declaration row: %w", packName, packInstallID, err)
	}
	row.name = strings.TrimSpace(row.name)
	row.category = strings.TrimSpace(row.category)
	row.status = strings.TrimSpace(row.status)
	return row, true, nil
}

// IsCapabilityDeclaredByConnectorPack returns (true, nil) when ALL of the
// following are satisfied in a SINGLE atomic snapshot:
//
//  1. An _System_InstalledPack row exists for Name=packName with:
//     Category="connector-pack", Status="active", IsDeleted=0,
//     non-blank PackInstallId.
//  2. AppliedPayloadHash and DesiredPayloadHash are both present, non-empty,
//     and equal (no in-flight install / no hash drift).
//  3. A _System_ConnectorCapabilityDeclaration row exists with
//     InstalledPackName=packName, Namespace=namespace, Kind=kind,
//     IsDeleted=0.
//
// All three groups of conditions are checked in ONE JOINed SQL query, NOT
// across multiple sequential queries. This prevents a concurrent lifecycle
// transition from changing the active pack row between an initial liveness
// read and a later declaration read.
//
// The SQL WHERE clause intentionally mirrors ResolveConnectorPackByNamespace
// for the connector-pack liveness conditions (Category/Status/IsDeleted/
// PackInstallId). Per `feedback_pr_loop_discipline_rules.md` Rule 3, the
// canonical resolver is the SSoT for "this connector-pack is live";
// this JOIN encodes the same predicate inline because cross-table
// atomicity requires a single query plan. If the resolver's predicate
// changes (e.g. a new column added to the active-row contract), this
// JOIN must be updated in lock-step.
//
// Returns (false, nil) for any missing/inactive/wrong-Category/
// blank-PackInstallId/hash-drift state OR a missing/soft-deleted
// capability declaration row.
// Returns (false, error) only for DB errors.
//
// Reads DB only — never accesses the source tree.
func IsCapabilityDeclaredByConnectorPack(ctx context.Context, db *sql.DB, packName, namespace, kind string) (bool, error) {
	packName = strings.TrimSpace(packName)
	namespace = strings.TrimSpace(namespace)
	kind = strings.TrimSpace(kind)
	if packName == "" || namespace == "" || kind == "" {
		return false, nil
	}

	var exists int
	err := db.QueryRowContext(ctx, `
SELECT 1
  FROM _System_ConnectorCapabilityDeclaration d
  JOIN _System_InstalledPack p ON p.Name = d.InstalledPackName
 WHERE p.Name = ?
   AND p.Category = ?
   AND p.Status = ?
   AND COALESCE(p.IsDeleted, 0) = 0
   AND COALESCE(p.PackInstallId, '') <> ''
   AND p.AppliedPayloadHash IS NOT NULL
   AND p.AppliedPayloadHash <> ''
   AND p.DesiredPayloadHash IS NOT NULL
   AND p.DesiredPayloadHash <> ''
   AND p.AppliedPayloadHash = p.DesiredPayloadHash
   AND d.Namespace = ?
   AND d.Kind = ?
   AND COALESCE(d.IsDeleted, 0) = 0
 LIMIT 1`,
		packName, string(ConnectorPack), installedPackStatusActive,
		namespace, kind,
	).Scan(&exists)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("query connector capability declaration %s:%s for pack %s: %w", namespace, kind, packName, err)
	}
	return true, nil
}