package automation

import (
	"context"
	"fmt"
	"log"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/backend/internal/domain/ports"
	"github.com/nexuscrm/backend/internal/features/metadata"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// SchedulerService handles scheduled flow execution
type SchedulerService struct {
	persistence           PersistenceProvider
	query                 ports.QueryService
	metadata              *metadata.MetadataService
	flowExecutor          *FlowExecutor
	loadScheduledInputsFn func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error)
	executeFlowVersionFn  func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error)
	// resumeInterviewFn + listDuePausedInterviewsFn provide test seams for
	// the Wait-element resume tick. Production wiring defers to the
	// flowExecutor (see resumePausedInterviews).
	resumeInterviewFn         func(ctx context.Context, id string, user *models.UserSession) error
	listDuePausedInterviewsFn func(ctx context.Context, now time.Time, limit int, user *models.UserSession) ([]string, error)
	// nowFn is an optional clock seam for tests. Production leaves it nil
	// (defaults to time.Now). Set once at construction; readers do not lock.
	nowFn        func() time.Time
	retryState   map[string]*scheduledFlowRetryState
	retryStateMu sync.Mutex
	stopCh       chan struct{}
	wg           sync.WaitGroup
	running      bool
	mu           sync.Mutex
}

// schedulerBackoffLadder is per-FlowDefinition self-protection: when a
// Scheduled flow keeps failing tick after tick, the wait between attempts
// grows so a single misconfigured flow (broken FlowDefinition.ActiveVersionId,
// stale FK, etc.) cannot self-DDoS the log + DB. Issue #2199.
var schedulerBackoffLadder = []time.Duration{
	1 * time.Minute,
	5 * time.Minute,
	15 * time.Minute,
	1 * time.Hour,
	24 * time.Hour,
}

// schedulerAlertStageIndex marks the ladder index past which scheduled-flow
// failures should be considered alert-worthy. Logs gain an "alert=true"
// marker for downstream log-pipeline scraping.
const schedulerAlertStageIndex = 3 // 1h and beyond

// scheduledFlowRetryState tracks per-flow consecutive-failure backoff. State
// is process-local self-protection (not a parallel SSoT for SF's
// AsyncApexJob); it is intentionally not persisted across restart — a real
// broken flow re-enters the ladder naturally on the next boot, and the
// alert at stage schedulerAlertStageIndex (1h) usually fires well before any
// restart cadence.
type scheduledFlowRetryState struct {
	consecutiveFailures int
	nextRetryAt         time.Time
	lastLoggedStage     int // -1 = never logged
}

// tickPhase classifies the outcome of a single executeScheduledDefinition
// call so the retry-state writer can decide whether to advance the ladder
// (allFailed / loadFailed), reset it (allSucceeded / mixed — the conservative
// reset on partial success preserves liveness for partially-broken flows),
// or leave it untouched (inputsEmpty — nothing was attempted).
type tickPhase int

const (
	phaseLoadFailed tickPhase = iota
	phaseInputsEmpty
	phaseExecAllFailed
	phaseExecMixed
	phaseExecAllSucceeded
)

// NewSchedulerService creates a new SchedulerService
func NewSchedulerService(
	persistence PersistenceProvider,
	query ports.QueryService,
	metadata *metadata.MetadataService,
	flowExecutor *FlowExecutor,
) *SchedulerService {
	return &SchedulerService{
		persistence:  persistence,
		query:        query,
		metadata:     metadata,
		flowExecutor: flowExecutor,
		stopCh:       make(chan struct{}),
	}
}

// StartScheduler starts the background scheduler
func (s *SchedulerService) StartScheduler() {
	s.mu.Lock()
	if s.running {
		s.mu.Unlock()
		return
	}
	s.running = true
	s.stopCh = make(chan struct{})
	s.mu.Unlock()

	// Drop any retry-state carried over from a previous Start/Stop cycle so
	// in-process restarts (tests, future hot-reload) start clean.
	s.retryStateMu.Lock()
	s.retryState = nil
	s.retryStateMu.Unlock()

	s.wg.Add(1)
	go s.runSchedulerLoop()
}

// now returns the wall clock unless a test seam is wired.
func (s *SchedulerService) now() time.Time {
	if s.nowFn != nil {
		return s.nowFn()
	}
	return time.Now()
}

// effectiveStateKey returns a stable per-flow-version key for retry-state
// lookup. Composite of FlowDefinition Id (or DeveloperName fallback for
// legacy rows seen on localhost — see issue #2199 Notes) and the active
// FlowVersion Id, so that an admin who fixes a broken Scheduled flow by
// activating a new version is not suppressed by the prior version's
// backoff window. Returns "" when neither def-key nor version-key can be
// derived so the caller skips retry tracking entirely.
func effectiveStateKey(def *models.FlowDefinition, version *models.FlowVersion) string {
	if def == nil {
		return ""
	}
	defPart := def.Id
	if defPart == "" {
		defPart = def.DeveloperName
	}
	if defPart == "" {
		return ""
	}
	if version != nil && version.Id != "" {
		return defPart + ":" + version.Id
	}
	return defPart
}

// shouldSkipForBackoff returns true when the flow is currently in a backoff
// window from prior consecutive failures. Returns false (i.e. proceed) when
// there is no key, no state, or the wait has elapsed.
func (s *SchedulerService) shouldSkipForBackoff(def *models.FlowDefinition, version *models.FlowVersion) bool {
	key := effectiveStateKey(def, version)
	if key == "" {
		return false
	}
	s.retryStateMu.Lock()
	defer s.retryStateMu.Unlock()
	state := s.retryState[key]
	if state == nil {
		return false
	}
	return s.now().Before(state.nextRetryAt)
}

// recordTickOutcome updates the per-flow-version retry state after one
// tick. The caller has already done all logging for this tick's per-input
// attempts; this helper only emits a single "stage transition" log line
// per ladder advance so the storm-of-stage-N-logs pattern can never recur.
func (s *SchedulerService) recordTickOutcome(def *models.FlowDefinition, version *models.FlowVersion, phase tickPhase) {
	key := effectiveStateKey(def, version)
	if key == "" {
		return
	}
	s.retryStateMu.Lock()
	defer s.retryStateMu.Unlock()
	if s.retryState == nil {
		s.retryState = make(map[string]*scheduledFlowRetryState)
	}
	state := s.retryState[key]
	if state == nil {
		state = &scheduledFlowRetryState{lastLoggedStage: -1}
		s.retryState[key] = state
	}

	switch phase {
	case phaseInputsEmpty:
		// Nothing was attempted — neutral.
		return
	case phaseExecAllSucceeded, phaseExecMixed:
		// Any progress resets the ladder so a flow that recovers after a
		// blip (or has one stubborn input among many healthy ones) doesn't
		// get throttled forever.
		if state.consecutiveFailures > 0 || state.lastLoggedStage >= 0 {
			log.Printf("Scheduler: flow '%s' recovered after %d consecutive failure(s); retry backoff cleared",
				def.DeveloperName, state.consecutiveFailures)
		}
		state.consecutiveFailures = 0
		state.nextRetryAt = time.Time{}
		state.lastLoggedStage = -1
		return
	case phaseLoadFailed, phaseExecAllFailed:
		state.consecutiveFailures++
		stage := state.consecutiveFailures - 1
		if stage >= len(schedulerBackoffLadder) {
			stage = len(schedulerBackoffLadder) - 1
		}
		state.nextRetryAt = s.now().Add(schedulerBackoffLadder[stage])
		if stage != state.lastLoggedStage {
			alert := ""
			if stage >= schedulerAlertStageIndex {
				alert = " alert=true"
			}
			log.Printf("Scheduler: flow '%s' has failed %d consecutive tick(s); next retry deferred by %s%s",
				def.DeveloperName, state.consecutiveFailures, schedulerBackoffLadder[stage], alert)
			state.lastLoggedStage = stage
		}
	}
}

// StopScheduler stops the background scheduler
func (s *SchedulerService) StopScheduler() {
	s.mu.Lock()
	if !s.running {
		s.mu.Unlock()
		return
	}
	s.running = false
	close(s.stopCh)
	s.mu.Unlock()

	s.wg.Wait()
}

func (s *SchedulerService) runSchedulerLoop() {
	defer s.wg.Done()

	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-s.stopCh:
			log.Println("Scheduler stopped")
			return
		case <-ticker.C:
			s.executeScheduledFlows()
			s.resumePausedInterviews()
		}
	}
}

// resumePausedInterviewsBatchSize bounds how many due interviews one tick
// will process. Keeps a single stuck scheduler tick from blocking for the
// whole table if catch-up after downtime produces a very large backlog.
const resumePausedInterviewsBatchSize = 200

// resumePausedInterviews loads all Paused FlowInterview rows whose
// PausedUntil has elapsed (catch-up safe: past-dated rows resume
// immediately) and dispatches each to the executor. Each resume runs in
// its own goroutine with panic recovery so one bad flow cannot stall the
// rest of the tick.
func (s *SchedulerService) resumePausedInterviews() {
	listFn := s.listDuePausedInterviewsFn
	resumeFn := s.resumeInterviewFn
	if listFn == nil || resumeFn == nil {
		// Default wiring (production): defer to flowExecutor + interview service.
		if s.flowExecutor == nil || s.flowExecutor.interviews == nil {
			return
		}
		if listFn == nil {
			listFn = s.flowExecutor.interviews.ListDuePausedInterviews
		}
		if resumeFn == nil {
			resumeFn = s.flowExecutor.ResumeInterview
		}
	}

	ctx := constants.WithSecurityContext(context.Background(), constants.SystemContext, "flow interview resume")
	systemUser := newSystemAutomationUser()

	ids, err := listFn(ctx, time.Now(), resumePausedInterviewsBatchSize, systemUser)
	if err != nil {
		log.Printf("Scheduler: failed to list due paused interviews: %v", err)
		return
	}
	if len(ids) == 0 {
		return
	}

	var wg sync.WaitGroup
	sem := make(chan struct{}, 10) // cap concurrent resumes
	for _, id := range ids {
		wg.Add(1)
		sem <- struct{}{}
		go func(interviewID string) {
			defer wg.Done()
			defer func() { <-sem }()
			defer func() {
				if r := recover(); r != nil {
					log.Printf("Scheduler: panic resuming interview %s: %v", interviewID, r)
				}
			}()
			if err := resumeFn(ctx, interviewID, systemUser); err != nil {
				log.Printf("Scheduler: failed to resume interview %s: %v", interviewID, err)
			}
		}(id)
	}
	wg.Wait()
}

func (s *SchedulerService) executeScheduledFlows() {
	ctx := context.Background()

	// Get all scheduled flow definitions
	flowDefs := s.metadata.GetScheduledFlowDefinitions(ctx)
	if len(flowDefs) == 0 {
		return
	}

	// Scheduled flows run as trusted system automation, not as an end user.
	systemUser := newSystemAutomationUser()

	now := time.Now()

	for _, def := range flowDefs {
		if def.ActiveVersionId == "" {
			continue
		}

		// Get the flow version to check schedule
		version := s.metadata.GetFlowVersionById(ctx, def.ActiveVersionId)
		if version == nil {
			continue
		}

		// Check if the flow should run based on schedule
		shouldRun := s.shouldRunFlow(version, now)
		if !shouldRun {
			continue
		}

		log.Printf("Scheduler: executing scheduled flow '%s'", def.DeveloperName)
		s.executeScheduledDefinition(ctx, def, version, systemUser)
	}
}

func (s *SchedulerService) executeScheduledDefinition(ctx context.Context, def *models.FlowDefinition, version *models.FlowVersion, systemUser *models.UserSession) {
	ctx = constants.WithSecurityContext(ctx, constants.SystemContext, "scheduled flow execution")

	if s.shouldSkipForBackoff(def, version) {
		// Silent — the stage-transition log was already emitted when the
		// backoff window opened. Issue #2199.
		return
	}

	inputSets, err := s.loadScheduledInputs(ctx, version, systemUser)
	if err != nil {
		log.Printf("Scheduler: failed to load scheduled inputs for flow '%s': %v", def.DeveloperName, err)
		s.recordTickOutcome(def, version, phaseLoadFailed)
		return
	}

	if len(inputSets) == 0 {
		log.Printf("Scheduler: no matching records for scheduled flow '%s'", def.DeveloperName)
		s.recordTickOutcome(def, version, phaseInputsEmpty)
		return
	}

	successes, failures := 0, 0
	for _, inputs := range inputSets {
		result, execErr := s.executeScheduledVersion(ctx, version, def.DeveloperName, inputs, systemUser)
		// Both an explicit error AND a result that signals !IsSuccess count
		// as a failure for backoff accounting; without the latter, soft
		// failures (validation errors carried inside the result) would
		// silently re-create the storm shape this fix targets.
		if execErr != nil || (result != nil && !result.IsSuccess) {
			failures++
			if execErr != nil {
				log.Printf("Scheduler: failed to execute flow '%s': %v", def.DeveloperName, execErr)
			} else if len(result.Errors) > 0 {
				log.Printf("Scheduler: flow '%s' completed with errors: %s", def.DeveloperName, result.Errors[0].Message)
			}
			continue
		}
		successes++
		log.Printf("Scheduler: flow '%s' executed successfully", def.DeveloperName)
	}

	switch {
	case failures == 0:
		s.recordTickOutcome(def, version, phaseExecAllSucceeded)
	case successes == 0:
		s.recordTickOutcome(def, version, phaseExecAllFailed)
	default:
		s.recordTickOutcome(def, version, phaseExecMixed)
	}
}

func (s *SchedulerService) loadScheduledInputs(ctx context.Context, version *models.FlowVersion, user *models.UserSession) ([]map[string]interface{}, error) {
	if s.loadScheduledInputsFn != nil {
		return s.loadScheduledInputsFn(ctx, version, user)
	}
	if s.flowExecutor == nil {
		return nil, fmt.Errorf("flow executor is not configured")
	}
	return s.flowExecutor.buildScheduledRunInputs(ctx, version, user)
}

func (s *SchedulerService) executeScheduledVersion(ctx context.Context, version *models.FlowVersion, flowAPIName string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	if s.executeFlowVersionFn != nil {
		return s.executeFlowVersionFn(ctx, version, flowAPIName, inputs, user)
	}
	if s.flowExecutor == nil {
		return nil, fmt.Errorf("flow executor is not configured")
	}
	return s.flowExecutor.executeFlowVersion(ctx, version, flowAPIName, inputs, user)
}

// shouldRunFlow checks if a scheduled flow should run at the given time.
//
// Schedule startTime strings (e.g. "05:00", "05:00:00", "05:00:00.000Z") are
// interpreted as UTC time-of-day; startDate is interpreted as a UTC calendar
// date. The comparison is performed in UTC so the trigger fires at the same
// wall-clock instant regardless of the backend host's local timezone (issue
// #2232).
func (s *SchedulerService) shouldRunFlow(version *models.FlowVersion, now time.Time) bool {
	// Parse metadata to check schedule
	if version.Metadata.Start == nil || version.Metadata.Start.Schedule == nil {
		return false
	}

	// Schedule strings are UTC-anchored — normalize the comparison clock to
	// UTC so Hour/Minute/Weekday read against the same wall-clock as the
	// metadata author intended.
	now = now.UTC()

	schedule := version.Metadata.Start.Schedule

	// Parse startDate (format: YYYY-MM-DD)
	var startDate time.Time
	if schedule.StartDate != "" {
		parsed, err := time.Parse("2006-01-02", schedule.StartDate)
		if err == nil {
			startDate = parsed
		}
	}

	// Don't run if before start date
	if !startDate.IsZero() && now.Before(startDate) {
		return false
	}

	// Parse startTime (format: HH:MM:SS or HH:MM)
	var targetHour, targetMinute int
	if schedule.StartTime != "" {
		parts := []int{}
		for _, p := range strings.Split(schedule.StartTime, ":") {
			if n, err := strconv.Atoi(p); err == nil {
				parts = append(parts, n)
			}
		}
		if len(parts) >= 2 {
			targetHour = parts[0]
			targetMinute = parts[1]
		}
	}

	// Check frequency
	switch schedule.Frequency {
	case "Once":
		// Run once at the scheduled time (within a 1-minute window).
		// Target instant is built in UTC because startTime / startDate are
		// UTC-anchored (see issue #2232).
		if !startDate.IsZero() && schedule.StartTime != "" {
			targetTime := time.Date(startDate.Year(), startDate.Month(), startDate.Day(),
				targetHour, targetMinute, 0, 0, time.UTC)
			// Run within a 1-minute window
			return now.After(targetTime) && now.Before(targetTime.Add(2*time.Minute))
		}
		return false

	case "Daily":
		// Run daily at the scheduled time (within a 1-minute window)
		return now.Hour() == targetHour && now.Minute() == targetMinute

	case "Weekly":
		// Run weekly on the same day as startDate at the scheduled time
		if !startDate.IsZero() {
			return now.Weekday() == startDate.Weekday() &&
				now.Hour() == targetHour && now.Minute() == targetMinute
		}
		return false

	default:
		return false
	}
}