package extension

import (
	"context"
	"fmt"
	"math"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
)

// CatalogBatchAgent is one active Agent selected for a bounded Workspace
// reconcile batch.
type CatalogBatchAgent struct {
	AgentID string
}

// CatalogRowError is a deterministic per-Agent catalog-row error. The service
// can keep processing healthy Agents in the same batch and surface one bounded
// warning after the batch instead of retrying the whole Workspace forever.
type CatalogRowError struct {
	WorkspaceID string
	AgentID     string
	Err         error
}

func (e CatalogRowError) Error() string {
	if e.Err == nil {
		return fmt.Sprintf("catalog row %q/%q is invalid", e.WorkspaceID, e.AgentID)
	}
	return fmt.Sprintf("catalog row %q/%q is invalid: %v", e.WorkspaceID, e.AgentID, e.Err)
}

func (e CatalogRowError) Unwrap() error { return e.Err }

// ListWorkspaceCatalogAgentBatch returns a stable page of active Agents in one
// Workspace. The cursor is the last processed agent_id from the previous batch.
func (tx *TxStore) ListWorkspaceCatalogAgentBatch(ctx context.Context, workspaceID string, afterAgentID string, limit int) ([]CatalogBatchAgent, error) {
	if tx == nil || tx.db == nil {
		return nil, fmt.Errorf("extension store transaction is not configured")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	afterAgentID = strings.TrimSpace(afterAgentID)
	if workspaceID == "" {
		return nil, fmt.Errorf("workspaceId is required")
	}
	if limit <= 0 {
		return nil, fmt.Errorf("limit must be positive")
	}
	rows, err := tx.db.Query(ctx, `
SELECT agent.id
FROM agents agent
JOIN members member
  ON member.id = agent.member_id
 AND member.workspace_id = agent.workspace_id
 AND member.type = 'agent'
 AND member.status = 'active'
JOIN machines machine
  ON machine.id = agent.machine_id
 AND machine.workspace_id = agent.workspace_id
 AND machine.status = 'active'
WHERE agent.workspace_id = $1
  AND agent.status <> 'archived'
  AND agent.id > $2
ORDER BY agent.id ASC
LIMIT $3
`, workspaceID, afterAgentID, limit)
	if err != nil {
		return nil, fmt.Errorf("list Workspace catalog Agent batch: %w", err)
	}
	defer rows.Close()
	agents := make([]CatalogBatchAgent, 0, limit)
	for rows.Next() {
		var agent CatalogBatchAgent
		if err := rows.Scan(&agent.AgentID); err != nil {
			return nil, fmt.Errorf("scan Workspace catalog Agent: %w", err)
		}
		if strings.TrimSpace(agent.AgentID) == "" {
			return nil, fmt.Errorf("Workspace catalog Agent id is empty")
		}
		agents = append(agents, agent)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate Workspace catalog Agents: %w", err)
	}
	return agents, nil
}

// LoadAgentExtensionCatalogsForUpdate batch-loads existing catalog rows for the
// requested Agents. Missing rows are omitted. Invalid durable rows are returned
// as per-Agent CatalogRowError values so one corrupt Agent does not starve the
// rest of the batch.
func (tx *TxStore) LoadAgentExtensionCatalogsForUpdate(ctx context.Context, workspaceID string, agentIDs []string) (map[string]AgentExtensionCatalog, []CatalogRowError, error) {
	if tx == nil || tx.db == nil {
		return nil, nil, fmt.Errorf("extension store transaction is not configured")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	agentIDs = normalizeNonEmptyStrings(agentIDs)
	if workspaceID == "" {
		return nil, nil, fmt.Errorf("workspaceId is required")
	}
	catalogs := make(map[string]AgentExtensionCatalog, len(agentIDs))
	if len(agentIDs) == 0 {
		return catalogs, nil, nil
	}
	requestedAgentsCTE, requestedAgentArgs, _, err := coredb.BuildStringValuesCTE("requested_agents", "agent_id", agentIDs, 2)
	if err != nil {
		return nil, nil, fmt.Errorf("build requested Agent catalog CTE: %w", err)
	}
	args := append([]any{workspaceID}, requestedAgentArgs...)
	rows, err := tx.db.Query(ctx, fmt.Sprintf(`
WITH %s
SELECT requested.agent_id,
       catalog.workspace_id, catalog.agent_id, catalog.version,
       catalog.catalog_state,
       catalog.enabled, catalog.created_at,
       catalog.updated_at
FROM requested_agents requested
JOIN agent_extension_catalogs catalog
  ON catalog.workspace_id = $1
 AND catalog.agent_id = requested.agent_id
ORDER BY requested.agent_id ASC
FOR UPDATE
`, requestedAgentsCTE), args...)
	if err != nil {
		return nil, nil, fmt.Errorf("load Agent Extension Catalog batch: %w", err)
	}
	defer rows.Close()
	var rowErrors []CatalogRowError
	for rows.Next() {
		var requestedAgentID string
		var catalog AgentExtensionCatalog
		if err := rows.Scan(
			&requestedAgentID,
			&catalog.WorkspaceID,
			&catalog.AgentID,
			&catalog.Version,
			&catalog.State,
			&catalog.Enabled,
			&catalog.CreatedAt,
			&catalog.UpdatedAt,
		); err != nil {
			return nil, nil, fmt.Errorf("scan Agent Extension Catalog batch: %w", err)
		}
		if catalog.WorkspaceID != workspaceID || catalog.AgentID != requestedAgentID {
			rowErrors = append(rowErrors, CatalogRowError{
				WorkspaceID: workspaceID,
				AgentID:     requestedAgentID,
				Err:         storeerr.Forbidden("agent extension catalog identity does not match the request"),
			})
			continue
		}
		if err := validateCatalogState(catalog.Version, catalog.State, catalog.Enabled); err != nil {
			rowErrors = append(rowErrors, CatalogRowError{
				WorkspaceID: workspaceID,
				AgentID:     requestedAgentID,
				Err:         err,
			})
			continue
		}
		catalogs[requestedAgentID] = catalog
	}
	if err := rows.Err(); err != nil {
		return nil, nil, fmt.Errorf("iterate Agent Extension Catalog batch: %w", err)
	}
	return catalogs, rowErrors, nil
}

// ApplyPrecomputedExtensionCatalogChange persists one already-computed
// effective Catalog state. It does not read authorization or selection state,
// so a Workspace reconcile worker can batch those reads once and perform sparse
// writes only for Agents whose state actually changed.
func (tx *TxStore) ApplyPrecomputedExtensionCatalogChange(
	ctx context.Context,
	workspaceID string,
	agentID string,
	current AgentExtensionCatalog,
	found bool,
	newState string,
	newEnabled bool,
	now time.Time,
) (int64, error) {
	if tx == nil || tx.db == nil {
		return 0, fmt.Errorf("extension store transaction is not configured")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	agentID = strings.TrimSpace(agentID)
	if workspaceID == "" || agentID == "" {
		return 0, fmt.Errorf("workspaceId and agentId are required")
	}
	now = now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	if err := validateCatalogState(1, newState, newEnabled); err != nil {
		return 0, err
	}
	if !found {
		result, err := tx.db.Exec(ctx, `
INSERT INTO agent_extension_catalogs (
    workspace_id, agent_id, version, catalog_state, enabled,
    created_at, updated_at
) VALUES (
    $1, $2, 1, $3, $4,
    $5, $5
)`, workspaceID, agentID, newState, newEnabled, now)
		if err != nil {
			return 0, fmt.Errorf("insert Agent Extension Catalog: %w", err)
		}
		if result.RowsAffected() == 0 {
			return 0, storeerr.Conflict("agent extension catalog insert raced")
		}
		if err := tx.insertAgentExtensionCatalogUpdate(ctx, workspaceID, agentID, 1, now); err != nil {
			return 0, err
		}
		return 1, nil
	}
	if current.WorkspaceID != workspaceID || current.AgentID != agentID {
		return 0, storeerr.Forbidden("agent extension catalog identity does not match the request")
	}
	if err := validateCatalogState(current.Version, current.State, current.Enabled); err != nil {
		return 0, err
	}
	if current.Version == math.MaxInt64 {
		return 0, storeerr.Forbidden("agent extension catalog version overflow")
	}
	if err := validateCatalogState(current.Version+1, newState, newEnabled); err != nil {
		return 0, err
	}
	newVersion := current.Version + 1
	result, err := tx.db.Exec(ctx, `
UPDATE agent_extension_catalogs
SET version = version + 1, catalog_state = $4,
    enabled = $5, updated_at = $6
WHERE workspace_id = $1 AND agent_id = $2 AND version = $3`,
		workspaceID, agentID, current.Version, newState, newEnabled, now)
	if err != nil {
		return 0, fmt.Errorf("advance Agent Extension Catalog: %w", err)
	}
	if result.RowsAffected() == 0 {
		return 0, storeerr.Conflict("agent extension catalog version changed concurrently")
	}
	if err := tx.insertAgentExtensionCatalogUpdate(ctx, workspaceID, agentID, newVersion, now); err != nil {
		return 0, err
	}
	return newVersion, nil
}

func (tx *TxStore) insertAgentExtensionCatalogUpdate(ctx context.Context, workspaceID string, agentID string, version int64, now time.Time) error {
	if _, err := tx.db.Exec(ctx, `
INSERT INTO agent_extension_catalog_updates (workspace_id, agent_id, version, status, attempts, created_at, updated_at)
VALUES ($1, $2, $3, 'pending', 0, $4, $4)`, workspaceID, agentID, version, now); err != nil {
		return fmt.Errorf("insert agent extension catalog update: %w", err)
	}
	return nil
}

func normalizeNonEmptyStrings(values []string) []string {
	result := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		result = append(result, value)
	}
	return result
}