package orchestrator

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
)

const parkedRunDelay = 365 * 24 * time.Hour

var (
	errRuntimeStateLimit   = errors.New("workflow Run state exceeds durable size limit")
	errRuntimePayloadLimit = errors.New("workflow Run payload exceeds durable size limit")
)

// ProjectTurnInput contains only already-loaded durable state. Projection is
// deterministic and performs no database or network operation, so a reducer
// turn can be tested independently before SaveTurn commits it atomically.
type ProjectTurnInput struct {
	Run                 run.Run
	Fence               run.TurnFence
	Graph               *runkernel.Graph
	Transition          runkernel.Transition
	ExistingActivations []run.Activation
	ExistingEffects     []run.Effect
	SourceKind          string
	Now                 time.Time
	// HasMoreWork forces an immediate follow-up turn when the bounded engine
	// stopped with reducer work that is not otherwise visible to runnableAt
	// (for example, another completed receipt for a waiting activation).
	HasMoreWork bool
	// BlockReadyOnTaskWatchDispatch parks otherwise-ready sibling activations
	// until the currently projected fixed-Task callback has either committed a
	// delivery receipt or failed. The effect dispatcher wakes the Run after that
	// short receipt boundary; callback execution itself remains parallel.
	BlockReadyOnTaskWatchDispatch bool
}

// ProjectTurn converts a pure graph transition into the generic Store
// mutation boundary. Reducer-local activation ids stay inside state JSON and
// logical_key; deterministic globally unique ids back relational rows.
func ProjectTurn(input ProjectTurnInput) (run.SaveTurnParams, error) {
	if input.Transition.State == nil || input.Graph == nil || input.Graph.Artifact == nil {
		return run.SaveTurnParams{}, fmt.Errorf("workflow Run projection requires graph and transition state")
	}
	if strings.TrimSpace(input.Run.ID) == "" || strings.TrimSpace(input.Run.WorkspaceID) == "" || strings.TrimSpace(input.Fence.LeaseToken) == "" {
		return run.SaveTurnParams{}, fmt.Errorf("workflow Run projection requires claimed Run identity")
	}
	if input.Now.IsZero() {
		return run.SaveTurnParams{}, fmt.Errorf("workflow Run projection requires now")
	}
	input.Now = input.Now.UTC().Truncate(time.Microsecond)
	artifacts, err := externalizeTransitionArtifacts(&input)
	if err != nil {
		return run.SaveTurnParams{}, fmt.Errorf("prepare workflow Run artifacts: %w", err)
	}
	statePayload, err := json.Marshal(input.Transition.State)
	if err != nil {
		return run.SaveTurnParams{}, fmt.Errorf("encode workflow Run state: %w", err)
	}
	statePayload, err = run.DeterministicStateJSON(statePayload)
	if err != nil {
		return run.SaveTurnParams{}, fmt.Errorf("%w: %v", errRuntimeStateLimit, err)
	}

	activationStatus := make(map[string]runkernel.ActivationStatus, len(input.Transition.State.Activations))
	for _, activation := range input.Transition.State.Activations {
		activationStatus[activation.ID] = activation.Status
	}
	intents := make(map[string]runkernel.Intent, len(input.Transition.Intents))
	liveIntents := make([]runkernel.Intent, 0, len(input.Transition.Intents))
	seenIntents := make(map[string]struct{}, len(input.Transition.Intents))
	for _, intent := range input.Transition.Intents {
		if strings.TrimSpace(intent.ActivationID) == "" {
			return run.SaveTurnParams{}, fmt.Errorf("workflow Run intent activationId is required")
		}
		if _, exists := seenIntents[intent.ActivationID]; exists {
			return run.SaveTurnParams{}, fmt.Errorf("activation %q emitted more than one effect intent", intent.ActivationID)
		}
		seenIntents[intent.ActivationID] = struct{}{}
		status, exists := activationStatus[intent.ActivationID]
		if !exists {
			return run.SaveTurnParams{}, fmt.Errorf("workflow Run intent references unknown activation %q", intent.ActivationID)
		}
		// A pure continuation in the same bounded reducer turn can resolve a
		// timer before its intent is ever committed. Persisting that obsolete
		// timer would leave a pending effect on a terminal Run.
		if status != runkernel.ActivationWaiting {
			continue
		}
		intents[intent.ActivationID] = intent
		liveIntents = append(liveIntents, intent)
	}
	input.Transition.Intents = liveIntents

	activationMutations, err := projectActivations(input, intents)
	if err != nil {
		return run.SaveTurnParams{}, err
	}
	effectMutations, err := projectEffects(input)
	if err != nil {
		return run.SaveTurnParams{}, err
	}
	eventDrafts, err := projectEvents(input.Transition.Events)
	if err != nil {
		return run.SaveTurnParams{}, err
	}

	drainingDispatch := input.Transition.State.Status.Terminal() && hasCrossedDispatchEffect(input.ExistingEffects)
	persistedStatus := string(input.Transition.State.Status)
	if drainingDispatch {
		// A dispatch lease is the durable commitment to perform external I/O.
		// Keep the Run nonterminal until that worker records its receipt; the
		// terminal reducer state is retained and will finalize on the wakeup.
		persistedStatus = run.RunStatusWaiting
	}
	params := run.SaveTurnParams{
		Fence: input.Fence, ExpectedRevision: input.Run.Revision,
		Status: persistedStatus, StateJSON: statePayload, Artifacts: artifacts,
		RunnableAt: runnableAt(
			input.Transition.State,
			input.Transition.Intents,
			input.ExistingEffects,
			input.Now,
			input.HasMoreWork,
			input.BlockReadyOnTaskWatchDispatch,
		),
		StartedAt: input.Run.StartedAt, ReleaseLease: true,
		Activations: activationMutations, Effects: effectMutations, Events: eventDrafts, Now: input.Now,
	}
	if len(input.Transition.State.Output) != 0 {
		params.OutputJSON, err = json.Marshal(input.Transition.State.Output)
		if err != nil {
			return run.SaveTurnParams{}, fmt.Errorf("encode workflow Run output: %w", err)
		}
	}
	if failure := input.Transition.State.Failure; failure != nil {
		params.ErrorCode = strings.TrimSpace(failure.Code)
		params.ErrorJSON, err = runFailureFromKernel(failure, "reducer")
		if err != nil {
			return run.SaveTurnParams{}, fmt.Errorf("encode workflow Run failure: %w", err)
		}
	}
	if input.Transition.State.Status.Terminal() && !drainingDispatch {
		completed := input.Now
		params.CompletedAt = &completed
		if input.Transition.State.Status == runkernel.StatusCanceled {
			params.CanceledAt = &completed
		}
	}
	if err := validateTurnPayloadSizes(params); err != nil {
		return run.SaveTurnParams{}, err
	}
	return params, nil
}

func validateTurnPayloadSizes(params run.SaveTurnParams) error {
	type payload struct {
		label    string
		value    []byte
		required bool
	}
	values := []payload{
		{label: "Run output", value: params.OutputJSON},
		{label: "Run error", value: params.ErrorJSON},
	}
	for index, mutation := range params.Activations {
		values = append(values,
			payload{label: fmt.Sprintf("activation[%d] input", index), value: mutation.Value.InputJSON, required: true},
			payload{label: fmt.Sprintf("activation[%d] output", index), value: mutation.Value.OutputJSON},
			payload{label: fmt.Sprintf("activation[%d] scope", index), value: mutation.Value.ScopeContextJSON, required: true},
			payload{label: fmt.Sprintf("activation[%d] error", index), value: mutation.Value.ErrorJSON},
		)
	}
	for index, mutation := range params.Effects {
		values = append(values,
			payload{label: fmt.Sprintf("effect[%d] request", index), value: mutation.Value.RequestJSON, required: true},
			payload{label: fmt.Sprintf("effect[%d] result", index), value: mutation.Value.ResultJSON},
			payload{label: fmt.Sprintf("effect[%d] error", index), value: mutation.Value.ErrorJSON},
		)
	}
	for index, event := range params.Events {
		values = append(values, payload{label: fmt.Sprintf("event[%d]", index), value: event.PayloadJSON, required: true})
	}
	for _, value := range values {
		if _, err := run.DeterministicPayloadJSON(value.label, value.value, value.required); err != nil {
			return fmt.Errorf("%w: %v", errRuntimePayloadLimit, err)
		}
	}
	return nil
}

func projectActivations(input ProjectTurnInput, intents map[string]runkernel.Intent) ([]run.ActivationMutation, error) {
	existing := make(map[string]run.Activation, len(input.ExistingActivations))
	for _, value := range input.ExistingActivations {
		if value.RunID != input.Run.ID || value.WorkspaceID != input.Run.WorkspaceID || strings.TrimSpace(value.LogicalKey) == "" {
			return nil, fmt.Errorf("persisted workflow Run activation belongs to a different Run or has no logical key")
		}
		if _, duplicate := existing[value.LogicalKey]; duplicate {
			return nil, fmt.Errorf("persisted workflow Run activation logical key %q is duplicated", value.LogicalKey)
		}
		existing[value.LogicalKey] = value
	}
	mutations := make([]run.ActivationMutation, 0, len(input.Transition.State.Activations))
	for _, activation := range input.Transition.State.Activations {
		value, err := projectActivation(input, activation, intents[activation.ID], existing[activation.ID])
		if err != nil {
			return nil, err
		}
		prior, found := existing[activation.ID]
		if !found {
			mutations = append(mutations, run.ActivationMutation{Operation: run.MutationInsert, Value: value})
			continue
		}
		if activationEquivalent(prior, value) {
			continue
		}
		mutations = append(mutations, run.ActivationMutation{
			Operation: run.MutationUpdate, Value: value, ExpectedStatus: prior.Status,
		})
	}
	return mutations, nil
}

type activationScopeContext struct {
	Lineage        []runkernel.ScopeRef `json:"lineage"`
	OwnedScopeIDs  []string             `json:"ownedScopeIds"`
	TriggerScopeID string               `json:"triggerScopeId,omitempty"`
	SelectedPorts  []string             `json:"selectedPorts"`
}

func projectActivation(input ProjectTurnInput, activation runkernel.Activation, intent runkernel.Intent, prior run.Activation) (run.Activation, error) {
	node, ok := input.Graph.Node(activation.NodeID)
	if !ok {
		return run.Activation{}, fmt.Errorf("activation %q references unknown node %q", activation.ID, activation.NodeID)
	}
	id, err := StableChildID("wra", input.Run.ID, activation.ID)
	if err != nil {
		return run.Activation{}, err
	}
	inputJSON, err := marshalValues(activation.Input)
	if err != nil {
		return run.Activation{}, fmt.Errorf("encode activation %q input: %w", activation.ID, err)
	}
	outputJSON, err := marshalOptionalValues(activation.Output)
	if err != nil {
		return run.Activation{}, fmt.Errorf("encode activation %q output: %w", activation.ID, err)
	}
	scopeJSON, err := json.Marshal(activationScopeContext{
		Lineage:       append([]runkernel.ScopeRef(nil), activation.Lineage...),
		OwnedScopeIDs: append([]string(nil), activation.OwnedScopeIDs...), TriggerScopeID: activation.TriggerScopeID,
		SelectedPorts: append([]string(nil), activation.SelectedPorts...),
	})
	if err != nil {
		return run.Activation{}, fmt.Errorf("encode activation %q scope context: %w", activation.ID, err)
	}
	value := run.Activation{
		ID: id, WorkspaceID: input.Run.WorkspaceID, RunID: input.Run.ID, LogicalKey: activation.ID,
		NodeID: activation.NodeID, Opcode: string(node.Opcode), Status: string(activation.Status),
		InputJSON: inputJSON, OutputJSON: outputJSON, ScopeContextJSON: scopeJSON,
		AttemptCount: prior.AttemptCount, CreatedAt: prior.CreatedAt, UpdatedAt: input.Now,
	}
	if activation.ParentActivationID != "" {
		value.ParentActivationID, err = StableChildID("wra", input.Run.ID, activation.ParentActivationID)
		if err != nil {
			return run.Activation{}, fmt.Errorf("project activation %q parent: %w", activation.ID, err)
		}
	}
	if activation.Status != runkernel.ActivationReady && value.AttemptCount == 0 {
		value.AttemptCount = 1
	}
	if len(activation.SelectedPorts) == 1 {
		value.SelectedPort = activation.SelectedPorts[0]
	}
	if activation.Failure != nil {
		value.ErrorCode = activation.Failure.Code
		value.ErrorJSON, err = runFailureFromKernel(activation.Failure, "reducer")
		if err != nil {
			return run.Activation{}, fmt.Errorf("encode activation %q failure: %w", activation.ID, err)
		}
	}
	if !prior.CreatedAt.IsZero() {
		value.StartedAt, value.CompletedAt = prior.StartedAt, prior.CompletedAt
	}
	if activation.Status != runkernel.ActivationReady && value.StartedAt == nil {
		started := input.Now
		value.StartedAt = &started
	}
	if activation.Status == runkernel.ActivationSucceeded || activation.Status == runkernel.ActivationFailed || activation.Status == runkernel.ActivationCanceled {
		if value.CompletedAt == nil {
			completed := input.Now
			value.CompletedAt = &completed
		}
	}
	if intent.ActivationID != "" {
		value.DeadlineAt = cloneTime(intent.Deadline)
		value.WakeAt = cloneTime(intent.WakeAt)
		if intent.Kind != runkernel.IntentWait {
			value.RetryAt = cloneTime(&input.Now)
		}
	} else if activation.Status == runkernel.ActivationWaiting && !prior.CreatedAt.IsZero() {
		// A waiting activation emits its effect intent only once. Subsequent
		// turns must retain its scheduling/deadline fields until completion.
		value.RetryAt = cloneTime(prior.RetryAt)
		value.WakeAt = cloneTime(prior.WakeAt)
		value.DeadlineAt = cloneTime(prior.DeadlineAt)
	}
	if !prior.CreatedAt.IsZero() {
		value.CreatedAt = prior.CreatedAt
	}
	return value, nil
}

func projectEffects(input ProjectTurnInput) ([]run.EffectMutation, error) {
	existing := make(map[string]run.Effect, len(input.ExistingEffects))
	for _, effect := range input.ExistingEffects {
		if effect.RunID != input.Run.ID || effect.WorkspaceID != input.Run.WorkspaceID || strings.TrimSpace(effect.IdempotencyKey) == "" {
			return nil, fmt.Errorf("persisted workflow Run effect belongs to a different Run or has no idempotency key")
		}
		if _, duplicate := existing[effect.IdempotencyKey]; duplicate {
			return nil, fmt.Errorf("persisted workflow Run effect idempotency key is duplicated")
		}
		existing[effect.IdempotencyKey] = effect
	}
	mutations := make([]run.EffectMutation, 0, len(input.Transition.Intents))
	for _, intent := range input.Transition.Intents {
		key, err := StableEffectIdempotencyKey(input.Run.ID, intent.ActivationID, intent.Kind)
		if err != nil {
			return nil, err
		}
		if _, found := existing[key]; found {
			continue
		}
		effectID, err := StableChildID("wfx", input.Run.ID, intent.ActivationID+"\x00"+string(intent.Kind))
		if err != nil {
			return nil, err
		}
		activationID, err := StableChildID("wra", input.Run.ID, intent.ActivationID)
		if err != nil {
			return nil, err
		}
		request := EffectRequest{Intent: intent}
		if intent.Kind == runkernel.IntentSendMessage {
			request.Message = &MessageRequest{
				WorkspaceID: input.Run.WorkspaceID, PlanID: input.Run.PlanID, PlanVersionID: input.Run.PlanVersionID,
				RunID: input.Run.ID, ActivationID: intent.ActivationID, EffectID: effectID, NodeID: intent.NodeID,
				SourceKind: normalizedSourceKind(input.SourceKind), WorkflowName: strings.TrimSpace(input.Run.WorkflowNameSnapshot),
				OwnerUserID:    input.Run.Identity.OwnerUserID,
				AuthorMemberID: input.Run.Identity.AuthorMemberID, TeamID: input.Run.TeamID,
				TargetRef: intent.Target, Content: intent.Text, ContentRef: intent.TextRef,
				ClientMessageID: "workflow-run:" + effectID, Callback: intent.Callback, StartThread: intent.StartThread,
				ResponderAgentRef:   strings.TrimSpace(intent.Responder),
				ResponderCandidates: append([]string(nil), intent.ResponderCandidates...),
				SelectOneStrategy:   intent.SelectOneStrategy,
				WorkScopeMode:       intent.WorkScopeMode,
			}
			if intent.Scope != nil {
				scope := *intent.Scope
				request.Message.Scope = &scope
			}
			if intent.WorkScope != nil {
				scope := *intent.WorkScope
				request.Message.WorkScope = &scope
			}
			request.Message.RelatedScopes = append([]domain.MessageScopeValue(nil), intent.RelatedScopes...)
			request.Message.TaskWatchHandoff = cloneTaskWatchHandoff(intent.TaskWatchHandoff)
		}
		payload, err := EncodeEffectRequest(request)
		if err != nil {
			return nil, err
		}
		retryAt := input.Now
		if intent.Kind == runkernel.IntentWait {
			retryAt = intent.WakeAt.UTC().Truncate(time.Microsecond)
		}
		value := run.Effect{
			ID: effectID, WorkspaceID: input.Run.WorkspaceID, RunID: input.Run.ID, ActivationID: activationID,
			Kind: string(intent.Kind), IdempotencyKey: key, Status: run.EffectStatusPending,
			RequestJSON: payload, RetryAt: &retryAt, DeadlineAt: cloneTime(intent.Deadline), CreatedAt: input.Now, UpdatedAt: input.Now,
		}
		mutations = append(mutations, run.EffectMutation{Operation: run.MutationInsert, Value: value})
	}

	activationByLogical := make(map[string]runkernel.Activation, len(input.Transition.State.Activations))
	for _, activation := range input.Transition.State.Activations {
		activationByLogical[activation.ID] = activation
	}
	for _, effect := range input.ExistingEffects {
		request, err := DecodeEffectRequest(effect.RequestJSON)
		if err != nil {
			return nil, fmt.Errorf("decode persisted workflow Run effect %q: %w", effect.ID, err)
		}
		activation, found := activationByLogical[request.Intent.ActivationID]
		if !found || activation.Status == runkernel.ActivationWaiting || effectTerminal(effect.Status) || effectCrossedDispatch(effect) {
			continue
		}
		value := effect
		value.Status = run.EffectStatusCanceled
		value.LeaseOwner, value.LeaseToken, value.LeaseExpiresAt = "", "", nil
		completed := input.Now
		value.CompletedAt = &completed
		mutations = append(mutations, run.EffectMutation{
			Operation: run.MutationUpdate, Value: value, ExpectedStatus: effect.Status,
		})
	}
	return mutations, nil
}

func cloneTaskWatchHandoff(value *runkernel.TaskWatchHandoff) *runkernel.TaskWatchHandoff {
	if value == nil {
		return nil
	}
	cloned := *value
	cloned.CausalPredecessorAgentIDs = append([]string{}, value.CausalPredecessorAgentIDs...)
	return &cloned
}

func projectEvents(values []runkernel.Event) ([]run.EventDraft, error) {
	drafts := make([]run.EventDraft, 0, len(values))
	for _, value := range values {
		payload, err := json.Marshal(value)
		if err != nil {
			return nil, fmt.Errorf("encode workflow Run event %q: %w", value.Kind, err)
		}
		drafts = append(drafts, run.EventDraft{Type: value.Kind, PayloadJSON: payload})
	}
	return drafts, nil
}

func runnableAt(
	state *runkernel.State,
	intents []runkernel.Intent,
	effects []run.Effect,
	now time.Time,
	hasMoreWork bool,
	blockReadyOnTaskWatchDispatch bool,
) time.Time {
	if hasMoreWork && !blockReadyOnTaskWatchDispatch {
		return now
	}
	if !blockReadyOnTaskWatchDispatch {
		for _, activation := range state.Activations {
			if activation.Status == runkernel.ActivationReady {
				return now
			}
		}
	}
	next := now.Add(parkedRunDelay)
	for _, intent := range intents {
		for _, candidate := range []*time.Time{intent.WakeAt, intent.Deadline} {
			if candidate != nil && candidate.After(now) && candidate.Before(next) {
				next = candidate.UTC().Truncate(time.Microsecond)
			}
		}
	}
	for _, effect := range effects {
		switch effect.Status {
		case run.EffectStatusPending, run.EffectStatusDispatching, run.EffectStatusDelivered, run.EffectStatusAwaiting:
		default:
			continue
		}
		candidates := []*time.Time{effect.RetryAt, effect.DeadlineAt}
		if effect.Status == run.EffectStatusDispatching {
			// While a dispatch commitment is in flight, only its lease expiry
			// is a meaningful graph wake time. The action deadline cannot
			// revoke a worker that may already have performed external I/O.
			candidates = []*time.Time{effect.LeaseExpiresAt}
		}
		for _, candidate := range candidates {
			if candidate != nil && candidate.After(now) && candidate.Before(next) {
				next = candidate.UTC().Truncate(time.Microsecond)
			}
		}
	}
	return next
}

func effectTerminal(status string) bool {
	return status == run.EffectStatusCompleted ||
		status == run.EffectStatusFailed ||
		status == run.EffectStatusCanceled
}

func hasCrossedDispatchEffect(effects []run.Effect) bool {
	for _, effect := range effects {
		if effectCrossedDispatch(effect) {
			return true
		}
	}
	return false
}

func effectCrossedDispatch(effect run.Effect) bool {
	return effect.Status == run.EffectStatusDispatching ||
		(effect.Status == run.EffectStatusPending && effect.AttemptCount > 0)
}

func activationEquivalent(left, right run.Activation) bool {
	return left.ID == right.ID && left.ParentActivationID == right.ParentActivationID && left.Status == right.Status && left.SelectedPort == right.SelectedPort &&
		left.AttemptCount == right.AttemptCount && left.ErrorCode == right.ErrorCode &&
		bytes.Equal(left.InputJSON, right.InputJSON) && bytes.Equal(left.OutputJSON, right.OutputJSON) &&
		bytes.Equal(left.ScopeContextJSON, right.ScopeContextJSON) && bytes.Equal(left.ErrorJSON, right.ErrorJSON) &&
		reflect.DeepEqual(left.RetryAt, right.RetryAt) && reflect.DeepEqual(left.WakeAt, right.WakeAt) &&
		reflect.DeepEqual(left.DeadlineAt, right.DeadlineAt) && reflect.DeepEqual(left.StartedAt, right.StartedAt) &&
		reflect.DeepEqual(left.CompletedAt, right.CompletedAt)
}

func marshalValues(values runkernel.Values) (json.RawMessage, error) {
	if values == nil {
		values = runkernel.Values{}
	}
	return json.Marshal(values)
}

func marshalOptionalValues(values runkernel.Values) (json.RawMessage, error) {
	if values == nil {
		return nil, nil
	}
	return json.Marshal(values)
}

func cloneTime(value *time.Time) *time.Time {
	if value == nil {
		return nil
	}
	normalized := value.UTC().Truncate(time.Microsecond)
	return &normalized
}

func normalizedSourceKind(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return "workflow_run"
	}
	return value
}

func sortActivationMutations(values []run.ActivationMutation) {
	sort.SliceStable(values, func(i, j int) bool { return values[i].Value.LogicalKey < values[j].Value.LogicalKey })
}