package orchestrator

import (
	"context"
	"errors"
	managedgo "github.com/tidbcloud/AgentTeam/internal/observability/logging"
	workflowrunstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	"log/slog"
	"sync"
	"time"
)

const defaultRuntimeWorkerPollInterval = time.Second

type ReducerRunner interface {
	RunOnce(context.Context) (RunOnceOutcome, error)
}

type EffectRunner interface {
	DispatchOnce(context.Context) (DispatchOutcome, error)
}

// LeaseReconciler releases provably-terminal execution leases after a restart
// so the next FIFO head can run; it never touches live owners. It is optional
// and runs once per worker pass when configured.
type LeaseReconciler interface {
	ReconcileOnce(context.Context) error
}

// RecoveryRunnable wakes awaiting send_message.callback effects whose frozen
// responder stopped working. It is optional, never touches runnableAt or the
// execution lease, and runs once per worker pass after dispatch/reconcile.
type RecoveryRunnable interface {
	RecoverOnce(context.Context) (RecoveryOutcome, error)
}

// ArtifactMaintenanceRunner owns bounded retention assignment, expiry sweep,
// and orphan cleanup for the same production Run store used by the worker.
type ArtifactMaintenanceRunner interface {
	RunArtifactMaintenance(context.Context, time.Time, int) (workflowrunstore.ArtifactMaintenanceResult, error)
}

type WorkerOptions struct {
	Logger              *slog.Logger
	PollInterval        time.Duration
	Reconciler          LeaseReconciler
	Recovery            RecoveryRunnable
	ArtifactMaintenance ArtifactMaintenanceRunner
}

// Worker owns only lifecycle/ticking. The reducer completes and releases all
// DB transactions before the effect dispatcher performs network I/O, and one
// pass never overlaps the next.
type Worker struct {
	reducer    ReducerRunner
	effects    EffectRunner
	reconciler LeaseReconciler
	recovery   RecoveryRunnable
	artifacts  ArtifactMaintenanceRunner
	logger     *slog.Logger
	interval   time.Duration
	start      sync.Once
}

type WorkerOutcome struct {
	Runs       RunOnceOutcome
	Effects    DispatchOutcome
	Reconciled int
	Recovery   RecoveryOutcome
	Artifacts  workflowrunstore.ArtifactMaintenanceResult
}

func NewWorker(reducer ReducerRunner, effects EffectRunner, options WorkerOptions) *Worker {
	interval := options.PollInterval
	if interval <= 0 {
		interval = defaultRuntimeWorkerPollInterval
	}
	return &Worker{reducer: reducer, effects: effects, reconciler: options.Reconciler, recovery: options.Recovery, artifacts: options.ArtifactMaintenance, logger: options.Logger, interval: interval}
}

func (worker *Worker) Start(ctx context.Context) {
	if worker == nil {
		return
	}
	worker.start.Do(func() { managedgo.Go(worker.logger, func() { worker.Run(ctx) }) })
}

func (worker *Worker) Run(ctx context.Context) {
	if worker == nil || ctx.Err() != nil {
		return
	}
	worker.runAndLog(ctx)
	ticker := time.NewTicker(worker.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			worker.runAndLog(ctx)
		}
	}
}

func (worker *Worker) RunOnce(ctx context.Context) (WorkerOutcome, error) {
	if worker == nil {
		return WorkerOutcome{}, nil
	}
	var outcome WorkerOutcome
	var failures []error
	if worker.reducer != nil {
		value, err := worker.reducer.RunOnce(ctx)
		outcome.Runs = value
		if err != nil {
			failures = append(failures, err)
		}
	}
	if worker.effects != nil {
		value, err := worker.effects.DispatchOnce(ctx)
		outcome.Effects = value
		if err != nil {
			failures = append(failures, err)
		}
	}
	if worker.reconciler != nil {
		if err := worker.reconciler.ReconcileOnce(ctx); err != nil {
			failures = append(failures, err)
		}
	}
	if worker.recovery != nil {
		value, err := worker.recovery.RecoverOnce(ctx)
		outcome.Recovery = value
		if err != nil {
			failures = append(failures, err)
		}
	}
	if worker.artifacts != nil {
		value, err := worker.artifacts.RunArtifactMaintenance(ctx, time.Now().UTC(), 100)
		outcome.Artifacts = value
		if err != nil {
			failures = append(failures, err)
		}
	}
	return outcome, errors.Join(failures...)
}

func (worker *Worker) runAndLog(ctx context.Context) {
	outcome, err := worker.RunOnce(ctx)
	if err != nil && !errors.Is(err, context.Canceled) {
		if worker.logger != nil {
			worker.logger.Warn("workflow Run worker pass failed", slog.Any("error", err))
		}
		return
	}
	if worker.logger == nil || outcome.Runs.Claimed == 0 && outcome.Effects.Candidates == 0 && outcome.Recovery.Candidates == 0 && outcome.Artifacts.AssignedExpiry == 0 && outcome.Artifacts.Deleted == 0 && outcome.Artifacts.OrphanedBlobs == 0 {
		return
	}
	worker.logger.Debug("workflow Run worker pass completed",
		slog.Int("runs_claimed", outcome.Runs.Claimed),
		slog.Int("runs_succeeded", outcome.Runs.Succeeded),
		slog.Int("runs_failed", outcome.Runs.Failed),
		slog.Int("effects_listed", outcome.Effects.Candidates),
		slog.Int("effects_claimed", outcome.Effects.Claimed),
		slog.Int("effects_resolved", outcome.Effects.Resolved),
		slog.Int("effects_failed", outcome.Effects.Failed),
		slog.Int("recovery_candidates", outcome.Recovery.Candidates),
		slog.Int("recovery_working", outcome.Recovery.Working),
		slog.Int("recovery_unknown", outcome.Recovery.Unknown),
		slog.Int("recovery_sent", outcome.Recovery.Sent),
		slog.Int("recovery_failed", outcome.Recovery.Failed),
		slog.Int64("artifacts_expiry_assigned", outcome.Artifacts.AssignedExpiry),
		slog.Int64("artifacts_deleted", outcome.Artifacts.Deleted),
		slog.Int64("artifact_orphans_deleted", outcome.Artifacts.OrphanedBlobs),
	)
}