// Package schedule executes AgentTeam-owned schedule occurrences through automation runs.
package schedule

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"github.com/go-sql-driver/mysql"

	"github.com/tidbcloud/AgentTeam/internal/core/runeffect"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	schedulestore "github.com/tidbcloud/AgentTeam/internal/store/schedule"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
)

const (
	defaultPollInterval = time.Second
	defaultBatchSize    = 20
	defaultMaxAttempts  = 3
	defaultRetryBackoff = 10 * time.Second
	defaultStaleRunAge  = 30 * time.Second
)

// Store is the schedule execution persistence port.
type Store interface {
	ListDueDefinitionIDs(ctx context.Context, params schedulestore.DueDefinitionListQuery) ([]string, error)
	ReserveDueAutomationRun(ctx context.Context, definitionID string, now time.Time) (schedulestore.ReservedRun, bool, error)
	ListRetryRunIDs(ctx context.Context, now time.Time, limit int) ([]string, error)
	ClaimRetryAutomationRun(ctx context.Context, runID string, now time.Time) (schedulestore.ReservedRun, bool, error)
	AbandonInactiveRetryRuns(ctx context.Context, now time.Time, limit int) (int, error)
	MarkAutomationRunRetryWaiting(ctx context.Context, params schedulestore.AutomationRunRetryUpdate) (schedulestore.AutomationRun, error)
	CompleteAutomationRun(ctx context.Context, params schedulestore.AutomationRunCompletion) (schedulestore.AutomationRun, error)
	ListStaleRunningRunIDs(ctx context.Context, recoveredBefore time.Time, limit int) ([]string, error)
	RecoverStaleRunningRun(ctx context.Context, runID string, recoveredBefore time.Time, now time.Time, maxAttempts int) (bool, error)
}

// EffectWriter writes the visible effect for a reserved schedule automation run.
type EffectWriter interface {
	WriteScheduleEffect(ctx context.Context, request EffectRequest) (runeffect.Effect, error)
}

// EffectRequest contains the immutable run snapshot that a writer needs to
// validate fire-time permissions and create exactly one effect.
type EffectRequest struct {
	WorkspaceID       string
	TargetKind        string
	TargetRef         string
	Content           string
	AlsoStartThread   bool
	MentionHints      []message.MessageMentionHint
	Metadata          map[string]any
	Source            message.ContributionSource
	ClientMessageID   string
	OwnerMemberID     string
	PrincipalMemberID string
	AuthorMemberID    string
	Definition        schedulestore.Definition
	Run               schedulestore.AutomationRun
}

// RunPassResult summarizes one scheduler pass.
type RunPassResult struct {
	RecoveredStale   int
	AbandonedRetries int
	Due              int
	Retried          int
}

// ServiceOptions configures schedule execution.
type ServiceOptions struct {
	Clock          func() time.Time
	PollInterval   time.Duration
	DueBatchSize   int
	RetryBatchSize int
	MaxAttempts    int
	RetryBackoff   time.Duration
	StaleRunAge    time.Duration
	Writer         EffectWriter
}

// Service reserves due schedules, writes effects, and completes automation runs.
type Service struct {
	logger         *slog.Logger
	store          Store
	clock          func() time.Time
	pollInterval   time.Duration
	dueBatchSize   int
	retryBatchSize int
	maxAttempts    int
	retryBackoff   time.Duration
	staleRunAge    time.Duration
	writer         EffectWriter
}

// New constructs a schedule execution service.
func New(logger *slog.Logger, store Store, options ServiceOptions) *Service {
	clock := options.Clock
	if clock == nil {
		clock = time.Now
	}
	pollInterval := options.PollInterval
	if pollInterval <= 0 {
		pollInterval = defaultPollInterval
	}
	dueBatchSize := options.DueBatchSize
	if dueBatchSize <= 0 {
		dueBatchSize = defaultBatchSize
	}
	retryBatchSize := options.RetryBatchSize
	if retryBatchSize <= 0 {
		retryBatchSize = defaultBatchSize
	}
	maxAttempts := options.MaxAttempts
	if maxAttempts <= 0 {
		maxAttempts = defaultMaxAttempts
	}
	retryBackoff := options.RetryBackoff
	if retryBackoff <= 0 {
		retryBackoff = defaultRetryBackoff
	}
	staleRunAge := options.StaleRunAge
	if staleRunAge <= 0 {
		staleRunAge = defaultStaleRunAge
	}
	return &Service{
		logger:         logger,
		store:          store,
		clock:          clock,
		pollInterval:   pollInterval,
		dueBatchSize:   dueBatchSize,
		retryBatchSize: retryBatchSize,
		maxAttempts:    maxAttempts,
		retryBackoff:   retryBackoff,
		staleRunAge:    staleRunAge,
		writer:         options.Writer,
	}
}

// Run starts schedule execution until ctx is cancelled.
func (service *Service) Run(ctx context.Context) {
	if service == nil || service.store == nil || service.writer == nil {
		return
	}
	runPass := func() {
		result, err := service.RunOnce(ctx)
		if err != nil && !errors.Is(err, context.Canceled) {
			service.logWarn("schedule pass failed", slog.Any("error", err))
			return
		}
		if result.RecoveredStale > 0 || result.AbandonedRetries > 0 || result.Retried > 0 || result.Due > 0 {
			service.logInfo("schedule pass completed", slog.Int("recovered_stale_runs", result.RecoveredStale), slog.Int("abandoned_retry_runs", result.AbandonedRetries), slog.Int("retried_runs", result.Retried), slog.Int("due_schedules", result.Due))
		}
	}

	runPass()
	ticker := time.NewTicker(service.pollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			runPass()
		}
	}
}

// RunOnce performs one bounded schedule execution pass.
func (service *Service) RunOnce(ctx context.Context) (RunPassResult, error) {
	if service == nil || service.store == nil || service.writer == nil {
		return RunPassResult{}, nil
	}

	now := service.now()
	result := RunPassResult{}
	var firstErr error

	recoveredBefore := now.Add(-service.staleRunAge)
	staleIDs, err := service.store.ListStaleRunningRunIDs(ctx, recoveredBefore, service.retryBatchSize)
	if err != nil {
		return result, err
	}
	for _, runID := range staleIDs {
		ok, err := service.store.RecoverStaleRunningRun(ctx, runID, recoveredBefore, now, service.maxAttempts)
		if err != nil {
			if firstErr == nil {
				firstErr = err
			}
			service.logWarn("recover stale schedule run failed", slog.String("run_id", runID), slog.Any("error", err))
			continue
		}
		if ok {
			result.RecoveredStale++
		}
	}

	abandonedRetries, err := service.store.AbandonInactiveRetryRuns(ctx, now, service.retryBatchSize)
	if err != nil {
		if firstErr == nil {
			firstErr = err
		}
		return result, firstErr
	}
	result.AbandonedRetries += abandonedRetries

	dueIDs, err := service.store.ListDueDefinitionIDs(ctx, schedulestore.DueDefinitionListQuery{
		Now:   now,
		Limit: service.dueBatchSize,
	})
	if err != nil {
		if firstErr == nil {
			firstErr = err
		}
		return result, firstErr
	}
	for _, definitionID := range dueIDs {
		dispatched, err := service.dispatchDueDefinition(ctx, definitionID)
		if err != nil {
			if firstErr == nil {
				firstErr = err
			}
			service.logWarn("dispatch due schedule failed", slog.String("schedule_definition_id", definitionID), slog.Any("error", err))
			continue
		}
		if dispatched {
			result.Due++
		}
	}

	retryIDs, err := service.store.ListRetryRunIDs(ctx, now, service.retryBatchSize)
	if err != nil {
		if firstErr == nil {
			firstErr = err
		}
		return result, firstErr
	}
	for _, runID := range retryIDs {
		dispatched, err := service.dispatchRetryRun(ctx, runID)
		if err != nil {
			if firstErr == nil {
				firstErr = err
			}
			service.logWarn("dispatch schedule retry run failed", slog.String("run_id", runID), slog.Any("error", err))
			continue
		}
		if dispatched {
			result.Retried++
		}
	}

	return result, firstErr
}

func (service *Service) dispatchDueDefinition(ctx context.Context, definitionID string) (bool, error) {
	reservation, ok, err := service.store.ReserveDueAutomationRun(ctx, definitionID, service.now())
	if err != nil || !ok {
		return false, err
	}
	return true, service.fireReservedRun(ctx, reservation)
}

func (service *Service) dispatchRetryRun(ctx context.Context, runID string) (bool, error) {
	reservation, ok, err := service.store.ClaimRetryAutomationRun(ctx, runID, service.now())
	if err != nil || !ok {
		return false, err
	}
	return true, service.fireReservedRun(ctx, reservation)
}

func (service *Service) fireReservedRun(ctx context.Context, reservation schedulestore.ReservedRun) error {
	request := effectRequestForReservedRun(reservation)
	effect, err := service.writer.WriteScheduleEffect(ctx, request)
	if err != nil {
		return service.handleAttemptError(ctx, reservation.Run, err)
	}
	_, err = service.store.CompleteAutomationRun(ctx, schedulestore.AutomationRunCompletion{
		WorkspaceID: reservation.Run.WorkspaceID,
		ID:          reservation.Run.ID,
		Status:      runeffect.StatusSucceeded,
		Effect:      effect,
		Now:         service.now(),
	})
	return err
}

func effectRequestForReservedRun(reservation schedulestore.ReservedRun) EffectRequest {
	run := reservation.Run
	definition := reservation.Definition
	messageOptions := MessageOptionsFromMetadata(run.Metadata)
	return EffectRequest{
		WorkspaceID:     run.WorkspaceID,
		TargetKind:      run.TargetKind,
		TargetRef:       run.TargetRef,
		Content:         run.MessageTemplateSnapshot,
		AlsoStartThread: messageOptions.AlsoStartThread && run.TargetKind == schedulestore.TargetKindChannel,
		MentionHints:    messageOptions.MentionHints,
		Source: message.ContributionSourceForAutomationRun(message.AutomationRunContributionSourceParams{
			RunID:               run.ID,
			AutomationID:        run.AutomationID,
			AutomationName:      run.AutomationNameSnapshot,
			SourceType:          run.SourceType,
			SourceOccurrenceRef: run.SourceOccurrenceRef,
		}),
		ClientMessageID:   run.IdempotencyKey,
		OwnerMemberID:     run.OwnerMemberIDSnapshot,
		PrincipalMemberID: run.PrincipalMemberIDSnapshot,
		AuthorMemberID:    run.AuthorMemberIDSnapshot,
		Definition:        definition,
		Run:               run,
	}
}

func (service *Service) handleAttemptError(ctx context.Context, run schedulestore.AutomationRun, attemptErr error) error {
	now := service.now()
	if retryableFailure(attemptErr) && run.AttemptCount < service.maxAttempts {
		nextRetryAt := now.Add(service.retryBackoff * time.Duration(run.AttemptCount))
		_, err := service.store.MarkAutomationRunRetryWaiting(ctx, schedulestore.AutomationRunRetryUpdate{
			WorkspaceID: run.WorkspaceID,
			ID:          run.ID,
			NextRetryAt: nextRetryAt,
			ErrorCode:   schedulestore.FailureCodeTransientWriteError,
			ErrorDetail: attemptErr.Error(),
			Now:         now,
		})
		return err
	}

	code, block := classifyFailure(attemptErr)
	_, err := service.store.CompleteAutomationRun(ctx, schedulestore.AutomationRunCompletion{
		WorkspaceID:     run.WorkspaceID,
		ID:              run.ID,
		Status:          runeffect.StatusFailed,
		Error:           runeffect.Error{Code: code, Detail: attemptErr.Error()},
		BlockDefinition: block,
		Now:             now,
	})
	return err
}

func classifyFailure(err error) (string, bool) {
	switch {
	case errors.Is(err, storeerr.ErrNotFound):
		return schedulestore.FailureCodeTargetNotFound, true
	case errors.Is(err, storeerr.ErrForbidden):
		return schedulestore.FailureCodeForbiddenCurrentPerms, true
	case errors.Is(err, storeerr.ErrBadRequest):
		return schedulestore.FailureCodeContentInvalid, true
	default:
		return schedulestore.FailureCodeTransientWriteError, false
	}
}

func retryableFailure(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	var mysqlErr *mysql.MySQLError
	if errors.As(err, &mysqlErr) {
		switch mysqlErr.Number {
		case 1205, 1213, 9007:
			return true
		}
	}
	return !errors.Is(err, storeerr.ErrNotFound) &&
		!errors.Is(err, storeerr.ErrForbidden) &&
		!errors.Is(err, storeerr.ErrBadRequest)
}

func (service *Service) now() time.Time {
	if service == nil || service.clock == nil {
		return time.Now().UTC()
	}
	return service.clock().UTC()
}

func (service *Service) logWarn(message string, args ...any) {
	if service != nil && service.logger != nil {
		service.logger.Warn(message, args...)
	}
}

func (service *Service) logInfo(message string, args ...any) {
	if service != nil && service.logger != nil {
		service.logger.Info(message, args...)
	}
}

func ValidateEffectRequest(request EffectRequest) error {
	if request.WorkspaceID == "" || request.TargetRef == "" || request.Content == "" || request.ClientMessageID == "" {
		return fmt.Errorf("schedule effect request requires workspace, target, content, and client id")
	}
	if err := request.Source.ValidateForWrite(); err != nil {
		return fmt.Errorf("schedule effect source: %w", err)
	}
	return nil
}