package invocation

import (
	"context"
	"errors"
	"github.com/tidbcloud/AgentTeam/internal/observability/logging"
	"log/slog"
	"sync"
	"time"
)

const defaultScheduleWorkerPollInterval = time.Second

// ScheduleDispatcherRunner is the bounded dispatcher seam owned by the
// server lifecycle worker.
type ScheduleDispatcherRunner interface {
	RunOnce(context.Context) (ScheduleDispatchOutcome, error)
}

// ScheduleWorkerOptions configures the server-owned schedule worker.
type ScheduleWorkerOptions struct {
	Logger       *slog.Logger
	PollInterval time.Duration
}

// ScheduleWorker repeatedly invokes one bounded dispatcher pass. Passes are
// deliberately run synchronously in a single goroutine, so a slow pass can
// never overlap the next ticker event.
type ScheduleWorker struct {
	dispatcher ScheduleDispatcherRunner
	logger     *slog.Logger
	interval   time.Duration
	start      sync.Once
}

// NewScheduleWorker constructs a server-owned schedule worker.
func NewScheduleWorker(dispatcher ScheduleDispatcherRunner, options ScheduleWorkerOptions) *ScheduleWorker {
	interval := options.PollInterval
	if interval <= 0 {
		interval = defaultScheduleWorkerPollInterval
	}
	return &ScheduleWorker{
		dispatcher: dispatcher,
		logger:     options.Logger,
		interval:   interval,
	}
}

// Start launches the worker at most once and returns immediately.
func (worker *ScheduleWorker) Start(ctx context.Context) {
	if worker == nil {
		return
	}
	worker.start.Do(func() {
		logging.Go(worker.logger, func() { worker.Run(ctx) })
	})
}

// Run performs one immediate pass, then polls until ctx is cancelled. A pass
// completes before the worker observes another tick, which guarantees there
// is at most one in-flight RunOnce call per worker.
func (worker *ScheduleWorker) Run(ctx context.Context) {
	if worker == nil || worker.dispatcher == nil {
		return
	}
	if 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)
		}
	}
}

// RunOnce exposes one bounded pass for tests and explicit maintenance calls.
func (worker *ScheduleWorker) RunOnce(ctx context.Context) (ScheduleDispatchOutcome, error) {
	if worker == nil || worker.dispatcher == nil {
		return ScheduleDispatchOutcome{}, nil
	}
	return worker.dispatcher.RunOnce(ctx)
}

func (worker *ScheduleWorker) runAndLog(ctx context.Context) {
	outcome, err := worker.RunOnce(ctx)
	if err != nil && !errors.Is(err, context.Canceled) {
		worker.logWarn("Workflow Invocation schedule pass failed", slog.Any("error", err))
	}
	for _, candidateErr := range outcome.Errors {
		if candidateErr == nil || errors.Is(candidateErr, context.Canceled) {
			continue
		}
		worker.logWarn("Workflow Invocation schedule candidate failed", slog.Any("error", candidateErr))
	}
	worker.logOutcome(outcome)
}

func (worker *ScheduleWorker) logOutcome(outcome ScheduleDispatchOutcome) {
	if worker.logger == nil || (outcome.DueListed == 0 && outcome.Attempted == 0 && len(outcome.Errors) == 0) {
		return
	}
	worker.logger.Debug("Workflow Invocation schedule pass completed",
		slog.Int("due_listed", outcome.DueListed),
		slog.Int("attempted", outcome.Attempted),
		slog.Int("created", outcome.Created),
		slog.Int("replayed", outcome.Replayed),
		slog.Int("stale", outcome.Stale),
		slog.Int("fenced", outcome.Fenced),
		slog.Int("blocked", outcome.Blocked),
		slog.Int("errors", len(outcome.Errors)),
	)
}

func (worker *ScheduleWorker) logWarn(message string, attrs ...any) {
	if worker.logger != nil {
		worker.logger.Warn(message, attrs...)
	}
}