jobsync

type Authorizer func(w http.ResponseWriter, r *http.Request) bool

type Bucket struct {
	At        time.Time
	Succeeded int64
	Failed    int64
}

type Client struct {
	// contains filtered or unexported fields
}

type ClientOption func(*Client)

type Counts struct {
	Scheduled, Enqueued, Running, Retrying, Succeeded, Dead, Cancelled int64
}

type DashboardConfig struct {
	// BasePath is where the dashboard is mounted, e.g. "/jobs". Links are built
	// relative to it, so it must match the mux pattern.
	BasePath string

	// Auth gates every request. Leaving it nil is the right thing on a laptop and
	// safe on a server: the dashboard then serves loopback clients only. Set it to
	// BasicAuth for a deployment, or AllowAll to deliberately open it up.
	Auth Authorizer
}

type Filter struct {
	States []State
	Queues []string
	Kinds  []string
	Tags   []string
	Search string // substring match over Kind, ID and LastError
	Since  time.Time
	Until  time.Time
	Offset int
	Limit  int
}

type Heartbeat interface {
	Heartbeat(ctx context.Context, info ServerInfo, ttl time.Duration) error
}

type Job struct {
	ID          string
	Kind        string // handler name, from Declare
	Queue       string
	Payload     []byte // JSON-encoded args
	Priority    int    // lower runs first
	Attempt     int    // completed attempts; 0 until the first failure
	MaxAttempts int
	State       State
	UniqueKey   string // empty means no deduplication
	Tags        []string
	LastError   string

	CreatedAt   time.Time
	ScheduledAt time.Time // when the job becomes fetchable
	LeasedUntil time.Time // meaningful only in StateRunning
	FinishedAt  time.Time // zero until the job reaches a terminal state
	Owner       string    // server ID holding the lease
}

type JobType[T any] struct {
	// contains filtered or unexported fields
}

type Locker interface {
	Lock(ctx context.Context, key string, ttl time.Duration) (token string, err error)
	Unlock(ctx context.Context, key, token string) error
}

type Monitor interface {
	Counts(ctx context.Context) (Counts, error)
	QueueStats(ctx context.Context) ([]QueueStat, error)
	Servers(ctx context.Context) ([]ServerInfo, error)

	// ListJobs returns a page of jobs and the total matching f.
	ListJobs(ctx context.Context, f Filter) ([]*Job, int64, error)
	JobHistory(ctx context.Context, id string) (*Job, []Transition, error)

	// Throughput returns succeeded and failed counts bucketed over [since, now].
	// Drivers that keep no history may return nil; the dashboard hides the graph.
	//
	// The window is bounded by whatever retention the driver is configured with:
	// a job deleted by the janitor is no longer counted. Asking for a year of
	// throughput from a store that keeps seven days of jobs returns seven days.
	Throughput(ctx context.Context, since time.Time, bucket time.Duration) ([]Bucket, error)

	// Requeue moves jobs back to StateEnqueued with Attempt reset. Delete removes
	// them outright. Both are no-ops for jobs currently in StateRunning.
	Requeue(ctx context.Context, ids []string) error
	Delete(ctx context.Context, ids []string) error
}

type Option func(*Job)

type QueueControl interface {
	PauseQueue(ctx context.Context, name string) error
	ResumeQueue(ctx context.Context, name string) error
	PausedQueues(ctx context.Context) ([]string, error)
}

type QueueStat struct {
	Name     string
	Enqueued int64
	Running  int64
	Paused   bool
	// OldestEnqueued is the age of the head of the queue, the number to alert on.
	OldestEnqueued time.Duration
}

type Result struct {
	State   State     // Succeeded, Retrying, Dead or Cancelled
	RetryAt time.Time // when State is Retrying
	Err     string
}

type Schedule struct {
	ID       string // caller-chosen, stable across deploys
	Kind     string
	Queue    string
	Payload  []byte
	Cron     string
	Timezone string
	LastRun  time.Time
	NextRun  time.Time
	Paused   bool
}

type Schedules interface {
	SaveSchedule(ctx context.Context, s Schedule) error
	ListSchedules(ctx context.Context) ([]Schedule, error)
	RemoveSchedule(ctx context.Context, id string) error

	// DueSchedules returns schedules with NextRun <= now and Paused false.
	DueSchedules(ctx context.Context, now time.Time) ([]Schedule, error)

	// AdvanceSchedule records that a run was fired. The caller holds the Locker
	// lock for the schedule, so this needs no atomicity of its own.
	AdvanceSchedule(ctx context.Context, id string, lastRun, nextRun time.Time) error

	// SetSchedulePaused stops or resumes a schedule without deleting it, which is
	// what an operator wants during an incident: the definition and its history
	// survive, and resuming does not re-fire whatever was missed.
	//
	// Setting it on an absent schedule is not an error.
	SetSchedulePaused(ctx context.Context, id string, paused bool) error
}

type Server struct {
	// contains filtered or unexported fields
}

type ServerConfig struct {
	// Queues are polled in the order given, so the first is the highest priority.
	Queues []string
	// Concurrency is the number of jobs this server runs at once.
	Concurrency int
	// Lease is how long a claimed job stays owned without a heartbeat. It bounds
	// how long a crashed server's work sits idle, so keep it well under the
	// patience of whoever is waiting for the job.
	Lease time.Duration
	// PollInterval is how long to wait after an empty fetch.
	PollInterval time.Duration
	// ScheduleInterval is how often recurring jobs are checked. Cron resolution
	// is one minute, so the default trades a little latency for far fewer
	// queries; lowering it does not make a cron job fire more often.
	ScheduleInterval time.Duration
	// ScheduleMaxAttempts is MaxAttempts for jobs created from a schedule.
	ScheduleMaxAttempts int
	// Backoff returns how long to wait before retrying a job that failed on
	// attempt n (1-based). Defaults to exponential with jitter, capped at an hour.
	Backoff func(attempt int) time.Duration
	Logger  *slog.Logger
}

type ServerInfo struct {
	ID          string
	Hostname    string
	Queues      []string
	Concurrency int
	StartedAt   time.Time
	HeartbeatAt time.Time
}

type State string

type Storage interface {
	// Enqueue persists jobs. A job whose ScheduledAt is in the future is stored
	// in StateScheduled and becomes fetchable when it comes due; otherwise it is
	// stored in StateEnqueued.
	//
	// A job whose ID already exists is ignored, silently. Callers rely on this
	// for idempotency: the scheduler derives a job id from (schedule, tick), so a
	// server crashing between enqueuing a tick and recording it refires the same
	// tick and the duplicate is a no-op instead of a second run.
	//
	// Ignored means ignored — not overwritten. Overwriting would resurrect a
	// finished job as a fresh one and run the work twice, which is the exact
	// failure the rule exists to prevent.
	//
	// When UniqueKey is set and a non-terminal job already holds that key, the
	// new job is dropped silently rather than returning an error.
	Enqueue(ctx context.Context, jobs []*Job) error

	// Fetch never returns a job from a paused queue, if the driver implements
	// QueueControl. The check belongs here and not in the caller: a paused queue
	// must be paused for every server, including one that has not noticed yet,
	// and a server filtering its own list would keep draining a queue an operator
	// just stopped.
	//
	// Fetch atomically claims at most n due jobs from queues, moving each to
	// StateRunning with a lease held by owner until now+lease. An empty result is
	// not an error.
	//
	// A job is due when its ScheduledAt has passed according to the STORAGE's
	// clock, never the calling process's. That is the only clock every server
	// shares, so it is the only one they can agree on. The cost is that a job
	// enqueued with ScheduledAt = time.Now() on a host whose clock runs ahead of
	// the storage is not instantly fetchable — it waits out the skew, then the
	// next poll. Callers that need a job to run now should not depend on it being
	// fetchable in the same millisecond it was enqueued.
	//
	// Jobs are returned in Priority then ScheduledAt order, best-effort within the
	// driver's priority granularity. SQL drivers honour Priority exactly. Redis
	// cannot: a sorted set score encodes one dimension, so ordering by priority
	// and filtering by due time at once needs a structure per priority level, and
	// an unbounded number of levels is unbounded by user input. Drivers with
	// coarse priority clamp it and document the range.
	Fetch(ctx context.Context, queues []string, n int, owner string, lease time.Duration) ([]*Job, error)

	// Extend renews the lease on jobs still running under owner. Jobs no longer
	// owned are skipped, not reported.
	Extend(ctx context.Context, ids []string, owner string, lease time.Duration) error

	// Finish applies r to a job owner holds. On Retrying the job returns to
	// StateScheduled with ScheduledAt set to r.RetryAt and Attempt incremented.
	//
	// Any terminal state (Succeeded, Dead, Cancelled) must also clear UniqueKey.
	// Otherwise a job keyed report:2026-10-05 holds that key forever once it
	// succeeds, and can never be enqueued again — including by a manual retry
	// from the dashboard.
	Finish(ctx context.Context, id, owner string, r Result) error

	// Reclaim returns jobs whose lease expired to StateEnqueued and reports how
	// many were recovered. Every server calls it periodically, so it must be
	// safe to run concurrently on every node.
	Reclaim(ctx context.Context) (int, error)

	Close() error
}

type Transition struct {
	State  State
	At     time.Time
	Reason string
}

type Tx interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

type TxEnqueuer interface {
	EnqueueTx(ctx context.Context, tx Tx, jobs []*Job) error
}