package extensionflow

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"github.com/tidbcloud/AgentTeam/internal/store/extension"
)

// CatalogDeliveryHub is the minimal Hub surface the delivery worker needs:
// push the catalog to an online Agent's machine. *daemon.Hub satisfies it.
type CatalogDeliveryHub interface {
	SyncMachineCatalog(ctx context.Context, machineID string) error
}

// StartCatalogDeliveryLoop launches a background worker that drains pending
// catalog update events: claim → resolve machine → SyncMachineCatalog → mark
// delivered (or requeue/fail). It runs until ctx is cancelled. Call once at
// server startup (Extension profile).
func (service *Service) StartCatalogDeliveryLoop(ctx context.Context, hub CatalogDeliveryHub, interval time.Duration) {
	if hub == nil || interval <= 0 {
		return
	}
	logging.Go(nil, func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				service.drainPendingCatalogUpdates(ctx, hub)
			}
		}
	})
}

// drainPendingCatalogUpdates claims and delivers as many pending updates as are
// available in one tick. It stops when no more claimable updates exist.
func (service *Service) drainPendingCatalogUpdates(ctx context.Context, hub CatalogDeliveryHub) {
	for {
		update, found, err := service.claimOneUpdate(ctx)
		if err != nil || !found {
			return
		}
		service.deliverOneUpdate(ctx, hub, update)
	}
}

func (service *Service) claimOneUpdate(ctx context.Context) (extension.CatalogUpdate, bool, error) {
	store := service.store
	var update extension.CatalogUpdate
	var found bool
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		u, f, e := store.InTx(tx).ClaimNextCatalogUpdate(ctx, "delivery-worker", 30*time.Second, service.now().UTC())
		update, found = u, f
		return e
	})
	return update, found, err
}

func (service *Service) deliverOneUpdate(ctx context.Context, hub CatalogDeliveryHub, update extension.CatalogUpdate) {
	now := service.now().UTC()

	// Resolve the Agent's current machine.
	machineID, err := service.resolveAgentMachine(ctx, update.WorkspaceID, update.AgentID)
	if err != nil {
		service.requeueUpdate(ctx, update, "resolve machine: "+err.Error(), now)
		return
	}

	// Fail-closed hint gate (stage-6 final-review blocker 5): read the
	// authoritative catalog hint BEFORE pushing, so a corrupt durable row (PAD
	// SPACE identity mismatch, bad revision/state) is never delivered to a
	// machine. SyncMachineCatalog re-reads it too; this worker-level read makes
	// the guarantee provable end-to-end without trusting the Hub.
	if _, hintErr := service.readCatalogHintForDelivery(ctx, update.WorkspaceID, update.AgentID); hintErr != nil {
		service.requeueUpdate(ctx, update, "hint read: "+hintErr.Error(), now)
		return
	}

	// Push the catalog to the machine (SyncMachineCatalog re-reads the hint
	// from the catalog row at delivery time — never a frozen payload).
	if err := hub.SyncMachineCatalog(ctx, machineID); err != nil {
		service.requeueUpdate(ctx, update, "sync: "+err.Error(), now)
		return
	}

	// Mark delivered (terminal).
	_ = coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		return service.store.InTx(tx).MarkCatalogUpdateDelivered(ctx, update.WorkspaceID, update.AgentID, update.Version, update.LeaseOwner, now)
	})
}

func (service *Service) requeueUpdate(ctx context.Context, update extension.CatalogUpdate, errMsg string, now time.Time) {
	const maxDeliveryAttempts = 5
	if update.Attempts >= maxDeliveryAttempts {
		// Terminal fail: exhausted retries. The event stays in 'failed' status
		// (poison-event backstop — no infinite requeue agentteam).
		_ = coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
			return service.store.InTx(tx).FailCatalogUpdate(ctx, update.WorkspaceID, update.AgentID, update.Version, update.LeaseOwner, errMsg, now)
		})
		return
	}
	_ = coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		return service.store.InTx(tx).RequeueCatalogUpdate(ctx, update.WorkspaceID, update.AgentID, update.Version, update.LeaseOwner, errMsg, now.Add(time.Minute), now)
	})
}

// resolveAgentMachine finds the Agent's current machine_id. The Agent must be
// active + in this workspace (the same canonical shape AssertWorkspaceAgent
// uses, but returning the machine_id).
func (service *Service) resolveAgentMachine(ctx context.Context, workspaceID, agentID string) (string, error) {
	var machineID string
	err := service.db.QueryRow(ctx, `
SELECT a.machine_id
FROM agents a
JOIN members m ON m.id = a.member_id AND m.workspace_id = a.workspace_id AND m.type = 'agent'
WHERE a.id = $1 AND a.workspace_id = $2 AND a.status <> 'archived'
LIMIT 1`, agentID, workspaceID).Scan(&machineID)
	return machineID, err
}

// readCatalogHintForDelivery reads the Agent's authoritative catalog hint for
// the delivery fail-closed gate. A corrupt durable row returns an error so the
// worker requeues instead of pushing a bad hint to a machine.
func (service *Service) readCatalogHintForDelivery(ctx context.Context, workspaceID, agentID string) (extension.CatalogHint, error) {
	var hint extension.CatalogHint
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		h, e := service.store.InTx(tx).ReadAgentCatalogHint(ctx, workspaceID, agentID)
		hint = h
		return e
	})
	return hint, err
}

// ReconcileAgentCatalogs materializes never-seen default-allow catalogs and
// reconciles existing catalogs after app connection/install lifecycle changes.
// Every candidate runs in an independent transaction. A corrupt candidate fails
// closed but does not starve later Agents; all errors are returned with
// errors.Join. The count is the number of Agents whose version actually
// advanced.
func (service *Service) ReconcileAgentCatalogs(ctx context.Context) (int, error) {
	if err := service.validate(); err != nil {
		return 0, err
	}
	if err := service.requireRegistry(); err != nil {
		return 0, err
	}
	candidates, err := service.store.ListCatalogReconcileCandidates(ctx)
	if err != nil {
		return 0, err
	}
	changedCount := 0
	var candidateErrors []error
	now := service.now().UTC()
	for _, candidate := range candidates {
		var changed bool
		applyErr := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
			var err error
			changed, err = service.applyAgentCatalogChange(ctx, tx, candidate.WorkspaceID, candidate.AgentID, now)
			return err
		})
		if applyErr != nil {
			candidateErrors = append(candidateErrors, fmt.Errorf(
				"reconcile catalog for %q/%q: %w",
				candidate.WorkspaceID,
				candidate.AgentID,
				applyErr,
			))
			continue
		}
		if changed {
			changedCount++
		}
	}
	return changedCount, errors.Join(candidateErrors...)
}

// BackfillDefaultAllowCatalogs is the compatibility entry point retained for
// callers/tests from the initial one-shot implementation. It now performs one
// complete bidirectional reconciliation pass.
func (service *Service) BackfillDefaultAllowCatalogs(ctx context.Context) (int, error) {
	return service.ReconcileAgentCatalogs(ctx)
}

// AgentCatalogSnapshot is a consistent point-in-time view of an Agent's
// Extension surface: the authorized entries plus the server-internal durable
// revision/state, read in one transaction so a concurrent Grant/Revoke cannot
// return entries from a different revision.
type AgentCatalogSnapshot struct {
	Entries []AgentCatalogEntry
	Version int64
	State   string
	Enabled bool
}

// AgentCatalogSnapshot returns the Agent's authorized Extension catalog entries
// AND the durable catalog state in a single transaction, so the Root response is
// a consistent snapshot (stage-6 final-review blocker 4).
func (service *Service) AgentCatalogSnapshot(ctx context.Context, workspaceID, agentID string) (AgentCatalogSnapshot, error) {
	var snap AgentCatalogSnapshot
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		extTx := service.store.InTx(tx)
		entries, _, err := service.projectAppHostedCatalogEntries(
			ctx,
			tx,
			extTx,
			workspaceID,
			agentID,
		)
		if err != nil {
			return err
		}
		catalog, found, err := extTx.ReadAgentExtensionCatalog(
			ctx,
			workspaceID,
			agentID,
		)
		if err != nil {
			return err
		}
		snap.Entries = make([]AgentCatalogEntry, 0, len(entries))
		for _, entry := range entries {
			actions := make([]string, 0, len(entry.Actions))
			for _, a := range entry.Actions {
				actions = append(actions, a.Name)
			}
			snap.Entries = append(snap.Entries, AgentCatalogEntry{
				ExtensionName:     entry.ExtensionName,
				Domain:            entry.DefinitionDomain,
				DefinitionVersion: entry.DefinitionVersion,
				DefinitionDigest:  entry.DefinitionDigest,
				Actions:           actions,
			})
		}
		if !found {
			if len(entries) != 0 {
				return fmt.Errorf(
					"agent catalog snapshot: live entries exist without a durable catalog",
				)
			}
			return nil
		}
		snap.Version = catalog.Version
		snap.State = catalog.State
		snap.Enabled = catalog.Enabled
		// Semantic consistency (stage-6 final-review blocker 4): the live
		// entries and the durable state MUST agree. v0 with non-empty entries,
		// or a stored state that does not match the recomputed one, means the
		// catalog is unreconciled (a lifecycle update was missed); fail closed
		// rather than return an inconsistent snapshot.
		if catalog.Version == 0 && len(entries) > 0 {
			return fmt.Errorf("agent catalog snapshot: catalog is v0 but live entries exist (unreconciled)")
		}
		if catalog.Version > 0 {
			recomputed, dErr := marshalCatalogState(entries)
			if dErr != nil {
				return fmt.Errorf("agent catalog snapshot: recompute state: %w", dErr)
			}
			if catalog.State != recomputed ||
				catalog.Enabled != (len(entries) > 0) {
				return fmt.Errorf(
					"agent catalog snapshot: stored state does not match live entries (stale)",
				)
			}
		}
		return nil
	})
	return snap, err
}

// AgentCatalogEntry is the public, secret-free projection of one effective
// App-hosted Extension selection: the Agent-visible name, the server-internal
// Definition identity needed for exact descriptor projection, and precise
// allowed action names. App host identities remain internal.
type AgentCatalogEntry struct {
	ExtensionName     string
	Domain            string
	DefinitionVersion int
	DefinitionDigest  string
	Actions           []string
}

// AgentCatalogEntries returns the Agent's flattened effective App-hosted
// Extension entries, the same surface the Catalog state is derived over.
func (service *Service) AgentCatalogEntries(ctx context.Context, workspaceID, agentID string) ([]AgentCatalogEntry, error) {
	var result []AgentCatalogEntry
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		extTx := service.store.InTx(tx)
		entries, _, e := service.projectAppHostedCatalogEntries(
			ctx,
			tx,
			extTx,
			workspaceID,
			agentID,
		)
		if e != nil {
			return e
		}
		result = make([]AgentCatalogEntry, 0, len(entries))
		for _, entry := range entries {
			actions := make([]string, 0, len(entry.Actions))
			for _, a := range entry.Actions {
				actions = append(actions, a.Name)
			}
			result = append(result, AgentCatalogEntry{
				ExtensionName:     entry.ExtensionName,
				Domain:            entry.DefinitionDomain,
				DefinitionVersion: entry.DefinitionVersion,
				DefinitionDigest:  entry.DefinitionDigest,
				Actions:           actions,
			})
		}
		return nil
	})
	return result, err
}

// ApplyAgentCatalogChange recomputes the Agent's flattened App-hosted Catalog
// within the caller's transaction. App authorization and selection mutations
// use this hook so the durable revision/state/outbox update is atomic.
func (service *Service) ApplyAgentCatalogChange(ctx context.Context, tx coredb.Tx, workspaceID, agentID string, now time.Time) error {
	_, err := service.applyAgentCatalogChange(ctx, tx, workspaceID, agentID, now)
	return err
}

// applyAgentCatalogChange is the shared effective-surface no-op authority used
// by both the same-transaction App Enable/Disable hook and the periodic
// reconciler. It reports whether the Catalog version actually advanced.
func (service *Service) applyAgentCatalogChange(ctx context.Context, tx coredb.Tx, workspaceID, agentID string, now time.Time) (bool, error) {
	extTx := service.store.InTx(tx)
	decisionsByInstallation, keepEmptyRow, err := service.agentCatalogChangeAuthorizationDecisions(ctx, extTx, workspaceID, agentID)
	if err != nil {
		return false, err
	}
	changed := false
	_, err = extTx.ApplyExtensionCatalogChange(ctx, workspaceID, agentID, now, func(current int64, storedState string) (string, bool, bool, error) {
		state, enabled, didChange, err := service.projectedCatalogChangeFromAuthorizationDecisions(
			ctx,
			tx,
			extTx,
			workspaceID,
			decisionsByInstallation,
			current,
			storedState,
		)
		changed = didChange
		return state, enabled, didChange, err
	}, extension.ApplyExtensionCatalogChangeOptions{
		KeepEmptyRowOnNoop: keepEmptyRow,
	})
	return changed, err
}

func (service *Service) agentCatalogChangeAuthorizationDecisions(
	ctx context.Context,
	extTx *extension.TxStore,
	workspaceID string,
	agentID string,
) (map[string]appAuthorizationDecision, bool, error) {
	decisions, err := extTx.ListAgentAppAuthorizationDecisions(ctx, workspaceID, agentID)
	if err != nil {
		return nil, false, err
	}
	decisionsByInstallation, err := validateAppAuthorizationDecisions(
		workspaceID,
		agentID,
		decisions,
	)
	if err != nil {
		return nil, false, err
	}
	for _, decision := range decisionsByInstallation {
		if !decision.Allowed {
			return decisionsByInstallation, true, nil
		}
	}
	return decisionsByInstallation, false, nil
}