triggers

type Admin interface {
	Status(ctx context.Context) (DispatcherStatus, error)
	// Gaps lists the gaps of a realm (those of every realm included: a window gap covers all),
	// newest first; resolved ones only when all is true.
	Gaps(ctx context.Context, realm string, all bool, limit int) ([]GapRecord, error)
	// ResolveGap closes an open gap of the realm; ok is false when there is none by that id.
	ResolveGap(ctx context.Context, realm, id string, at time.Time) (ok bool, err error)
}

type Dispatcher struct {
	// Source opens the change streams; State and Queue are the durable parts.
	Source changefeed.Source
	State  State
	Queue  Queue
	// Holder names this instance in the lease; it must be unique per process.
	Holder string
	// Config returns a realm's app configuration, for conditions that use appConfig; may be
	// nil. reg is the configuration the dispatcher is running.
	Config func(ctx context.Context, reg *registry.Registry, realm string) map[string]any
	// Limits returns a realm's trigger limits (the instance's, lowered by the realm's); nil
	// is registry.DefaultTriggerLimits for every realm.
	Limits   func(reg *registry.Registry, realm string) registry.TriggerLimits
	Observer Observer
	Log      *slog.Logger
	Now      func() time.Time

	LeaseTTL       time.Duration
	FlushEvery     time.Duration
	WatermarkEvery time.Duration
	BatchMax       int
	// Backoff is the wait between attempts to queue a batch that failed; nil is 100ms
	// doubling to 5s.
	Backoff func(attempt int) time.Duration
	// contains filtered or unexported fields
}

type DispatcherStatus struct {
	// Found is false before any instance has ever dispatched.
	Found bool
	// Held is whether some instance holds the lease at the moment.
	Held         bool
	Epoch        int64
	Holder       string
	LeaseExpires time.Time
	// WatermarkAt is when the position was last stored: its age is what to watch.
	WatermarkAt time.Time
}

type DocSource interface {
	// Count says how many documents were updated in [since, until).
	Count(ctx context.Context, mongoDatabase, collection string, since, until time.Time) (int64, error)
	// Updated calls fn for each document updated in [since, until), oldest first, in API shape.
	Updated(ctx context.Context, mongoDatabase, collection string, since, until time.Time, fn func(doc map[string]any) error) error
}

type ErrRescanNeedsConfirm struct{ Documents int64 }

type Gap struct {
	Realm      string
	Collection string // "" for a window gap: it covers every collection
	Trigger    string // "" when it is not about one trigger
	Reason     string
	From, To   time.Time
	Count      int
	Keys       []string // document ids, at most MaxGapKeys
}

type GapRecord struct {
	Gap
	ID         string
	RecordedAt time.Time
	// ResolvedAt is zero while the gap is open.
	ResolvedAt time.Time
}

type Job struct {
	// ID is derived from the change and the trigger, so a replay of the change produces the
	// same job and the queue collapses it.
	ID        string
	Realm     string
	Database  string // the function's database
	Function  string // the function's name
	Input     json.RawMessage
	Origin    string // trigger:<database>/<collection>/<trigger>
	Timeout   time.Duration
	RequestID string
	EventID   string
	// ChainRoot and ChainDepth place the run in a chain of writes caused by trigger runs: its
	// own writes are stamped with them.
	ChainRoot  string
	ChainDepth int
	// SerialKey and SerialOrder make an `ordered` trigger's jobs for one document run one at a
	// time, in the order the changes were made (the cluster time of the change).
	SerialKey   string
	SerialOrder int64
}

type Lease struct {
	Holder string
	Epoch  int64
}

type Observer interface {
	// Event counts a change a trigger considered, by outcome: matched, no_match, missing_preimage or error.
	Event(realm, collection, trigger, outcome string)
	// Enqueued counts the jobs queued and the replays the queue collapsed.
	Enqueued(realm string, created, duplicates int)
	// Lag is the time from a change to the jobs it made being queued.
	Lag(d time.Duration)
	// Watermark records when the position was last stored.
	Watermark(at time.Time)
	// Lease records whether this instance holds the lease, and its epoch.
	Lease(held bool, epoch int64)
	// Gap counts a gap recorded.
	Gap(realm, reason string)
	// Dropped counts an event that matched and was not queued, by reason: chain (too deep) or
	// burst (a chained loop above its rate).
	Dropped(realm, collection, trigger, reason string)
	// Backlog records a realm's queued trigger jobs.
	Backlog(realm string, n int)
}

type Queue interface {
	// Enqueue inserts the jobs of one realm. A job whose ID exists already is a replay and
	// is not an error: created counts the new ones.
	Enqueue(ctx context.Context, realm string, jobs []Job) (created int, err error)
	// Backlog counts the realm's trigger jobs that are queued or running: what the
	// dispatcher keeps below the realm's ceiling.
	Backlog(ctx context.Context, realm string) (int, error)
}

type RescanRequest struct {
	Realm, Database, Collection string
	// Trigger limits it to one trigger; "" is every one that can be rescanned.
	Trigger      string
	Since, Until time.Time
	// Confirm allows a scan over RescanConfirmOver documents.
	Confirm bool
	// Run identifies this rescan in the job ids, so its jobs don't collapse with the originals.
	Run string
}

type RescanResult struct {
	Scanned    int
	Queued     int
	Duplicates int
	// NotRescannable names the triggers left out and why.
	NotRescannable map[string]string
}

type State interface {
	// AcquireLease takes the lease when nobody holds it or the holder's has expired, with
	// the database's clock. ok is false when someone else holds it.
	AcquireLease(ctx context.Context, holder string, ttl time.Duration) (l Lease, ok bool, err error)
	// RenewLease extends it; ok is false when it is no longer this lease.
	RenewLease(ctx context.Context, l Lease, ttl time.Duration) (ok bool, err error)
	// ReleaseLease gives it up so another instance may take it at once.
	ReleaseLease(ctx context.Context, l Lease) error
	// Watermark returns the stored position and when it was stored; found is false before
	// the first.
	Watermark(ctx context.Context) (pos changefeed.Position, at time.Time, found bool, err error)
	// SetWatermark stores the position, only while l is still the lease; ok is false when
	// it is not (the caller must stop).
	SetWatermark(ctx context.Context, l Lease, pos changefeed.Position, at time.Time) (ok bool, err error)
	// RecordGap stores a gap for an operator to see.
	RecordGap(ctx context.Context, g Gap) error
}