package run

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	invocationlang "github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runartifact"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runlimit"
)

type rowScanner interface{ Scan(...any) error }

func deterministicObject(label string, value json.RawMessage, required bool) (json.RawMessage, error) {
	return DeterministicPayloadJSON(label, value, required)
}

func deterministicRunState(value json.RawMessage) (json.RawMessage, error) {
	return DeterministicStateJSON(value)
}

// DeterministicPayloadJSON is the exact pure validation boundary used by the Run
// Store for each input/output/effect/event JSON object. Runtime projection uses
// the same helper before SaveTurn so deterministic resource failures never
// become an endlessly retried Store BadRequest.
func DeterministicPayloadJSON(label string, value json.RawMessage, required bool) (json.RawMessage, error) {
	if len(value) == 0 {
		if required {
			return nil, fmt.Errorf("%s is required", label)
		}
		return nil, nil
	}
	canonical, err := invocationlang.DeterministicJSONObject(label, value, maxRuntimeJSON)
	if err != nil {
		return nil, err
	}
	return canonical, nil
}

// CanonicalArtifactPayloadJSON is the additive callback/effect boundary. It
// accepts a complete deterministic JSON object up to the durable artifact
// ceiling; the transaction store externalizes any value above MaxRunPayloadBytes
// before writing activation/effect rows. Ordinary Run payloads continue using
// the 4 MiB deterministic payload boundary.
func CanonicalArtifactPayloadJSON(label string, value json.RawMessage, required bool) (json.RawMessage, error) {
	if len(value) > runartifact.MaxArtifactBytes {
		return nil, fmt.Errorf("%s exceeds the durable artifact budget", label)
	}
	return canonicalPayloadWithBudget(label, value, required, runartifact.MaxArtifactBytes)
}

func canonicalPayloadWithBudget(label string, value json.RawMessage, required bool, budget int) (json.RawMessage, error) {
	if len(value) == 0 {
		if required {
			return nil, fmt.Errorf("%s is required", label)
		}
		return nil, nil
	}
	canonical, err := invocationlang.DeterministicJSONObjectWithBudget(label, value, budget, runlimit.MaxRunPayloadJSONNesting, runlimit.MaxRunPayloadJSONValues)
	if err != nil {
		return nil, err
	}
	return canonical, nil
}

// DeterministicStateJSON validates the aggregate reducer snapshot with its own
// structural budget. Reusing the ordinary 8192-value input budget here would
// reject a well-bounded Run long before MaxRunActivations or the 32 MiB state
// limit is reached.
func DeterministicStateJSON(value json.RawMessage) (json.RawMessage, error) {
	if len(value) == 0 {
		return nil, fmt.Errorf("workflow Run state is required")
	}
	return invocationlang.DeterministicJSONObjectWithBudget(
		"workflow Run state",
		value,
		MaxRunStateBytes,
		MaxRunStateJSONNesting,
		MaxRunStateJSONValues,
	)
}

func utc(value time.Time) time.Time { return value.UTC().Truncate(time.Microsecond) }

func utcPtr(value *time.Time) *time.Time {
	if value == nil {
		return nil
	}
	normalized := utc(*value)
	return &normalized
}

func nullableString(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return strings.TrimSpace(value)
}

func nullableJSON(value json.RawMessage) any {
	if len(value) == 0 {
		return nil
	}
	return []byte(value)
}

func nullableTime(value *time.Time) any {
	if value == nil {
		return nil
	}
	return utc(*value)
}

func nullTimePtr(value sql.NullTime) *time.Time {
	if !value.Valid {
		return nil
	}
	normalized := utc(value.Time)
	return &normalized
}

func runSelectSQL() string {
	return `
SELECT r.id, r.workspace_id, r.invocation_id,
       r.plan_id, r.plan_version_id, r.team_id, r.workflow_name_snapshot,
       COALESCE(p.plan_key, ''), COALESCE(p.display_name, ''),
       COALESCE(i.binding_id, ''), COALESCE(b.binding_key, ''), COALESCE(b.display_name, ''),
       i.source_kind, COALESCE(i.invoked_by_actor_type, ''), COALESCE(i.invoked_by_actor_id, ''), COALESCE(i.invoked_by_member_id, ''),
       r.status, r.input_json, r.input_digest,
       r.state_json, r.output_json, r.error_code, r.error_json,
       r.owner_user_id_snapshot, r.owner_member_id_snapshot,
       r.principal_member_id_snapshot, r.author_member_id_snapshot,
       r.queued_at, r.started_at, r.completed_at, r.canceled_at,
       r.revision, r.last_event_seq, r.runnable_at,
       r.lease_owner, r.lease_token, r.lease_expires_at,
       r.created_at, r.updated_at
FROM workflow_run_instances r
LEFT JOIN workflow_plans p ON p.id = r.plan_id
LEFT JOIN workflow_invocations i ON i.id = r.invocation_id
LEFT JOIN workflow_bindings b ON b.id = i.binding_id`
}

func scanRun(scanner rowScanner) (Run, error) {
	var value Run
	var output, errorJSON []byte
	var errorCode, leaseOwner, leaseToken sql.NullString
	var teamID sql.NullString
	var sourceKind, invokedByType, invokedByID, invokedByMemberID sql.NullString
	var started, completed, canceled, leaseExpires sql.NullTime
	if err := scanner.Scan(
		&value.ID, &value.WorkspaceID, &value.InvocationID,
		&value.PlanID, &value.PlanVersionID, &teamID, &value.WorkflowNameSnapshot,
		&value.PlanKey, &value.PlanDisplayName,
		&value.BindingID, &value.BindingKey, &value.BindingDisplayName,
		&sourceKind, &invokedByType, &invokedByID, &invokedByMemberID,
		&value.Status, &value.InputJSON, &value.InputDigest,
		&value.StateJSON, &output, &errorCode, &errorJSON,
		&value.Identity.OwnerUserID, &value.Identity.OwnerMemberID,
		&value.Identity.PrincipalMemberID, &value.Identity.AuthorMemberID,
		&value.QueuedAt, &started, &completed, &canceled,
		&value.Revision, &value.LastEventSeq, &value.RunnableAt,
		&leaseOwner, &leaseToken, &leaseExpires,
		&value.CreatedAt, &value.UpdatedAt,
	); err != nil {
		return Run{}, err
	}
	value.TeamID = teamID.String
	value.SourceKind = sourceKind.String
	value.InvokedBy = invocationlang.Actor{Type: invokedByType.String, ID: invokedByID.String, MemberID: invokedByMemberID.String}
	input, err := deterministicObject("persisted workflow Run input", value.InputJSON, true)
	if err != nil {
		return Run{}, fmt.Errorf("workflow Run %s has invalid input: %w", value.ID, err)
	}
	state, err := deterministicRunState(value.StateJSON)
	if err != nil {
		return Run{}, fmt.Errorf("workflow Run %s has invalid state: %w", value.ID, err)
	}
	value.OutputJSON, err = deterministicObject("persisted workflow Run output", output, false)
	if err != nil {
		return Run{}, fmt.Errorf("workflow Run %s has invalid output: %w", value.ID, err)
	}
	value.ErrorJSON, err = deterministicObject("persisted workflow Run error", errorJSON, false)
	if err != nil {
		return Run{}, fmt.Errorf("workflow Run %s has invalid error: %w", value.ID, err)
	}
	value.InputJSON, value.StateJSON = input, state
	value.WorkflowNameSnapshot = strings.TrimSpace(value.WorkflowNameSnapshot)
	if value.WorkflowNameSnapshot == "" || len(value.WorkflowNameSnapshot) > 255 {
		return Run{}, fmt.Errorf("workflow Run %s has invalid workflow name snapshot", value.ID)
	}
	value.PlanKey = strings.TrimSpace(value.PlanKey)
	value.PlanDisplayName = strings.TrimSpace(value.PlanDisplayName)
	value.BindingID = strings.TrimSpace(value.BindingID)
	value.BindingKey = strings.TrimSpace(value.BindingKey)
	value.BindingDisplayName = strings.TrimSpace(value.BindingDisplayName)
	value.ErrorCode = errorCode.String
	value.LeaseOwner, value.LeaseToken = leaseOwner.String, leaseToken.String
	value.StartedAt, value.CompletedAt, value.CanceledAt = nullTimePtr(started), nullTimePtr(completed), nullTimePtr(canceled)
	value.LeaseExpires = nullTimePtr(leaseExpires)
	value.QueuedAt, value.RunnableAt = utc(value.QueuedAt), utc(value.RunnableAt)
	value.CreatedAt, value.UpdatedAt = utc(value.CreatedAt), utc(value.UpdatedAt)
	return value, nil
}

func activationSelectSQL() string {
	return `
SELECT 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
FROM workflow_run_activations`
}

func scanActivation(scanner rowScanner) (Activation, error) {
	var value Activation
	var parent, selected, errorCode sql.NullString
	var output, errorJSON []byte
	var retry, wake, deadline, started, completed sql.NullTime
	if err := scanner.Scan(
		&value.ID, &value.WorkspaceID, &value.RunID, &value.LogicalKey, &parent,
		&value.NodeID, &value.Opcode, &value.Status, &value.InputJSON, &output, &value.ScopeContextJSON,
		&selected, &value.AttemptCount, &retry, &wake, &deadline,
		&errorCode, &errorJSON, &started, &completed, &value.CreatedAt, &value.UpdatedAt,
	); err != nil {
		return Activation{}, err
	}
	var err error
	value.InputJSON, err = deterministicObject("persisted workflow Run activation input", value.InputJSON, true)
	if err != nil {
		return Activation{}, fmt.Errorf("workflow Run activation %s has invalid input: %w", value.ID, err)
	}
	value.ScopeContextJSON, err = deterministicObject("persisted workflow Run activation scope context", value.ScopeContextJSON, true)
	if err != nil {
		return Activation{}, fmt.Errorf("workflow Run activation %s has invalid scope context: %w", value.ID, err)
	}
	value.OutputJSON, err = deterministicObject("persisted workflow Run activation output", output, false)
	if err != nil {
		return Activation{}, fmt.Errorf("workflow Run activation %s has invalid output: %w", value.ID, err)
	}
	value.ErrorJSON, err = deterministicObject("persisted workflow Run activation error", errorJSON, false)
	if err != nil {
		return Activation{}, fmt.Errorf("workflow Run activation %s has invalid error: %w", value.ID, err)
	}
	value.ParentActivationID, value.SelectedPort, value.ErrorCode = parent.String, selected.String, errorCode.String
	value.RetryAt, value.WakeAt, value.DeadlineAt = nullTimePtr(retry), nullTimePtr(wake), nullTimePtr(deadline)
	value.StartedAt, value.CompletedAt = nullTimePtr(started), nullTimePtr(completed)
	value.CreatedAt, value.UpdatedAt = utc(value.CreatedAt), utc(value.UpdatedAt)
	return value, nil
}

func effectSelectSQL() string {
	return `
SELECT id, workspace_id, run_id, activation_id, approval_assignee_member_id, message_id, effect_kind,
       callback_responder_agent_id, 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, queued_at,
       recovery_attempt_count, first_non_working_at, next_recovery_at, recovery_paused_at,
       recovery_epoch, recovery_cycle_attempt_count, recovery_cycle_first_non_working_at,
       recovery_cycle_next_recovery_at, park_state, park_armed_at, park_parked_at, park_cleared_at,
       created_at, updated_at
FROM workflow_run_effects`
}

func scanEffect(scanner rowScanner) (Effect, error) {
	var value Effect
	var result, errorJSON []byte
	var approvalAssignee, callbackResponder, messageID, leaseOwner, leaseToken, errorCode sql.NullString
	var retry, deadline, leaseExpires, delivered, completed, queued sql.NullTime
	var firstNonWorking, nextRecovery, recoveryPaused sql.NullTime
	var cycleFirstNonWorking, cycleNextRecovery sql.NullTime
	var parkArmed, parkParked, parkCleared sql.NullTime
	if err := scanner.Scan(
		&value.ID, &value.WorkspaceID, &value.RunID, &value.ActivationID, &approvalAssignee, &messageID, &value.Kind,
		&callbackResponder,
		&value.IdempotencyKey, &value.Status, &value.RequestJSON, &result, &value.AttemptCount,
		&retry, &deadline, &leaseOwner, &leaseToken, &leaseExpires,
		&errorCode, &errorJSON, &delivered, &completed, &queued,
		&value.RecoveryAttemptCount, &firstNonWorking, &nextRecovery, &recoveryPaused,
		&value.RecoveryEpoch, &value.RecoveryCycleAttemptCount, &cycleFirstNonWorking, &cycleNextRecovery,
		&value.ParkState, &parkArmed, &parkParked, &parkCleared,
		&value.CreatedAt, &value.UpdatedAt,
	); err != nil {
		return Effect{}, err
	}
	var err error
	value.RequestJSON, err = deterministicObject("persisted workflow Run effect request", value.RequestJSON, true)
	if err != nil {
		return Effect{}, fmt.Errorf("workflow Run effect %s has invalid request: %w", value.ID, err)
	}
	value.ResultJSON, err = deterministicObject("persisted workflow Run effect result", result, false)
	if err != nil {
		return Effect{}, fmt.Errorf("workflow Run effect %s has invalid result: %w", value.ID, err)
	}
	value.ErrorJSON, err = deterministicObject("persisted workflow Run effect error", errorJSON, false)
	if err != nil {
		return Effect{}, fmt.Errorf("workflow Run effect %s has invalid error: %w", value.ID, err)
	}
	value.ApprovalAssigneeMemberID = approvalAssignee.String
	value.CallbackResponderAgentID = callbackResponder.String
	value.MessageID = messageID.String
	value.LeaseOwner, value.LeaseToken, value.ErrorCode = leaseOwner.String, leaseToken.String, errorCode.String
	value.RetryAt, value.DeadlineAt, value.LeaseExpiresAt = nullTimePtr(retry), nullTimePtr(deadline), nullTimePtr(leaseExpires)
	value.DeliveredAt, value.CompletedAt, value.QueuedAt = nullTimePtr(delivered), nullTimePtr(completed), nullTimePtr(queued)
	value.FirstNonWorkingAt, value.NextRecoveryAt, value.RecoveryPausedAt =
		nullTimePtr(firstNonWorking), nullTimePtr(nextRecovery), nullTimePtr(recoveryPaused)
	value.RecoveryCycleFirstNonWorkingAt, value.RecoveryCycleNextRecoveryAt =
		nullTimePtr(cycleFirstNonWorking), nullTimePtr(cycleNextRecovery)
	value.ParkArmedAt, value.ParkParkedAt, value.ParkClearedAt = nullTimePtr(parkArmed), nullTimePtr(parkParked), nullTimePtr(parkCleared)
	value.CreatedAt, value.UpdatedAt = utc(value.CreatedAt), utc(value.UpdatedAt)
	return value, nil
}

func eventSelectSQL() string {
	return `
SELECT id, workspace_id, run_id, seq, event_type, payload_json, created_at
FROM workflow_run_instance_events`
}

func scanEvent(scanner rowScanner) (Event, error) {
	var value Event
	if err := scanner.Scan(&value.ID, &value.WorkspaceID, &value.RunID, &value.Seq, &value.Type, &value.PayloadJSON, &value.CreatedAt); err != nil {
		return Event{}, err
	}
	payload, err := deterministicObject("persisted workflow Run event payload", value.PayloadJSON, true)
	if err != nil {
		return Event{}, fmt.Errorf("workflow Run event %s has invalid payload: %w", value.ID, err)
	}
	value.PayloadJSON, value.CreatedAt = payload, utc(value.CreatedAt)
	return value, nil
}

func mapReadError(message string, err error) error {
	if coredb.IsNotFound(err) {
		return ErrNotFound
	}
	if err != nil {
		return fmt.Errorf("%s: %w", message, err)
	}
	return nil
}