package invocation

import (
	"context"
	"fmt"
	"strings"
	"time"
)

const defaultScheduleDispatchBatchSize = 50

// ScheduleCandidate is an opaque store-issued handle for one due schedule
// occurrence. The store's private implementation may carry placement and
// generation fences in addition to these read-only diagnostic values. The
// dispatcher must only pass candidates back to the same materializer; it must
// never construct or reinterpret them.
type ScheduleCandidate interface {
	SourceID() string
	WorkspaceID() string
	OccurrenceAt() time.Time
}

// ListDueScheduleInput bounds one due-schedule scan at one stable instant.
type ListDueScheduleInput struct {
	Now   time.Time
	Limit int
}

// ScheduleMaterializeDisposition classifies a completed materialization
// attempt. These are durable control-flow outcomes, not transient errors.
type ScheduleMaterializeDisposition string

const (
	// ScheduleMaterializeCreated means a new Invocation and its initial Run
	// were accepted for the occurrence.
	ScheduleMaterializeCreated ScheduleMaterializeDisposition = "created"
	// ScheduleMaterializeReplayed means the occurrence was already accepted;
	// the existing immutable Invocation was returned idempotently.
	ScheduleMaterializeReplayed ScheduleMaterializeDisposition = "replayed"
	// ScheduleMaterializeStale means the source no longer owns the listed due
	// occurrence (for example, another dispatcher already advanced it).
	ScheduleMaterializeStale ScheduleMaterializeDisposition = "stale"
	// ScheduleMaterializeFenced means the candidate's placement or activation
	// generation changed after it was listed.
	ScheduleMaterializeFenced ScheduleMaterializeDisposition = "fenced"
	// ScheduleMaterializeBlocked means current durable state intentionally
	// prevents acceptance, while the pass itself completed without an
	// infrastructure error.
	ScheduleMaterializeBlocked ScheduleMaterializeDisposition = "blocked"
)

// ScheduleMaterializeResult is the classification returned after the store
// atomically revalidates a candidate, accepts or replays its Invocation, and
// advances the source cursor when appropriate.
type ScheduleMaterializeResult struct {
	Disposition ScheduleMaterializeDisposition
	// Acceptance is populated for created and replayed dispositions. It is
	// empty for stale, fenced, and blocked candidates.
	Acceptance AcceptResult
}

// ScheduledMaterializer is the persistence boundary used by the pure
// dispatcher pass. Implementations own placement fencing, transaction lock
// order, Invocation acceptance, idempotency, and schedule cursor movement.
type ScheduledMaterializer interface {
	ListDueScheduleCandidates(context.Context, ListDueScheduleInput) ([]ScheduleCandidate, error)
	MaterializeScheduled(context.Context, ScheduleCandidate, time.Time) (ScheduleMaterializeResult, error)
}

// ScheduleClock supplies one deterministic instant per dispatcher pass.
type ScheduleClock interface {
	Now() time.Time
}

type systemScheduleClock struct{}

func (systemScheduleClock) Now() time.Time { return time.Now().UTC() }

// ScheduleDispatcherOptions configures a single bounded dispatcher pass.
type ScheduleDispatcherOptions struct {
	// Clock defaults to the system UTC clock.
	Clock ScheduleClock
	// BatchSize is both the list limit and the maximum number of candidates
	// materialized by RunOnce. Values <= 0 use a default of 50.
	BatchSize int
}

// ScheduleDispatcher performs stateless, explicitly invoked schedule passes.
// It does not own a ticker, goroutine, lease loop, or background worker.
type ScheduleDispatcher struct {
	materializer ScheduledMaterializer
	clock        ScheduleClock
	batchSize    int
}

// NewScheduleDispatcher constructs a bounded schedule dispatcher.
func NewScheduleDispatcher(materializer ScheduledMaterializer, options ScheduleDispatcherOptions) (*ScheduleDispatcher, error) {
	if materializer == nil {
		return nil, fmt.Errorf("scheduled materializer is required")
	}
	clock := options.Clock
	if clock == nil {
		clock = systemScheduleClock{}
	}
	batchSize := options.BatchSize
	if batchSize <= 0 {
		batchSize = defaultScheduleDispatchBatchSize
	}
	return &ScheduleDispatcher{
		materializer: materializer,
		clock:        clock,
		batchSize:    batchSize,
	}, nil
}

// ScheduleDispatchOutcome reports the complete result of one pass. Candidate
// failures are aggregated in Errors and do not prevent later candidates in
// the same bounded batch from being attempted.
type ScheduleDispatchOutcome struct {
	// DueListed is the number of candidates returned by the store. It can be
	// greater than Attempted only if an implementation violates the requested
	// list limit; RunOnce still enforces its own hard bound.
	DueListed int
	// Attempted is the number of MaterializeScheduled calls made.
	Attempted int
	Created   int
	Replayed  int
	Stale     int
	Fenced    int
	Blocked   int
	Errors    []error
}

// RunOnce scans and processes at most one configured batch. It reads the
// clock once so listing and every materialization use the same DATETIME(6)
// instant. Only initialization/list failures are returned directly; isolated
// candidate failures are retained in the outcome.
func (dispatcher *ScheduleDispatcher) RunOnce(ctx context.Context) (ScheduleDispatchOutcome, error) {
	if dispatcher == nil || dispatcher.materializer == nil || dispatcher.clock == nil || dispatcher.batchSize <= 0 {
		return ScheduleDispatchOutcome{}, fmt.Errorf("schedule dispatcher is not initialized")
	}

	now := dispatcher.clock.Now().UTC().Truncate(time.Microsecond)
	if now.IsZero() {
		return ScheduleDispatchOutcome{}, fmt.Errorf("schedule dispatcher clock returned a zero time")
	}

	candidates, err := dispatcher.materializer.ListDueScheduleCandidates(ctx, ListDueScheduleInput{
		Now:   now,
		Limit: dispatcher.batchSize,
	})
	if err != nil {
		return ScheduleDispatchOutcome{}, fmt.Errorf("list due schedule candidates: %w", err)
	}

	outcome := ScheduleDispatchOutcome{DueListed: len(candidates)}
	if len(candidates) > dispatcher.batchSize {
		outcome.Errors = append(outcome.Errors, fmt.Errorf(
			"list due schedule candidates returned %d candidates for limit %d",
			len(candidates), dispatcher.batchSize,
		))
		candidates = candidates[:dispatcher.batchSize]
	}

	for index, candidate := range candidates {
		if candidate == nil {
			outcome.Errors = append(outcome.Errors, fmt.Errorf("schedule candidate %d is nil", index))
			continue
		}

		outcome.Attempted++
		result, err := dispatcher.materializer.MaterializeScheduled(ctx, candidate, now)
		if err != nil {
			outcome.Errors = append(outcome.Errors, fmt.Errorf("schedule candidate %s: %w", scheduleCandidateLabel(candidate), err))
			continue
		}

		switch result.Disposition {
		case ScheduleMaterializeCreated:
			outcome.Created++
		case ScheduleMaterializeReplayed:
			outcome.Replayed++
		case ScheduleMaterializeStale:
			outcome.Stale++
		case ScheduleMaterializeFenced:
			outcome.Fenced++
		case ScheduleMaterializeBlocked:
			outcome.Blocked++
		default:
			outcome.Errors = append(outcome.Errors, fmt.Errorf(
				"schedule candidate %s returned unknown materialization disposition %q",
				scheduleCandidateLabel(candidate), result.Disposition,
			))
		}
	}
	return outcome, nil
}

func scheduleCandidateLabel(candidate ScheduleCandidate) string {
	sourceID := strings.TrimSpace(candidate.SourceID())
	if sourceID == "" {
		sourceID = "<unknown-source>"
	}
	workspaceID := strings.TrimSpace(candidate.WorkspaceID())
	if workspaceID == "" {
		workspaceID = "<unknown-workspace>"
	}
	return workspaceID + "/" + sourceID
}