package records

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

	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
)

const RuntimeRollupDeltaWorkFamily = "runtime.rollup-delta.v1"

const runtimeRollupDeltaWorkHintReadSQL = `SELECT 1
  FROM _System_RollupDelta USE INDEX (idx_rollup_delta_scope_claim)
  WHERE ScopeDigest = ? AND Status = ?
    AND (NextAttemptAt IS NULL OR NextAttemptAt <= NOW())
  LIMIT 1`

const runtimeRollupDeltaWorkHintSQL = `SELECT EXISTS(` + runtimeRollupDeltaWorkHintReadSQL + `)`

// RollupDeltaDefinitionBinding preserves the family API while the immutable
// metadata identity has one generic Catalog-owned representation shared by all
// metadata-driven Runtime work.
type RollupDeltaDefinitionBinding = catalogsnapshot.RuntimeWorkBinding

// RollupDeltaExecutionBinder verifies the persisted content binding and
// returns a context carrying the exact RuntimeMetadataView. Artifact I/O must
// happen before the queue transaction starts.
type RollupDeltaExecutionBinder func(context.Context, RollupDeltaDefinitionBinding) (context.Context, error)

type runtimeRollupDeltaWorkerConfig struct {
	db        *sql.DB
	ownership *persistence.RuntimeWorkOwnershipRepository
	scope     persistence.RuntimeWorkScope
	binder    RollupDeltaExecutionBinder
	max       int
}

var runtimeRollupDeltaRequiredColumns = map[string]string{
	"Id": "varchar(255)", "SourceObject": "varchar(255)", "SourceRecordId": "varchar(255)",
	"ParentObject": "varchar(255)", "ParentRecordId": "varchar(255)", "RollupField": "varchar(255)",
	"Delta": "bigint", "DedupeKey": "varchar(64)",
	"ScopeDigest": "char(64)",
	"OrgId":       "varchar(255)", "AppInstallationId": "varchar(255)", "RuntimeCapabilityId": "varchar(255)",
	"CatalogRevision": "decimal(20,0)", "SchemaGeneration": "decimal(20,0)", "EngineGeneration": "decimal(20,0)",
	"CatalogSnapshotDigest": "char(71)", "RuntimeMetadataViewDigest": "char(71)",
	"Status": "varchar(32)", "RetryCount": "int", "LastError": "text", "LastErrorAt": "datetime",
	"NextAttemptAt": "datetime", "ProcessedAt": "datetime",
	"RepairId": "varchar(255)", "RepairReason": "varchar(1024)", "RepairCount": "int", "LastRepairAt": "datetime(6)",
	"CreatedDate": "datetime", "LastModifiedDate": "datetime",
}

// VerifyRuntimeRollupDeltaQueueSchema proves the durable producer/consumer
// table can carry one exact metadata binding plus bounded retry/repair state.
// Worker ownership and scope remain separate authorities.
func VerifyRuntimeRollupDeltaQueueSchema(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return errors.New("rollup delta: database is required")
	}
	if err := persistence.VerifyRequiredTableColumns(ctx, db, "_System_RollupDelta", runtimeRollupDeltaRequiredColumns); err != nil {
		return fmt.Errorf("rollup delta: %w", err)
	}
	if err := persistence.VerifyExactTableIndex(
		ctx, db, "_System_RollupDelta", "idx_rollup_delta_scope_claim", false,
		[]string{"ScopeDigest", "Status", "NextAttemptAt", "CreatedDate", "Id"},
	); err != nil {
		return fmt.Errorf("rollup delta: %w", err)
	}
	if err := persistence.VerifyExactTableIndex(
		ctx, db, "_System_RollupDelta", "idx_rollup_delta_scope_parent_lane", false,
		[]string{"ScopeDigest", "ParentObject", "ParentRecordId", "Status", "NextAttemptAt", "CreatedDate"},
	); err != nil {
		return fmt.Errorf("rollup delta: %w", err)
	}
	// A shape-only check cannot prove the digest read path is complete. Old
	// Runtime producers can write NULL after an earlier additive backfill.
	// Deployment must fence those writers before its final schema apply; this
	// readiness guard refuses to advertise a healthy but incomplete queue.
	var unindexed int
	err := db.QueryRowContext(ctx, `SELECT 1
FROM _System_RollupDelta USE INDEX (idx_rollup_delta_scope_claim)
WHERE ScopeDigest IS NULL AND Status IN ('pending', 'failed')
  AND OrgId IS NOT NULL AND AppInstallationId IS NOT NULL AND RuntimeCapabilityId IS NOT NULL
LIMIT 1`).Scan(&unindexed)
	if err == nil {
		return errors.New("rollup delta: scoped work is missing ScopeDigest; stop old Runtime producers and reapply the canonical schema before worker activation")
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("rollup delta: verify scope digest cutover: %w", err)
	}
	return nil
}

// RuntimeRollupDeltaState is an operational observation. Payload or tenant
// identifiers never leave the worker metrics boundary.
type RuntimeRollupDeltaState struct {
	PendingCount int64
	FailedCount  int64
	OldestAge    time.Duration
}

// RuntimeRollupDeltaObservationStore owns the diagnostic read path for one
// exact Runtime scope. Its output is never admission or claim authority.
type RuntimeRollupDeltaObservationStore struct {
	db        *sql.DB
	ownership *persistence.RuntimeWorkOwnershipRepository
	scope     persistence.RuntimeWorkScope
}

func NewRuntimeRollupDeltaObservationStore(
	db *sql.DB,
	expected persistence.RuntimeWorkOwnership,
	scope persistence.RuntimeWorkScope,
) (*RuntimeRollupDeltaObservationStore, error) {
	if db == nil || scope.Validate() != nil {
		return nil, errors.New("rollup delta observation: database and Runtime scope are required")
	}
	ownership, err := persistence.NewRuntimeWorkOwnershipRepository(db, expected)
	if err != nil {
		return nil, err
	}
	return &RuntimeRollupDeltaObservationStore{db: db, ownership: ownership, scope: scope}, nil
}

func (s *RuntimeRollupDeltaObservationStore) Observe(ctx context.Context) (RuntimeRollupDeltaState, error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return RuntimeRollupDeltaState{}, errors.New("rollup delta observation: store is unavailable")
	}
	if err := s.ownership.VerifyActive(ctx); err != nil {
		return RuntimeRollupDeltaState{}, err
	}
	var state RuntimeRollupDeltaState
	var oldestSeconds sql.NullFloat64
	err := s.db.QueryRowContext(ctx, `SELECT
  COALESCE(SUM(CASE WHEN Status = 'pending' THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN Status = 'failed' THEN 1 ELSE 0 END), 0),
  TIMESTAMPDIFF(MICROSECOND, MIN(CASE WHEN Status = 'pending' THEN CreatedDate END), NOW(6)) / 1000000.0
FROM _System_RollupDelta
WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND Status IN (?, ?)`,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID,
		rollupDeltaStatusPending, rollupDeltaStatusFailed,
	).Scan(&state.PendingCount, &state.FailedCount, &oldestSeconds)
	if err != nil {
		return RuntimeRollupDeltaState{}, fmt.Errorf("rollup delta observation: observe Runtime queue: %w", err)
	}
	if oldestSeconds.Valid && oldestSeconds.Float64 > 0 {
		state.OldestAge = time.Duration(oldestSeconds.Float64 * float64(time.Second))
	}
	return state, nil
}

// ConfigureRuntimeRollupDeltaWorker turns this engine graph into the sole
// fenced consumer for rollup-delta work. It is intentionally separate from
// enqueue wiring so a Runtime API can never start an implicit worker.
func (ps *PersistenceService) ConfigureRuntimeRollupDeltaWorker(
	db *sql.DB,
	ownership *persistence.RuntimeWorkOwnershipRepository,
	scope persistence.RuntimeWorkScope,
	binder RollupDeltaExecutionBinder,
	maxAttempts int,
) error {
	if ps == nil || db == nil || ownership == nil || scope.Validate() != nil || binder == nil || maxAttempts <= 0 || maxAttempts > 20 {
		return errors.New("rollup delta: worker ownership, definition binder, and bounded attempts are required")
	}
	ps.rollupDeltaWorker = &runtimeRollupDeltaWorkerConfig{db: db, ownership: ownership, scope: scope, binder: binder, max: maxAttempts}
	return nil
}

// CheckRuntimeRollupDeltaReady proves both the payload schema and the exact
// deployment authority before the process advertises readiness.
func (ps *PersistenceService) CheckRuntimeRollupDeltaReady(ctx context.Context) error {
	if ps == nil || ps.rollupDeltaWorker == nil {
		return errors.New("rollup delta: fenced Runtime Worker graph is required")
	}
	worker := ps.rollupDeltaWorker
	if err := VerifyRuntimeRollupDeltaQueueSchema(ctx, worker.db); err != nil {
		return err
	}
	return worker.ownership.CheckReady(ctx)
}

// HasRuntimeRollupDeltaWork is a bounded admission hint for the exact Runtime
// scope. It is not claim authority: ProcessRollupDeltaBatch rechecks the row
// under the existing ownership and transaction fences. Callers must fail open
// to that authoritative path when this hint cannot be read.
func (ps *PersistenceService) HasRuntimeRollupDeltaWork(ctx context.Context) (bool, error) {
	if ps == nil || ps.rollupDeltaWorker == nil {
		return false, errors.New("rollup delta: fenced Runtime Worker graph is required")
	}
	worker := ps.rollupDeltaWorker
	if err := worker.ownership.VerifyActive(ctx); err != nil {
		return false, err
	}
	var exists bool
	scopeDigest, err := worker.scope.Digest()
	if err != nil {
		return false, err
	}
	err = worker.db.QueryRowContext(ctx, runtimeRollupDeltaWorkHintSQL,
		scopeDigest, rollupDeltaStatusPending,
	).Scan(&exists)
	if err != nil {
		return false, fmt.Errorf("rollup delta: hint Runtime queue: %w", err)
	}
	return exists, nil
}

func (ps *PersistenceService) runtimeRollupDeltaBinding(ctx context.Context) (RollupDeltaDefinitionBinding, bool) {
	if ps == nil || ps.metadata == nil {
		return RollupDeltaDefinitionBinding{}, false
	}
	view, ok := ps.metadata.BoundRuntimeMetadataView(ctx)
	if !ok {
		return RollupDeltaDefinitionBinding{}, false
	}
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	return binding, err == nil
}

// BindRuntimeMetadataView is the shared request/worker context seam. It is
// exposed here only as a convenience for worker composition tests; metadata
// remains the owner of the context key.
func BindRuntimeMetadataView(ctx context.Context, view *catalogsnapshot.RuntimeMetadataView) context.Context {
	return metadata.WithRuntimeMetadataView(ctx, view)
}

type RollupDeltaRepair = persistence.RuntimePoisonRepair

// RuntimeRollupDeltaRepairStore is the short-lived operator repair surface.
// It shares the exact ownership lock with the long-running worker but does not
// construct a serving or mutation-command graph.
type RuntimeRollupDeltaRepairStore struct {
	db        *sql.DB
	ownership *persistence.RuntimeWorkOwnershipRepository
	scope     persistence.RuntimeWorkScope
}

func NewRuntimeRollupDeltaRepairStore(db *sql.DB, ownership *persistence.RuntimeWorkOwnershipRepository, scope persistence.RuntimeWorkScope) (*RuntimeRollupDeltaRepairStore, error) {
	if db == nil || ownership == nil || scope.Validate() != nil {
		return nil, errors.New("rollup delta repair: database, ownership authority, and work scope are required")
	}
	return &RuntimeRollupDeltaRepairStore{db: db, ownership: ownership, scope: scope}, nil
}

func (s *RuntimeRollupDeltaRepairStore) CheckReady(ctx context.Context) error {
	if s == nil || s.db == nil || s.ownership == nil || s.scope.Validate() != nil {
		return errors.New("rollup delta repair: store is unavailable")
	}
	if err := VerifyRuntimeRollupDeltaQueueSchema(ctx, s.db); err != nil {
		return fmt.Errorf("rollup delta repair: %w", err)
	}
	return s.ownership.CheckReady(ctx)
}

func (s *RuntimeRollupDeltaRepairStore) Repair(ctx context.Context, repair RollupDeltaRepair) (err error) {
	if s == nil || s.db == nil || s.ownership == nil {
		return errors.New("rollup delta repair: store is unavailable")
	}
	if err := repair.Validate(); err != nil {
		return err
	}
	if err := s.ownership.VerifyActive(ctx); err != nil {
		return err
	}
	tx, cleanup, err := persistence.BeginPessimisticTx(ctx, s.db)
	if err != nil {
		return fmt.Errorf("rollup delta repair: begin transaction: %w", err)
	}
	defer func() { _ = cleanup() }()
	defer func() {
		if err != nil {
			_ = tx.Rollback()
		}
	}()
	if err = s.ownership.LockActive(ctx, tx); err != nil {
		return err
	}
	var retryCount int
	var lastError string
	if err = tx.QueryRowContext(ctx, `SELECT RetryCount, COALESCE(LastError, '')
FROM _System_RollupDelta
WHERE Id = ? AND Status = 'failed'
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
FOR UPDATE`, repair.RowID, s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID).Scan(&retryCount, &lastError); err != nil {
		return fmt.Errorf("rollup delta repair: lock exact poison row: %w", err)
	}
	if !repair.Matches(retryCount, lastError) {
		return errors.New("rollup delta repair: poison observation changed")
	}
	result, err := tx.ExecContext(ctx, `UPDATE _System_RollupDelta
SET Status = 'pending', RetryCount = 0, NextAttemptAt = NULL,
    RepairId = ?, RepairReason = ?, RepairCount = RepairCount + 1,
    LastRepairAt = CURRENT_TIMESTAMP(6), LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND Status = 'failed' AND RetryCount = ?
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?`,
		repair.RepairID, strings.TrimSpace(repair.Reason), repair.RowID, repair.ExpectedRetryCount,
		s.scope.OrgID, s.scope.AppInstallationID, s.scope.RuntimeCapabilityID)
	if err != nil {
		return fmt.Errorf("rollup delta repair: requeue exact poison row: %w", err)
	}
	rows, err := result.RowsAffected()
	if err != nil || rows != 1 {
		return errors.New("rollup delta repair: poison row changed before replay")
	}
	if err = tx.Commit(); err != nil {
		return fmt.Errorf("rollup delta repair: commit exact replay: %w", err)
	}
	return nil
}

// RepairRuntimeRollupDelta requeues one exact poison row under the same
// deployment fence as normal completion. It never edits the business effect or
// creates a second queue identity.
func (ps *PersistenceService) RepairRuntimeRollupDelta(ctx context.Context, repair RollupDeltaRepair) error {
	if ps == nil || ps.rollupDeltaWorker == nil {
		return errors.New("rollup delta repair: fenced Runtime Worker graph is required")
	}
	worker := ps.rollupDeltaWorker
	store, err := NewRuntimeRollupDeltaRepairStore(worker.db, worker.ownership, worker.scope)
	if err != nil {
		return err
	}
	return store.Repair(ctx, repair)
}