package invocation

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

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

type lockedDeployment struct {
	WorkspaceID         string
	BindingID           string
	BindingSourceID     string
	BindingActivation   int64
	BindingRevisionID   string
	BindingRevision     int64
	PlanID              string
	PlanVersionID       string
	WorkflowName        string
	RevisionDigest      string
	ConfigJSON          []byte
	PlanArtifactJSON    []byte
	PlanCompilerVersion string
	PlanVersionDigest   string
	BindingState        string
	ActiveSourceID      string
	ActiveRevisionID    string
	SourceStatus        string
	SourceKind          string
	NextOccurrenceAt    *time.Time
	OwnerUserID         string
	OwnerMemberID       string
	BindingCreatedBy    Actor
	PlanCreatedBy       Actor
	PlanPublishedBy     Actor
	SourceActivatedBy   Actor
}

func deploymentSelectSQL() string {
	return `
SELECT b.workspace_id, b.id, s.id, s.activation_number,
       r.id, r.revision_number, r.plan_id, r.plan_version_id,
       p.display_name, r.digest, r.config_json, v.artifact,
       v.compiler_version, v.digest,
       b.state, COALESCE(b.active_source_id, ''), COALESCE(b.active_revision_id, ''),
       s.status, s.source_kind, s.next_occurrence_at,
       b.owner_user_id, b.owner_member_id,
       b.created_by_actor_type, b.created_by_actor_id, b.created_by_member_id,
       p.created_by_actor_type, p.created_by_actor_id, p.created_by_member_id,
       v.published_by_actor_type, v.published_by_actor_id, v.published_by_member_id,
       s.activated_by_actor_type, s.activated_by_actor_id, s.activated_by_member_id
FROM workflow_bindings b
JOIN workflow_binding_sources s
  ON s.workspace_id = b.workspace_id AND s.binding_id = b.id
JOIN workflow_binding_revisions r
  ON r.workspace_id = s.workspace_id
 AND r.binding_id = s.binding_id
 AND r.id = s.binding_revision_id
JOIN workflow_plans p
  ON p.workspace_id = r.workspace_id AND p.id = r.plan_id
JOIN workflow_plan_versions v
  ON v.workspace_id = r.workspace_id
 AND v.plan_id = r.plan_id
 AND v.id = r.plan_version_id`
}

func scanDeployment(scanner rowScanner) (lockedDeployment, error) {
	var value lockedDeployment
	var next sql.NullTime
	if err := scanner.Scan(
		&value.WorkspaceID, &value.BindingID, &value.BindingSourceID, &value.BindingActivation,
		&value.BindingRevisionID, &value.BindingRevision, &value.PlanID, &value.PlanVersionID,
		&value.WorkflowName, &value.RevisionDigest, &value.ConfigJSON, &value.PlanArtifactJSON,
		&value.PlanCompilerVersion, &value.PlanVersionDigest,
		&value.BindingState, &value.ActiveSourceID, &value.ActiveRevisionID,
		&value.SourceStatus, &value.SourceKind, &next,
		&value.OwnerUserID, &value.OwnerMemberID,
		&value.BindingCreatedBy.Type, &value.BindingCreatedBy.ID, &value.BindingCreatedBy.MemberID,
		&value.PlanCreatedBy.Type, &value.PlanCreatedBy.ID, &value.PlanCreatedBy.MemberID,
		&value.PlanPublishedBy.Type, &value.PlanPublishedBy.ID, &value.PlanPublishedBy.MemberID,
		&value.SourceActivatedBy.Type, &value.SourceActivatedBy.ID, &value.SourceActivatedBy.MemberID,
	); err != nil {
		return lockedDeployment{}, err
	}
	if next.Valid {
		normalized := next.Time.UTC()
		value.NextOccurrenceAt = &normalized
	}
	return value, nil
}

func loadDeployment(ctx context.Context, queryer coredb.QueryRower, workspaceID string, bindingID string, sourceID string, forUpdate bool) (lockedDeployment, error) {
	if forUpdate {
		return lockDeploymentInOrder(ctx, queryer, workspaceID, bindingID, sourceID)
	}
	query := deploymentSelectSQL() + `
WHERE b.workspace_id = $1 AND b.id = $2`
	args := []any{workspaceID, bindingID}
	if strings.TrimSpace(sourceID) != "" {
		query += " AND s.id = $3"
		args = append(args, sourceID)
	} else {
		query += " AND s.id = b.active_source_id"
	}
	value, err := scanDeployment(queryer.QueryRow(ctx, query, args...))
	if err != nil {
		if coredb.IsNotFound(err) {
			var bindingExists int
			if checkErr := queryer.QueryRow(ctx, `
SELECT COUNT(*) FROM workflow_bindings WHERE workspace_id = $1 AND id = $2
`, workspaceID, bindingID).Scan(&bindingExists); checkErr != nil {
				return lockedDeployment{}, fmt.Errorf("check workflow invocation Binding existence: %w", checkErr)
			}
			if bindingExists == 0 {
				return lockedDeployment{}, ErrNotFound
			}
			var sourceRevisionExists int
			checkQuery := `
SELECT COUNT(*)
FROM workflow_bindings b
JOIN workflow_binding_sources s
  ON s.workspace_id = b.workspace_id AND s.binding_id = b.id
JOIN workflow_binding_revisions r
  ON r.workspace_id = s.workspace_id AND r.binding_id = s.binding_id AND r.id = s.binding_revision_id
WHERE b.workspace_id = $1 AND b.id = $2`
			checkArgs := []any{workspaceID, bindingID}
			if strings.TrimSpace(sourceID) != "" {
				checkQuery += " AND s.id = $3"
				checkArgs = append(checkArgs, sourceID)
			} else {
				checkQuery += " AND s.id = b.active_source_id"
			}
			if checkErr := queryer.QueryRow(ctx, checkQuery, checkArgs...).Scan(&sourceRevisionExists); checkErr != nil {
				return lockedDeployment{}, fmt.Errorf("check workflow invocation source/revision existence: %w", checkErr)
			}
			if sourceRevisionExists == 0 {
				return lockedDeployment{}, ErrInactiveDeployment
			}
			return lockedDeployment{}, ErrPlanVersionUnavailable
		}
		return lockedDeployment{}, fmt.Errorf("load workflow invocation deployment: %w", err)
	}
	return value, nil
}

// lockDeploymentInOrder resolves the exact immutable deployment with one
// point locking read per aggregate. A multi-table SELECT ... FOR UPDATE does
// not promise lock acquisition in SQL text order under TiDB's optimizer and
// can invert Binding lifecycle's Binding->Source/PlanVersion order. Keep this
// sequence aligned with schedule materialization and Binding mutations:
// Binding -> Source -> BindingRevision -> Plan -> PlanVersion.
func lockDeploymentInOrder(
	ctx context.Context,
	queryer coredb.QueryRower,
	workspaceID string,
	bindingID string,
	sourceID string,
) (lockedDeployment, error) {
	var value lockedDeployment
	var bindingPlanID string
	if err := queryer.QueryRow(ctx, `
SELECT workspace_id, id, plan_id, state,
       COALESCE(active_source_id, ''), COALESCE(active_revision_id, ''),
       owner_user_id, owner_member_id,
       created_by_actor_type, created_by_actor_id, created_by_member_id
FROM workflow_bindings
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, workspaceID, bindingID).Scan(
		&value.WorkspaceID, &value.BindingID, &bindingPlanID, &value.BindingState,
		&value.ActiveSourceID, &value.ActiveRevisionID,
		&value.OwnerUserID, &value.OwnerMemberID,
		&value.BindingCreatedBy.Type, &value.BindingCreatedBy.ID, &value.BindingCreatedBy.MemberID,
	); err != nil {
		if coredb.IsNotFound(err) {
			return lockedDeployment{}, ErrNotFound
		}
		return lockedDeployment{}, fmt.Errorf("lock workflow invocation Binding: %w", err)
	}

	resolvedSourceID := strings.TrimSpace(sourceID)
	if resolvedSourceID == "" {
		resolvedSourceID = strings.TrimSpace(value.ActiveSourceID)
	}
	if resolvedSourceID == "" {
		return lockedDeployment{}, ErrInactiveDeployment
	}
	var next sql.NullTime
	if err := queryer.QueryRow(ctx, `
SELECT id, activation_number, binding_revision_id, status, source_kind,
       next_occurrence_at,
       activated_by_actor_type, activated_by_actor_id, activated_by_member_id
FROM workflow_binding_sources
WHERE workspace_id = $1 AND binding_id = $2 AND id = $3
FOR UPDATE
`, workspaceID, bindingID, resolvedSourceID).Scan(
		&value.BindingSourceID, &value.BindingActivation, &value.BindingRevisionID,
		&value.SourceStatus, &value.SourceKind, &next,
		&value.SourceActivatedBy.Type, &value.SourceActivatedBy.ID, &value.SourceActivatedBy.MemberID,
	); err != nil {
		if coredb.IsNotFound(err) {
			return lockedDeployment{}, ErrInactiveDeployment
		}
		return lockedDeployment{}, fmt.Errorf("lock workflow invocation Binding source: %w", err)
	}
	if next.Valid {
		occurrence := next.Time.UTC().Truncate(time.Microsecond)
		value.NextOccurrenceAt = &occurrence
	}

	if err := queryer.QueryRow(ctx, `
SELECT id, revision_number, plan_id, plan_version_id, digest, config_json
FROM workflow_binding_revisions
WHERE workspace_id = $1 AND binding_id = $2 AND id = $3
FOR UPDATE
`, workspaceID, bindingID, value.BindingRevisionID).Scan(
		&value.BindingRevisionID, &value.BindingRevision, &value.PlanID,
		&value.PlanVersionID, &value.RevisionDigest, &value.ConfigJSON,
	); err != nil {
		if coredb.IsNotFound(err) {
			return lockedDeployment{}, ErrInactiveDeployment
		}
		return lockedDeployment{}, fmt.Errorf("lock workflow invocation Binding revision: %w", err)
	}
	if strings.TrimSpace(bindingPlanID) != strings.TrimSpace(value.PlanID) {
		return lockedDeployment{}, ErrDeploymentChanged
	}

	if err := queryer.QueryRow(ctx, `
SELECT display_name, created_by_actor_type, created_by_actor_id, created_by_member_id
FROM workflow_plans
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, workspaceID, value.PlanID).Scan(
		&value.WorkflowName, &value.PlanCreatedBy.Type, &value.PlanCreatedBy.ID, &value.PlanCreatedBy.MemberID,
	); err != nil {
		if coredb.IsNotFound(err) {
			return lockedDeployment{}, ErrPlanVersionUnavailable
		}
		return lockedDeployment{}, fmt.Errorf("lock workflow invocation Plan: %w", err)
	}

	if err := queryer.QueryRow(ctx, `
SELECT artifact, compiler_version, digest,
       published_by_actor_type, published_by_actor_id, published_by_member_id
FROM workflow_plan_versions
WHERE workspace_id = $1 AND plan_id = $2 AND id = $3
FOR UPDATE
`, workspaceID, value.PlanID, value.PlanVersionID).Scan(
		&value.PlanArtifactJSON, &value.PlanCompilerVersion, &value.PlanVersionDigest,
		&value.PlanPublishedBy.Type, &value.PlanPublishedBy.ID, &value.PlanPublishedBy.MemberID,
	); err != nil {
		if coredb.IsNotFound(err) {
			return lockedDeployment{}, ErrPlanVersionUnavailable
		}
		return lockedDeployment{}, fmt.Errorf("lock workflow invocation PlanVersion: %w", err)
	}
	return value, nil
}

func (store *Store) LoadManualPlanVersion(ctx context.Context, workspaceID string, planID string, requestedPlanVersionID string) (ManualPlanVersion, error) {
	if err := store.Validate(); err != nil {
		return ManualPlanVersion{}, err
	}
	workspaceID, planID = strings.TrimSpace(workspaceID), strings.TrimSpace(planID)
	requestedPlanVersionID = strings.TrimSpace(requestedPlanVersionID)
	if workspaceID == "" || planID == "" {
		return ManualPlanVersion{}, storeerr.BadRequest("workspaceId and planId are required")
	}
	var value ManualPlanVersion
	var defaultVersionID sql.NullString
	var archivedAt sql.NullTime
	err := store.db.QueryRow(ctx, `
SELECT id, display_name, default_version_id, archived_at
FROM workflow_plans
WHERE workspace_id = $1 AND id = $2
`, workspaceID, planID).Scan(&value.PlanID, &value.WorkflowName, &defaultVersionID, &archivedAt)
	if err != nil {
		if coredb.IsNotFound(err) {
			return ManualPlanVersion{}, ErrNotFound
		}
		return ManualPlanVersion{}, fmt.Errorf("load direct workflow Plan: %w", err)
	}
	if archivedAt.Valid {
		return ManualPlanVersion{}, ErrInactiveDeployment
	}
	selectedVersionID := requestedPlanVersionID
	if selectedVersionID == "" {
		if !defaultVersionID.Valid || strings.TrimSpace(defaultVersionID.String) == "" {
			return ManualPlanVersion{}, ErrPlanVersionUnavailable
		}
		selectedVersionID = strings.TrimSpace(defaultVersionID.String)
	}
	value.WorkspaceID = workspaceID
	value.RequestedPlanVersionID = requestedPlanVersionID
	err = store.db.QueryRow(ctx, `
SELECT id, artifact, compiler_version, digest
FROM workflow_plan_versions
WHERE workspace_id = $1 AND plan_id = $2 AND id = $3
`, workspaceID, planID, selectedVersionID).Scan(
		&value.PlanVersionID, &value.PlanArtifactJSON,
		&value.PlanCompilerVersion, &value.PlanVersionDigest,
	)
	if err != nil {
		if coredb.IsNotFound(err) {
			return ManualPlanVersion{}, ErrPlanVersionUnavailable
		}
		return ManualPlanVersion{}, fmt.Errorf("load direct workflow PlanVersion: %w", err)
	}
	return value, nil
}

type lockedManualPlanVersion struct {
	ManualPlanVersion
	PlanCreatedBy   Actor
	PlanPublishedBy Actor
}

// lockManualPlanVersion locks the mutable Plan role before the immutable
// version. If the request selected "default", the role must still point at the
// exact version whose artifact was validated by the service.
func lockManualPlanVersion(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID string,
	planID string,
	requestedPlanVersionID string,
	expectedPlanVersionID string,
	expectedDigest string,
) (lockedManualPlanVersion, error) {
	var value lockedManualPlanVersion
	var defaultVersionID sql.NullString
	var archivedAt sql.NullTime
	if err := tx.QueryRow(ctx, `
SELECT id, display_name, default_version_id, archived_at,
       created_by_actor_type, created_by_actor_id, created_by_member_id
FROM workflow_plans
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, workspaceID, planID).Scan(
		&value.PlanID, &value.WorkflowName, &defaultVersionID, &archivedAt,
		&value.PlanCreatedBy.Type, &value.PlanCreatedBy.ID, &value.PlanCreatedBy.MemberID,
	); err != nil {
		if coredb.IsNotFound(err) {
			return lockedManualPlanVersion{}, ErrNotFound
		}
		return lockedManualPlanVersion{}, fmt.Errorf("lock direct workflow Plan: %w", err)
	}
	if archivedAt.Valid {
		return lockedManualPlanVersion{}, ErrInactiveDeployment
	}
	selectedVersionID := strings.TrimSpace(requestedPlanVersionID)
	if selectedVersionID == "" {
		if !defaultVersionID.Valid || strings.TrimSpace(defaultVersionID.String) == "" {
			return lockedManualPlanVersion{}, ErrPlanVersionUnavailable
		}
		selectedVersionID = strings.TrimSpace(defaultVersionID.String)
	}
	if selectedVersionID != strings.TrimSpace(expectedPlanVersionID) {
		return lockedManualPlanVersion{}, ErrDeploymentChanged
	}
	value.WorkspaceID = workspaceID
	value.RequestedPlanVersionID = strings.TrimSpace(requestedPlanVersionID)
	if err := tx.QueryRow(ctx, `
SELECT id, artifact, compiler_version, digest,
       published_by_actor_type, published_by_actor_id, published_by_member_id
FROM workflow_plan_versions
WHERE workspace_id = $1 AND plan_id = $2 AND id = $3
FOR UPDATE
`, workspaceID, planID, selectedVersionID).Scan(
		&value.PlanVersionID, &value.PlanArtifactJSON,
		&value.PlanCompilerVersion, &value.PlanVersionDigest,
		&value.PlanPublishedBy.Type, &value.PlanPublishedBy.ID, &value.PlanPublishedBy.MemberID,
	); err != nil {
		if coredb.IsNotFound(err) {
			return lockedManualPlanVersion{}, ErrPlanVersionUnavailable
		}
		return lockedManualPlanVersion{}, fmt.Errorf("lock direct workflow PlanVersion: %w", err)
	}
	if strings.TrimSpace(value.PlanVersionDigest) == "" || value.PlanVersionDigest != strings.TrimSpace(expectedDigest) {
		return lockedManualPlanVersion{}, ErrDeploymentChanged
	}
	return value, nil
}