package run

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

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

// FailClaimedRun atomically records a deterministic reducer failure without
// orphaning an external write that has already crossed the dispatch boundary.
// Work that has not crossed the dispatch boundary is canceled immediately. A
// dispatching Effect, or a pending Effect whose attempt count proves a writer
// already crossed that boundary and scheduled an idempotent retry, is left
// fenced. The aggregate remains waiting until every such receipt settles; the
// next turn calls this method again and completes the already-persisted logical
// failure.
func (store *Store) FailClaimedRun(ctx context.Context, params FailClaimedRunParams) (FailClaimedRunResult, error) {
	if err := store.Validate(); err != nil {
		return FailClaimedRunResult{}, err
	}
	params.Fence.WorkspaceID = strings.TrimSpace(params.Fence.WorkspaceID)
	params.Fence.RunID = strings.TrimSpace(params.Fence.RunID)
	params.Fence.LeaseToken = strings.TrimSpace(params.Fence.LeaseToken)
	params.ErrorCode = strings.TrimSpace(params.ErrorCode)
	if params.Fence.WorkspaceID == "" || params.Fence.RunID == "" || params.Fence.LeaseToken == "" ||
		params.ExpectedRevision <= 0 || params.ErrorCode == "" || len(params.ErrorCode) > 128 {
		return FailClaimedRunResult{}, storeerr.BadRequest("workflow Run failure requires a complete fence, expectedRevision, and bounded error code")
	}
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = utc(params.Now)
	var err error
	params.StateJSON, err = deterministicRunState(params.StateJSON)
	if err != nil {
		return FailClaimedRunResult{}, storeerr.BadRequest(err.Error())
	}
	params.ErrorJSON, err = deterministicObject("failed workflow Run error", params.ErrorJSON, true)
	if err != nil {
		return FailClaimedRunResult{}, storeerr.BadRequest(err.Error())
	}

	var result FailClaimedRunResult
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		// Effect claims take the placement fence before mutating an Effect.
		// Match that order so a new dispatch cannot slip between the child scan
		// and aggregate failure projection.
		var lockedWorkspaceID string
		if err := tx.QueryRow(ctx, `
SELECT workspace_id
FROM workspace_placements
WHERE workspace_id=$1
FOR UPDATE`, params.Fence.WorkspaceID).Scan(&lockedWorkspaceID); err != nil {
			return mapReadError("lock workflow Run workspace placement for failure", err)
		}
		txStore := store.InTx(tx)
		run, err := txStore.LoadRunForUpdate(ctx, params.Fence, params.Now)
		if err != nil {
			return err
		}
		if run.Revision != params.ExpectedRevision {
			return storeerr.Conflict("workflow Run revision changed")
		}

		// The first deterministic failure is authoritative throughout a drain.
		// Re-entering after an Effect receipt must not rewrite its state/cause or
		// append duplicate run_failed events.
		appendFailureEvent := run.ErrorCode == ""
		if !appendFailureEvent {
			params.StateJSON = run.StateJSON
			params.ErrorCode = run.ErrorCode
			params.ErrorJSON = run.ErrorJSON
		}

		rows, err := tx.Query(ctx, `
SELECT id,status,retry_at,lease_expires_at
FROM workflow_run_effects
WHERE workspace_id=$1 AND run_id=$2
  AND (status='dispatching' OR (status='pending' AND attempt_count>0))
ORDER BY id
FOR UPDATE`, run.WorkspaceID, run.ID)
		if err != nil {
			return fmt.Errorf("lock crossed-dispatch workflow Run effects for failure: %w", err)
		}
		drainBarriers := 0
		var earliest time.Time
		for rows.Next() {
			var effectID, status string
			var retryAt sql.NullTime
			var leaseExpires sql.NullTime
			if err := rows.Scan(&effectID, &status, &retryAt, &leaseExpires); err != nil {
				rows.Close()
				return fmt.Errorf("scan crossed-dispatch workflow Run effect for failure: %w", err)
			}
			drainBarriers++
			var wakeAt sql.NullTime
			switch status {
			case EffectStatusDispatching:
				wakeAt = leaseExpires
			case EffectStatusPending:
				wakeAt = retryAt
			}
			if wakeAt.Valid && (earliest.IsZero() || wakeAt.Time.Before(earliest)) {
				earliest = utc(wakeAt.Time)
			}
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return fmt.Errorf("iterate crossed-dispatch workflow Run effects for failure: %w", err)
		}

		if _, err := tx.Exec(ctx, `
UPDATE workflow_run_activations
SET status='canceled',completed_at=$3,retry_at=NULL,wake_at=NULL,
    error_code=$4,error_json=$5,updated_at=$3
WHERE workspace_id=$1 AND run_id=$2 AND status IN ('ready','waiting')`,
			run.WorkspaceID, run.ID, params.Now, params.ErrorCode, params.ErrorJSON); err != nil {
			return fmt.Errorf("cancel workflow Run activations after deterministic failure: %w", err)
		}
		// queued_for_agent callbacks are cancelable and awaiting callbacks already
		// hold an execution lease; terminal them and release every lease here.
		if err := cancelCallbackExecutionLeases(ctx, tx, run.WorkspaceID, run.ID, params.ErrorCode, params.ErrorJSON, params.Now); err != nil {
			return err
		}

		lastSeq := run.LastEventSeq
		var appended *Event
		if appendFailureEvent {
			events, nextSeq, err := txStore.insertEvents(ctx, run, []EventDraft{{Type: "run_failed", PayloadJSON: params.ErrorJSON}}, params.Now)
			if err != nil {
				return err
			}
			lastSeq = nextSeq
			appended = &events[0]
		}

		status := RunStatusFailed
		completedAt := nullableTime(&params.Now)
		runnableAt := params.Now
		if drainBarriers != 0 {
			status = RunStatusWaiting
			completedAt = nil
			if !earliest.IsZero() && earliest.After(runnableAt) {
				runnableAt = earliest
			}
		}
		update, err := tx.Exec(ctx, `
UPDATE workflow_run_instances
SET status=$4,state_json=$5,output_json=NULL,error_code=$6,error_json=$7,
    completed_at=$8,canceled_at=NULL,runnable_at=$9,
    lease_owner=NULL,lease_token=NULL,lease_expires_at=NULL,
    last_event_seq=$10,revision=revision+1,updated_at=$11
WHERE workspace_id=$1 AND id=$2 AND lease_token=$3 AND revision=$12`,
			run.WorkspaceID, run.ID, params.Fence.LeaseToken, status, params.StateJSON,
			params.ErrorCode, params.ErrorJSON, completedAt, runnableAt, lastSeq, params.Now, params.ExpectedRevision)
		if err != nil {
			return fmt.Errorf("fail claimed workflow Run: %w", err)
		}
		if update.RowsAffected() == 0 {
			return storeerr.Conflict("workflow Run failure turn changed")
		}
		failed, err := scanRun(tx.QueryRow(ctx, runSelectSQL()+` WHERE r.workspace_id=$1 AND r.id=$2`, run.WorkspaceID, run.ID))
		if err != nil {
			return fmt.Errorf("load failed workflow Run: %w", err)
		}
		result = FailClaimedRunResult{Run: failed, Event: appended, Draining: drainBarriers != 0}
		return nil
	})
	return result, err
}