package run

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

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

func (store *Store) SaveTurn(ctx context.Context, params SaveTurnParams) (SaveTurnResult, error) {
	if err := store.Validate(); err != nil {
		return SaveTurnResult{}, err
	}
	var result SaveTurnResult
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		result, err = store.InTx(tx).SaveTurn(ctx, params)
		return err
	})
	return result, err
}

// SaveTurn atomically persists one complete reducer turn. It validates the
// lease and semantic revision, applies child mutations, appends events using
// the Run-owned cursor, and finally CAS-updates the aggregate snapshot.
func (store *TxStore) SaveTurn(ctx context.Context, params SaveTurnParams) (SaveTurnResult, error) {
	if err := store.Validate(); err != nil {
		return SaveTurnResult{}, err
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)
	run, err := store.LoadRunForUpdate(ctx, params.Fence, params.Now)
	if err != nil {
		return SaveTurnResult{}, err
	}
	if params.ExpectedRevision <= 0 || run.Revision != params.ExpectedRevision {
		return SaveTurnResult{}, storeerr.Conflict("workflow Run revision changed")
	}
	params.Status = strings.TrimSpace(params.Status)
	params.ErrorCode = strings.TrimSpace(params.ErrorCode)
	if !validRunStatus(params.Status) || len(params.ErrorCode) > 128 || params.RunnableAt.IsZero() {
		return SaveTurnResult{}, storeerr.BadRequest("valid workflow Run status, runnableAt, and bounded error code are required")
	}
	params.RunnableAt = utc(params.RunnableAt)
	params.StateJSON, err = deterministicRunState(params.StateJSON)
	if err != nil {
		return SaveTurnResult{}, storeerr.BadRequest(err.Error())
	}
	params.OutputJSON, err = deterministicObject("workflow Run output", params.OutputJSON, false)
	if err != nil {
		return SaveTurnResult{}, storeerr.BadRequest(err.Error())
	}
	params.ErrorJSON, err = deterministicObject("workflow Run error", params.ErrorJSON, false)
	if err != nil {
		return SaveTurnResult{}, storeerr.BadRequest(err.Error())
	}
	if err := store.persistArtifacts(ctx, params.Artifacts, params.Now, run); err != nil {
		return SaveTurnResult{}, err
	}
	params.StartedAt, params.CompletedAt, params.CanceledAt = utcPtr(params.StartedAt), utcPtr(params.CompletedAt), utcPtr(params.CanceledAt)
	if params.Status == RunStatusSucceeded || params.Status == RunStatusFailed {
		if params.CompletedAt == nil || params.CanceledAt != nil {
			return SaveTurnResult{}, storeerr.BadRequest("terminal workflow Run must have completedAt and no canceledAt")
		}
		params.ReleaseLease = true
	}
	if params.Status == RunStatusCanceled {
		if params.CompletedAt == nil || params.CanceledAt == nil {
			return SaveTurnResult{}, storeerr.BadRequest("canceled workflow Run requires completedAt and canceledAt")
		}
		params.ReleaseLease = true
	}
	for _, mutation := range params.Activations {
		if err := store.applyActivationMutation(ctx, run, mutation, params.Now); err != nil {
			return SaveTurnResult{}, err
		}
	}
	for _, mutation := range params.Effects {
		if err := store.applyEffectMutation(ctx, run, mutation, params.Now); err != nil {
			return SaveTurnResult{}, err
		}
	}
	if params.Status == RunStatusSucceeded || params.Status == RunStatusFailed || params.Status == RunStatusCanceled {
		if err := store.ensureNoLiveChildrenForTerminalTurn(ctx, run); err != nil {
			return SaveTurnResult{}, err
		}
	}
	events, lastSeq, err := store.insertEvents(ctx, run, params.Events, params.Now)
	if err != nil {
		return SaveTurnResult{}, err
	}
	leaseOwner, leaseToken, leaseExpires := nullableString(run.LeaseOwner), nullableString(run.LeaseToken), nullableTime(run.LeaseExpires)
	if params.ReleaseLease {
		leaseOwner, leaseToken, leaseExpires = nil, nil, nil
	}
	result, err := store.tx.Exec(ctx, `
UPDATE workflow_run_instances
SET status=$4, state_json=$5, output_json=$6, error_code=$7, error_json=$8,
    started_at=$9, completed_at=$10, canceled_at=$11,
    runnable_at=$12, lease_owner=$13, lease_token=$14, lease_expires_at=$15,
    last_event_seq=$16, revision=revision+1, updated_at=$17
WHERE workspace_id=$1 AND id=$2 AND lease_token=$3 AND revision=$18`,
		params.Fence.WorkspaceID, params.Fence.RunID, params.Fence.LeaseToken,
		params.Status, params.StateJSON, nullableJSON(params.OutputJSON), nullableString(params.ErrorCode), nullableJSON(params.ErrorJSON),
		nullableTime(params.StartedAt), nullableTime(params.CompletedAt), nullableTime(params.CanceledAt),
		params.RunnableAt, leaseOwner, leaseToken, leaseExpires,
		lastSeq, params.Now, params.ExpectedRevision)
	if err != nil {
		return SaveTurnResult{}, fmt.Errorf("save workflow Run turn: %w", err)
	}
	if result.RowsAffected() == 0 {
		return SaveTurnResult{}, storeerr.Conflict("workflow Run turn changed")
	}
	saved, err := scanRun(store.tx.QueryRow(ctx, runSelectSQL()+` WHERE r.workspace_id=$1 AND r.id=$2`, params.Fence.WorkspaceID, params.Fence.RunID))
	if err != nil {
		return SaveTurnResult{}, fmt.Errorf("load saved workflow Run turn: %w", err)
	}
	return SaveTurnResult{Run: saved, Events: events}, nil
}

// ensureNoLiveChildrenForTerminalTurn is deliberately called after applying
// the turn's child mutations. A normal terminal projection closes its final
// Activation and Effect in the same transaction; an incomplete projection is
// rejected before the aggregate can claim completion. Locking reads make the
// check current under TiDB repeatable-read, and dispatching is live because its
// external write remains committed until a receipt is recorded.
func (store *TxStore) ensureNoLiveChildrenForTerminalTurn(ctx context.Context, run Run) error {
	var childID string
	err := store.tx.QueryRow(ctx, `
SELECT id
FROM workflow_run_activations
WHERE workspace_id=$1 AND run_id=$2 AND status IN ('ready','waiting')
ORDER BY id
LIMIT 1 FOR UPDATE`, run.WorkspaceID, run.ID).Scan(&childID)
	switch {
	case err == nil:
		return storeerr.Conflict("terminal workflow Run still has a live activation")
	case !coredb.IsNotFound(err):
		return fmt.Errorf("check terminal workflow Run activations: %w", err)
	}
	err = store.tx.QueryRow(ctx, `
SELECT id
FROM workflow_run_effects
WHERE workspace_id=$1 AND run_id=$2
  AND status IN ('pending','dispatching','delivered','awaiting','waiting_for_responder')
ORDER BY id
LIMIT 1 FOR UPDATE`, run.WorkspaceID, run.ID).Scan(&childID)
	switch {
	case err == nil:
		return storeerr.Conflict("terminal workflow Run still has a live effect")
	case !coredb.IsNotFound(err):
		return fmt.Errorf("check terminal workflow Run effects: %w", err)
	default:
		return nil
	}
}

func (store *TxStore) applyActivationMutation(ctx context.Context, run Run, mutation ActivationMutation, now time.Time) error {
	value := mutation.Value
	if value.WorkspaceID == "" {
		value.WorkspaceID = run.WorkspaceID
	}
	if value.RunID == "" {
		value.RunID = run.ID
	}
	if value.CreatedAt.IsZero() {
		value.CreatedAt = now
	}
	value.UpdatedAt = now
	value, err := normalizeActivation(value, true)
	if err != nil {
		return err
	}
	if value.WorkspaceID != run.WorkspaceID || value.RunID != run.ID {
		return storeerr.BadRequest("workflow Run activation belongs to a different Run")
	}
	switch strings.TrimSpace(mutation.Operation) {
	case MutationInsert:
		_, err := store.tx.Exec(ctx, `
INSERT INTO workflow_run_activations (
 id,workspace_id,run_id,logical_key,parent_activation_id,node_id,opcode,status,
 input_json,output_json,scope_context_json,selected_port,attempt_count,
 retry_at,wake_at,deadline_at,error_code,error_json,started_at,completed_at,created_at,updated_at
) VALUES (
 $1,$2,$3,$4,$5,$6,$7,$8,
 $9,$10,$11,$12,$13,
 $14,$15,$16,$17,$18,$19,$20,$21,$22
)`, value.ID, value.WorkspaceID, value.RunID, value.LogicalKey, nullableString(value.ParentActivationID), value.NodeID, value.Opcode, value.Status,
			value.InputJSON, nullableJSON(value.OutputJSON), value.ScopeContextJSON, nullableString(value.SelectedPort), value.AttemptCount,
			nullableTime(value.RetryAt), nullableTime(value.WakeAt), nullableTime(value.DeadlineAt), nullableString(value.ErrorCode), nullableJSON(value.ErrorJSON),
			nullableTime(value.StartedAt), nullableTime(value.CompletedAt), value.CreatedAt, value.UpdatedAt)
		if err != nil {
			if coredb.IsUniqueViolation(err) {
				return storeerr.Conflict("workflow Run activation logical key already exists")
			}
			return fmt.Errorf("insert workflow Run activation: %w", err)
		}
		return nil
	case MutationUpdate:
		expected := strings.TrimSpace(mutation.ExpectedStatus)
		if expected == "" || !validActivationStatus(expected) {
			return storeerr.BadRequest("workflow Run activation update requires expectedStatus")
		}
		result, err := store.tx.Exec(ctx, `
UPDATE workflow_run_activations
SET status=$5, input_json=$6, output_json=$7, scope_context_json=$8, selected_port=$9, attempt_count=$10,
    retry_at=$11,wake_at=$12,deadline_at=$13,error_code=$14,error_json=$15,
    started_at=$16,completed_at=$17,updated_at=$18
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND status=$4`,
			value.WorkspaceID, value.RunID, value.ID, expected, value.Status,
			value.InputJSON, nullableJSON(value.OutputJSON), value.ScopeContextJSON, nullableString(value.SelectedPort), value.AttemptCount,
			nullableTime(value.RetryAt), nullableTime(value.WakeAt), nullableTime(value.DeadlineAt), nullableString(value.ErrorCode), nullableJSON(value.ErrorJSON),
			nullableTime(value.StartedAt), nullableTime(value.CompletedAt), value.UpdatedAt)
		if err != nil {
			return fmt.Errorf("update workflow Run activation: %w", err)
		}
		if result.RowsAffected() == 0 {
			return storeerr.Conflict("workflow Run activation status changed")
		}
		return nil
	default:
		return storeerr.BadRequest("workflow Run activation mutation must be insert or update")
	}
}

func (store *TxStore) applyEffectMutation(ctx context.Context, run Run, mutation EffectMutation, now time.Time) error {
	value := mutation.Value
	if value.WorkspaceID == "" {
		value.WorkspaceID = run.WorkspaceID
	}
	if value.RunID == "" {
		value.RunID = run.ID
	}
	if value.CreatedAt.IsZero() {
		value.CreatedAt = now
	}
	value.UpdatedAt = now
	value, err := normalizeEffect(value, true)
	if err != nil {
		return err
	}
	if value.WorkspaceID != run.WorkspaceID || value.RunID != run.ID {
		return storeerr.BadRequest("workflow Run effect belongs to a different Run")
	}
	switch strings.TrimSpace(mutation.Operation) {
	case MutationInsert:
		if value.ApprovalAssigneeMemberID != "" {
			return storeerr.BadRequest("approval assignee must be frozen by the effect dispatcher")
		}
		if value.MessageID != "" {
			return storeerr.BadRequest("message binding must be frozen by the message transaction")
		}
		if value.CallbackResponderAgentID != "" {
			return storeerr.BadRequest("callback responder must be frozen by the effect dispatcher")
		}
		_, err := store.tx.Exec(ctx, `
INSERT INTO workflow_run_effects (
 id,workspace_id,run_id,activation_id,approval_assignee_member_id,callback_responder_agent_id,effect_kind,idempotency_key,status,
 request_json,result_json,attempt_count,retry_at,deadline_at,
 lease_owner,lease_token,lease_expires_at,error_code,error_json,
 delivered_at,completed_at,created_at,updated_at
) VALUES (
	$1,$2,$3,$4,NULL,NULL,$5,$6,$7,
	$8,$9,$10,$11,$12,
	$13,$14,$15,$16,$17,
	$18,$19,$20,$21
)`, value.ID, value.WorkspaceID, value.RunID, value.ActivationID, value.Kind, value.IdempotencyKey, value.Status,
			value.RequestJSON, nullableJSON(value.ResultJSON), value.AttemptCount, nullableTime(value.RetryAt), nullableTime(value.DeadlineAt),
			nullableString(value.LeaseOwner), nullableString(value.LeaseToken), nullableTime(value.LeaseExpiresAt), nullableString(value.ErrorCode), nullableJSON(value.ErrorJSON),
			nullableTime(value.DeliveredAt), nullableTime(value.CompletedAt), value.CreatedAt, value.UpdatedAt)
		if err != nil {
			if coredb.IsUniqueViolation(err) {
				return storeerr.Conflict("workflow Run effect idempotency key already exists")
			}
			return fmt.Errorf("insert workflow Run effect: %w", err)
		}
		return nil
	case MutationUpdate:
		expected := strings.TrimSpace(mutation.ExpectedStatus)
		if expected == "" || !validEffectStatus(expected) {
			return storeerr.BadRequest("workflow Run effect update requires expectedStatus")
		}
		result, err := store.tx.Exec(ctx, `
UPDATE workflow_run_effects
SET status=$5,result_json=$6,attempt_count=$7,retry_at=$8,deadline_at=$9,
    lease_owner=$10,lease_token=$11,lease_expires_at=$12,error_code=$13,error_json=$14,
    delivered_at=$15,completed_at=$16,updated_at=$17
WHERE workspace_id=$1 AND run_id=$2 AND id=$3 AND status=$4`,
			value.WorkspaceID, value.RunID, value.ID, expected, value.Status,
			nullableJSON(value.ResultJSON), value.AttemptCount, nullableTime(value.RetryAt), nullableTime(value.DeadlineAt),
			nullableString(value.LeaseOwner), nullableString(value.LeaseToken), nullableTime(value.LeaseExpiresAt), nullableString(value.ErrorCode), nullableJSON(value.ErrorJSON),
			nullableTime(value.DeliveredAt), nullableTime(value.CompletedAt), value.UpdatedAt)
		if err != nil {
			return fmt.Errorf("update workflow Run effect: %w", err)
		}
		if result.RowsAffected() == 0 {
			return storeerr.Conflict("workflow Run effect status changed")
		}
		// A reducer-projected cancel of a callback effect must release its Agent
		// execution lease and mark its FIFO queue item terminal in the same
		// transaction, otherwise a queued/awaiting cancel would leak capacity.
		if value.Status == EffectStatusCanceled && value.CallbackResponderAgentID != "" {
			if err := releaseCallbackAgentExecutionLease(ctx, store.tx, value.WorkspaceID, value.CallbackResponderAgentID, value.ID, callbackQueueStatusCanceled, now); err != nil {
				return err
			}
		}
		return nil
	default:
		return storeerr.BadRequest("workflow Run effect mutation must be insert or update")
	}
}

func (store *TxStore) insertEvents(ctx context.Context, run Run, drafts []EventDraft, now time.Time) ([]Event, int64, error) {
	sequence := run.LastEventSeq
	events := make([]Event, 0, len(drafts))
	for _, draft := range drafts {
		normalized, err := normalizeEventDraft(draft)
		if err != nil {
			return nil, 0, err
		}
		id, err := identifiers.NewEntityID("wre")
		if err != nil {
			return nil, 0, fmt.Errorf("generate workflow Run event id: %w", err)
		}
		sequence++
		event := Event{ID: id, WorkspaceID: run.WorkspaceID, RunID: run.ID, Seq: sequence, Type: normalized.Type, PayloadJSON: normalized.PayloadJSON, CreatedAt: now}
		if _, err := store.tx.Exec(ctx, `
INSERT INTO workflow_run_instance_events (id,workspace_id,run_id,seq,event_type,payload_json,created_at)
VALUES ($1,$2,$3,$4,$5,$6,$7)`, event.ID, event.WorkspaceID, event.RunID, event.Seq, event.Type, event.PayloadJSON, event.CreatedAt); err != nil {
			return nil, 0, fmt.Errorf("append workflow Run event: %w", err)
		}
		events = append(events, event)
	}
	return events, sequence, nil
}

// AppendEvents is the generic non-turn event boundary used by callbacks and
// effect receipts. It locks the Run before allocating sequences and increments
// the aggregate revision so a concurrent reducer SaveTurn is fenced.
func (store *TxStore) AppendEvents(ctx context.Context, workspaceID, runID string, drafts []EventDraft, now time.Time) ([]Event, Run, error) {
	if err := store.Validate(); err != nil {
		return nil, Run{}, err
	}
	workspaceID, runID = strings.TrimSpace(workspaceID), strings.TrimSpace(runID)
	if workspaceID == "" || runID == "" || len(drafts) == 0 {
		return nil, Run{}, storeerr.BadRequest("workspaceId, runId, and at least one event are required")
	}
	if now.IsZero() {
		now = time.Now()
	}
	now = utc(now)
	run, err := scanRun(store.tx.QueryRow(ctx, runSelectSQL()+` WHERE r.workspace_id=$1 AND r.id=$2 FOR UPDATE`, workspaceID, runID))
	if err != nil {
		return nil, Run{}, mapReadError("lock workflow Run for event append", err)
	}
	events, lastSeq, err := store.insertEvents(ctx, run, drafts, now)
	if err != nil {
		return nil, Run{}, err
	}
	result, err := store.tx.Exec(ctx, `
UPDATE workflow_run_instances
SET last_event_seq=$3,revision=revision+1,updated_at=$4
WHERE workspace_id=$1 AND id=$2 AND revision=$5`, workspaceID, runID, lastSeq, now, run.Revision)
	if err != nil {
		return nil, Run{}, fmt.Errorf("advance workflow Run event cursor: %w", err)
	}
	if result.RowsAffected() == 0 {
		return nil, Run{}, storeerr.Conflict("workflow Run changed while appending events")
	}
	updated, err := scanRun(store.tx.QueryRow(ctx, runSelectSQL()+` WHERE r.workspace_id=$1 AND r.id=$2`, workspaceID, runID))
	if err != nil {
		return nil, Run{}, fmt.Errorf("load workflow Run after event append: %w", err)
	}
	return events, updated, nil
}

func (store *Store) AppendEvents(ctx context.Context, workspaceID, runID string, drafts []EventDraft, now time.Time) ([]Event, Run, error) {
	if err := store.Validate(); err != nil {
		return nil, Run{}, err
	}
	var events []Event
	var run Run
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		events, run, err = store.InTx(tx).AppendEvents(ctx, workspaceID, runID, drafts, now)
		return err
	})
	return events, run, err
}