package orchestrator

import (
	"context"
	"time"

	workflowrunstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runartifact"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runlimit"
)

// ProjectAndSaveTurn is the production orchestration seam shared by the runtime
// engine and integration tests. It projects one reducer transition through the
// public projector (ProjectTurn), applies the bounded-size validation the
// runtime enforces before persistence, and saves the turn atomically via
// SaveTurn. The runtime engine calls it for every reducer turn; real-DB tests
// call it to drive the EXACT production persistence path — the projector builds
// the durable EffectRequest and row mutations, and SaveTurn persists them — with
// no hand-built EffectRequest and no handwritten row inserts.
//
// hasMoreWork / blockReadyOnTaskWatch are the bounded-engine scheduling flags
// the engine derives and the projector folds into the runnableAt scheduling.
func ProjectAndSaveTurn(
	ctx context.Context,
	repository Repository,
	claim workflowrunstore.ClaimedRun,
	graph *runkernel.Graph,
	transition runkernel.Transition,
	existingActivations []workflowrunstore.Activation,
	existingEffects []workflowrunstore.Effect,
	sourceKind string,
	now time.Time,
	hasMoreWork bool,
	blockReadyOnTaskWatch bool,
) (workflowrunstore.SaveTurnResult, error) {
	return projectAndSaveTurn(ctx, repository, claim, graph, transition, existingActivations, existingEffects, sourceKind, now, hasMoreWork, blockReadyOnTaskWatch, nil)
}

// ProjectAndSaveTurnWithArtifactCommitHook is the same production boundary with
// a narrowly scoped post-commit seam. The hook is called only when this exact
// turn committed at least one oversized Join aggregate artifact; earlier
// MaxReductions turns and unrelated message artifacts do not trigger it.
func ProjectAndSaveTurnWithArtifactCommitHook(
	ctx context.Context,
	repository Repository,
	claim workflowrunstore.ClaimedRun,
	graph *runkernel.Graph,
	transition runkernel.Transition,
	existingActivations []workflowrunstore.Activation,
	existingEffects []workflowrunstore.Effect,
	sourceKind string,
	now time.Time,
	hasMoreWork bool,
	blockReadyOnTaskWatch bool,
	afterArtifactCommit func() error,
) (workflowrunstore.SaveTurnResult, error) {
	return projectAndSaveTurn(ctx, repository, claim, graph, transition, existingActivations, existingEffects, sourceKind, now, hasMoreWork, blockReadyOnTaskWatch, afterArtifactCommit)
}

func projectAndSaveTurn(
	ctx context.Context,
	repository Repository,
	claim workflowrunstore.ClaimedRun,
	graph *runkernel.Graph,
	transition runkernel.Transition,
	existingActivations []workflowrunstore.Activation,
	existingEffects []workflowrunstore.Effect,
	sourceKind string,
	now time.Time,
	hasMoreWork bool,
	blockReadyOnTaskWatch bool,
	afterArtifactCommit func() error,
) (workflowrunstore.SaveTurnResult, error) {
	params, err := ProjectTurn(ProjectTurnInput{
		Run: claim.Run, Fence: claim.Fence, Graph: graph, Transition: transition,
		ExistingActivations: existingActivations, ExistingEffects: existingEffects,
		SourceKind: sourceKind, Now: now,
		HasMoreWork: hasMoreWork, BlockReadyOnTaskWatchDispatch: blockReadyOnTaskWatch,
	})
	if err != nil {
		return workflowrunstore.SaveTurnResult{}, err
	}
	if err := validateTurnPayloadSizes(params); err != nil {
		return workflowrunstore.SaveTurnResult{}, err
	}
	result, err := repository.SaveTurn(ctx, params)
	if err != nil {
		return workflowrunstore.SaveTurnResult{}, err
	}
	if afterArtifactCommit != nil && hasJoinAggregateArtifact(params.Artifacts) {
		if err := afterArtifactCommit(); err != nil {
			return workflowrunstore.SaveTurnResult{}, err
		}
	}
	return result, nil
}

func hasJoinAggregateArtifact(drafts []runartifact.Draft) bool {
	for _, draft := range drafts {
		if draft.Ref.Kind == runartifact.KindJoinAggregate && draft.Ref.CanonicalByteLength > runlimit.MaxRunPayloadBytes {
			return true
		}
	}
	return false
}