store

package
v0.0.0-...-8584414 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 11, 2026 License: AGPL-3.0 Imports: 20 Imported by: 0

Documentation

Overview

Package store persists controller state (nodes, jobs, flows) in SQLite or Postgres, selected by DSN.

Index

Constants

View Source 
const (
	// LockKeyScanner guards scanner.RunLoop (scripts-share discovery).
	LockKeyScanner int64 = 1001
	// LockKeyRetention guards the hourly job-history prune loop.
	LockKeyRetention int64 = 1002
	// LockKeyDigest guards the hourly Discord digest flush loop.
	LockKeyDigest int64 = 1003
	// LockKeyDiskAlert guards the disk-alert cooldown sweep. (The per-node
	// alert itself fires inline on the heartbeat path; this key guards any
	// future periodic sweep. Kept reserved so the key table is complete.)
	LockKeyDiskAlert int64 = 1004
	// LockKeyBackup guards the scheduled DB-snapshot loop.
	LockKeyBackup int64 = 1005
	// LockKeyNodeAlert guards the node online/offline alert loop. Each tick
	// diffs node liveness against an in-memory baseline and sends Discord
	// alerts on transitions (online→offline, offline→online).
	LockKeyNodeAlert int64 = 1006
)

LockKey constants are distinct int64 keys for each background loop that must run as a singleton across controller replicas. They are passed to TryAdvisoryLock so two replicas never run the same loop concurrently.

The values are arbitrary but permanent — a running cluster may have rows in pg_locks with these obj/subID pairs, so changing a key mid-life would orphan an existing lock (the old holder keeps it under the old key; the new holder can acquire a different key and both run). Never renumber.

Variables

View Source 
var ErrJobNotFinishable = errors.New("job is not in a finishable state (already terminal/cancelled)")

ErrJobNotFinishable is returned by FinishJobWithReport when the job is no longer in an assignable/running state — a concurrent cancel (or a prior completion that raced the Terminal() pre-check) already moved the job out of 'assigned'/'running'. Callers use this to answer idempotently instead of erroring: the job is already in a terminal state, so the completion report is a no-op, not a failure.

Functions

func RunLoopSupervisor 

func RunLoopSupervisor(ctx context.Context, st *Store, log *slog.Logger, key int64, lockName string, body func(ctx context.Context))

RunLoopSupervisor wraps a background loop so it runs on exactly one replica at a time on Postgres (advisory lock), and runs unchanged on SQLite (the lock is a no-op always-succeed). The wrapped loop body runs while the lock is held; on lock loss or ctx cancel it exits and the supervisor re-acquires after reAcquireDelay, until ctx is cancelled.

The body runs under a sub-context derived from ctx. When the lock is lost (keepalive fails on PG) OR the parent ctx is cancelled (SIGTERM), the sub-context is cancelled so the body stops cleanly; the supervisor waits for the body goroutine to return before releasing and (on loss only) retrying acquisition. SIGTERM stops both the body and the supervisor.

On SQLite the lock is always held and never lost, so the supervisor is a transparent wrapper: the body runs until the parent ctx is cancelled, identical to the pre-HA `go body(ctx)` it replaces.

lockName is a human label for logs only; it does not affect the lock key.

Types

type SeriesProgress 

type SeriesProgress struct {
	Done   int
	Failed int
	Active int
	Seen   int
}

SeriesProgress holds the per-series encode progress counts derived from the jobs table. EpisodesDone counts distinct episode_dir values with at least one 'done' job; EpisodesFailed counts distinct episode_dirs whose jobs are all 'failed' (an episode that eventually succeeded is done, not failed); EpisodesActive counts distinct episode_dirs with an in-flight job (pending/assigned/running); EpisodesSeen is the distinct episode_dir count across ALL statuses — used as the fallback total when no scaffolded folders are readable on disk.

type Store 

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

Store wraps the database with typed operations. The backend is SQLite or Postgres, selected by the DSN passed to Open. Queries written with ? placeholders are rebound to $N on Postgres by the q() helper; every query in the package routes through q() so the dialect split lives in one place.

func Open 

func Open(path string) (*Store, error)

Open opens (creating if needed) a SQLite database at the given file path or a Postgres database at the given DSN, and applies migrations. A DSN starting with "postgres://" or "postgresql://" selects Postgres via the pgx stdlib driver (pool size 10); anything else is treated as a SQLite file path and opened with modernc.org/sqlite exactly as before — SQLite behavior is byte-for-byte unchanged (single-writer, DELETE journal, pragmas applied per-connection).

The SQLite journal_mode is DELETE (rollback journal), NOT WAL: on the Docker host's Ceph-backed volume the WAL/shm path corrupts its read view — sessions written through the WAL become invisible to subsequent lookups (reproduced on a pristine DB with this exact binary; the incident of 2026-08-28 turned every login into a 401). The rollback journal uses no shm/mmap and is the safe choice for a single-writer, low-volume store.

func (*Store) APITokenByHash 

func (s *Store) APITokenByHash(ctx context.Context, hash string) (*model.APIToken, error)

APITokenByHash looks a token up by its hash. Returns (nil, nil) when unknown so callers can treat it as an auth failure without an error path.

func (*Store) ActiveJobForNode 

func (s *Store) ActiveJobForNode(ctx context.Context, nodeID int64) (*model.Job, error)

ActiveJobForNode returns the node's assigned/running job, if any.

func (*Store) ActiveJobsForNode 

func (s *Store) ActiveJobsForNode(ctx context.Context, nodeID int64) ([]*model.Job, error)

ActiveJobsForNode returns all assigned/running jobs for a node (the set the controller needs for multi-job orphan recovery).

func (*Store) AppendAudit 

func (s *Store) AppendAudit(ctx context.Context, actor, action, object, detail string) error

AppendAudit records one audit event. Fire-and-forget by contract: the caller logs errors but never fails the audited operation because the audit write failed.

func (*Store) AppendProvisionRunLog 

func (s *Store) AppendProvisionRunLog(ctx context.Context, id int64, chunk string, capBytes int) error

AppendProvisionRunLog atomically appends output to the run's log column. SQLite's || concat handles the append; the UI tails via GetProvisionRun. When capBytes > 0 the persisted log is bounded to the last capBytes bytes (substr with a negative offset keeps the tail — where the failure is) so a long run cannot grow the row without limit.

func (*Store) AssignJob 

func (s *Store) AssignJob(ctx context.Context, jobID, nodeID int64) error

AssignJob atomically hands a pending job to a node. It fails if the node already has an active (assigned/running) job — the one-job-per-node rule.

Race-safety under active/active replicas (Phase 3 HA):

The capacity check SELECTs the node row FOR UPDATE on Postgres (a no-op on SQLite). This serializes the per-node capacity check across replicas: two controllers dispatching DIFFERENT jobs to the SAME node at once cannot both pass the active-count guard and overshoot max_concurrent_jobs — the second tx blocks on the row lock until the first commits, then re-reads the now- higher active count and rejects. SQLite's single-writer pool already serializes every transaction, so forUpdate is a no-op there.

The job flip itself relies on the conditional UPDATE + RowsAffected==1 guard below: `UPDATE jobs SET status='assigned' … WHERE status='pending'` is atomic on both backends, so even if two replicas select the same pending job (NextAssignableJobForNode is an unlocked read), exactly one UPDATE matches (RowsAffected==1) and the other gets 0 rows → error → the handler breaks the assignment loop and re-evaluates on the next heartbeat. This is why SKIP LOCKED on NextAssignableJob* is NOT needed: the handler's select-then-AssignJob pattern already de-races via the conditional UPDATE, and a lost race is a clean error + retry, never a double-assignment. See the 3.2 decision note in ha_notes.go.

func (*Store) AvgFlowDuration 

func (s *Store) AvgFlowDuration(ctx context.Context, flowID int64) (avgSec float64, samples int, err error)

AvgFlowDuration returns the average wall-clock duration (seconds) and sample count of DONE jobs on a flow — the ETA basis. Only done jobs count: failed/cancelled runs have partial durations that would skew the average low. (0, 0, nil) means no history yet; callers gate confidence on the sample count.

func (*Store) BackupTo 

func (s *Store) BackupTo(ctx context.Context, dest string) error

BackupTo writes a consistent snapshot of the database to dest using SQLite's VACUUM INTO — an online, transaction-safe copy that needs no Fails if dest already exists so a caller bug can't clobber a previous good backup.

Postgres note: this method is SQLite-only (VACUUM INTO). On Postgres it returns an error — the PG backup path (pg_dump or COPY-based dump) is a separate task. The API layer's backup endpoint must branch on backend.

func (*Store) CancelJob 

func (s *Store) CancelJob(ctx context.Context, id int64) (int64, error)

CancelJob marks a pending/assigned job cancelled and returns rows affected. Terminal and running jobs are refused so the lifecycle cannot regress.

func (*Store) Close 

func (s *Store) Close() error

Close releases the database handle.

func (*Store) ConsumePairingCode 

func (s *Store) ConsumePairingCode(ctx context.Context, codeHash string, nodeID int64) (*model.PairingCode, error)

ConsumePairingCode atomically marks a code used by nodeID. It fails when the code is unknown, expired, or already consumed (one-shot semantics).

func (*Store) CountActiveJobs 

func (s *Store) CountActiveJobs(ctx context.Context) (int, error)

CountActiveJobs returns the number of jobs that are pending, assigned, or running — i.e. not in a terminal state. Used by the backup-restore guard to refuse a restore while work is in flight.

func (*Store) CountFinishedTasksForNode 

func (s *Store) CountFinishedTasksForNode(ctx context.Context, nodeID int64) (int, error)

CountFinishedTasksForNode counts terminal jobs completed by a node — used only for display; the reboot counter itself comes from the agent heartbeat.

func (*Store) CountJobsByStatus 

func (s *Store) CountJobsByStatus(ctx context.Context) (map[string]int, error)

CountJobsByStatus returns the job count per status in one grouped query (Prometheus exposition + dashboards). Keys are the raw status strings.

func (*Store) CountUsersByRole 

func (s *Store) CountUsersByRole(ctx context.Context, role string) (int, error)

CountUsersByRole counts accounts with the given role. Used by the last-admin guard before a delete or role downgrade.

func (*Store) CreateAPIToken 

func (s *Store) CreateAPIToken(ctx context.Context, tok *model.APIToken) error

CreateAPIToken inserts a token row and fills ID/CreatedAt. Name and hash uniqueness are enforced by the table constraints (duplicate name returns an error so callers can 409).

func (*Store) CreateFlow 

func (s *Store) CreateFlow(ctx context.Context, f *model.Flow) (*model.Flow, error)

CreateFlow inserts a new flow.

func (*Store) CreateJob 

func (s *Store) CreateJob(ctx context.Context, j *model.Job) (*model.Job, error)

CreateJob inserts a new pending job. Priority is honored so the queue can dispatch higher-priority jobs first; the other v2 fields (full_log, step_timings_json, retry_count, next_retry_at) default in the schema and are populated later by the completion/report paths.

func (*Store) CreateNode 

func (s *Store) CreateNode(ctx context.Context, name, tokenHash string) (*model.Node, error)

CreateNode registers a node with a bcrypt hash of its token.

func (*Store) CreatePairingCode 

func (s *Store) CreatePairingCode(ctx context.Context, codeHash, nameHint string, ttl time.Duration) (*model.PairingCode, error)

CreatePairingCode stores a hashed one-shot code expiring after ttl.

func (*Store) CreateProvisionRun 

func (s *Store) CreateProvisionRun(ctx context.Context, pr *model.ProvisionRun) (*model.ProvisionRun, error)

CreateProvisionRun inserts a new run row and returns it with its ID.

func (*Store) CreateSession 

func (s *Store) CreateSession(ctx context.Context, tokenHash string, userID int64, expiresAt time.Time) error

CreateSession stores a hashed session token with its expiry.

func (*Store) CreateShare 

func (s *Store) CreateShare(ctx context.Context, sh *model.Share) (*model.Share, error)

CreateShare inserts a share, encrypting Password at rest. The returned struct keeps the plaintext password (the caller just supplied it).

func (*Store) CreateUser 

func (s *Store) CreateUser(ctx context.Context, username, passwordHash, role string) (*model.User, error)

CreateUser inserts a user with a pre-hashed (bcrypt) password.

func (*Store) DBPath 

func (s *Store) DBPath() string

DBPath returns the SQLite database file path (for backup restore's file swap) or the raw DSN string on Postgres. Callers branch on IsPostgres() before using the SQLite path.

func (*Store) DefaultFlow 

func (s *Store) DefaultFlow(ctx context.Context) (*model.Flow, error)

DefaultFlow returns the flow marked default, or an error when none exists.

func (*Store) DeleteAPIToken 

func (s *Store) DeleteAPIToken(ctx context.Context, id int64) error

DeleteAPIToken removes a token. Deleting an unknown ID is not an error (idempotent), so a double-click in the UI doesn't surface a failure.

func (*Store) DeleteFlow 

func (s *Store) DeleteFlow(ctx context.Context, id int64) error

DeleteFlow removes a flow only when no job references it. The schema FK (jobs.flow_id REFERENCES flows(id)) enforces this; refusing up front gives an honest error instead of a constraint violation.

func (*Store) DeleteJob 

func (s *Store) DeleteJob(ctx context.Context, id int64) (bool, error)

DeleteJob removes a TERMINAL job (done/failed/cancelled) from the jobs table. Non-terminal jobs (pending/assigned/running) are refused so an in-flight encode is never orphaned by a delete. Returns changed=true when exactly one row was deleted; changed=false when the job was missing or non-terminal (the caller distinguishes via a prior GetJob or answers 404 vs 409). The guard is a WHERE status IN ('done','failed','cancelled') clause so the check and the delete are atomic — a job that went terminal between the handler's pre-check and this call is safely deleted, and a job that went non-terminal is safely left alone.

func (*Store) DeleteNode 

func (s *Store) DeleteNode(ctx context.Context, id int64) error

DeleteNode removes a node row (used to roll back failed pairing).

func (*Store) DeleteNodeIfNotBusy 

func (s *Store) DeleteNodeIfNotBusy(ctx context.Context, id int64) (int64, error)

DeleteNodeIfNotBusy removes a node only if it is not busy, atomically. Returns rows affected: 0 means the node is busy (or gone) — the caller must not proceed, otherwise a job dispatched between check and delete would be orphaned mid-encode (the TOCTOU the old check-then-act had).

func (*Store) DeleteSession 

func (s *Store) DeleteSession(ctx context.Context, tokenHash string) error

DeleteSession revokes one session (logout).

func (*Store) DeleteShare 

func (s *Store) DeleteShare(ctx context.Context, id int64) error

DeleteShare removes a share row.

func (*Store) DeleteStepTemplate 

func (s *Store) DeleteStepTemplate(ctx context.Context, id int64) error

DeleteStepTemplate removes a custom template. Built-ins and templates referenced by flows are refused.

func (*Store) DeleteUser 

func (s *Store) DeleteUser(ctx context.Context, userID int64) error

DeleteUser removes a user account. The sessions table has ON DELETE CASCADE so the user's sessions die with the row. Does NOT guard against last-admin deletion — the handler enforces that policy before calling.

func (*Store) DeleteUserSessions 

func (s *Store) DeleteUserSessions(ctx context.Context, userID int64, keepTokenHash string) error

DeleteUserSessions revokes every session of a user except the one identified by keepTokenHash (the session performing the action). Used on password change so stale or stolen sessions die immediately.

func (*Store) FinishJob 

func (s *Store) FinishJob(ctx context.Context, id int64, status model.JobStatus, exitCode int, errMsg string, outputs []string, logTail string) error

FinishJob marks a job done or failed with exit code, error, and outputs.

func (*Store) FinishJobWithReport 

func (s *Store) FinishJobWithReport(ctx context.Context, id int64, status model.JobStatus, exitCode int, errMsg string, outputs []string, logTail string, fullLog string, timings []model.StepTiming, metrics map[string]float64) error

FinishJobWithReport marks a job terminal and persists the agent's full completion report: the v1 finish columns (status/exit_code/error/outputs/ log_tail/finished_at) PLUS the v2 observability fields (full_log + step_timings_json) the Phase-A agent report carries. It is the completion path for agents that report logs and timings; the older FinishJob remains for the two callers (orphan recovery, render-failed) that correctly pass no log. full_log is capped at 1 MiB via capFullLog; timings are marshaled via marshalStepTimings (the column's marshal contract).

The UPDATE is guarded by status IN ('assigned','running') so a concurrent cancel (or a duplicate completion that raced past the Terminal() pre-check) cannot re-finish a job that already left the live state. When the guard matches zero rows, ErrJobNotFinishable is returned so the caller can answer idempotently (the job is already terminal — say "already_recorded", don't notify or retry) instead of 500-ing. The handler's Terminal() pre-check remains the fast path; this guard closes the narrow window between that check and this write.

func (*Store) FlowByName 

func (s *Store) FlowByName(ctx context.Context, name string) (*model.Flow, error)

FlowByName loads a flow by unique name.

func (*Store) GetFlow 

func (s *Store) GetFlow(ctx context.Context, id int64) (*model.Flow, error)

GetFlow loads a flow by ID.

func (*Store) GetJob 

func (s *Store) GetJob(ctx context.Context, id int64) (*model.Job, error)

GetJob loads a job by ID.

func (*Store) GetNode 

func (s *Store) GetNode(ctx context.Context, id int64) (*model.Node, error)

GetNode loads a node by ID.

func (*Store) GetPairingCode 

func (s *Store) GetPairingCode(ctx context.Context, id int64) (*model.PairingCode, error)

GetPairingCode loads one code by ID.

func (*Store) GetProvisionRun 

func (s *Store) GetProvisionRun(ctx context.Context, id int64) (*model.ProvisionRun, error)

GetProvisionRun loads one run (log included — callers tail it).

func (*Store) GetSeries 

func (s *Store) GetSeries(ctx context.Context, id int64) (*model.Series, error)

GetSeries loads a series by ID.

func (*Store) GetSettings 

func (s *Store) GetSettings(ctx context.Context) (*model.Settings, error)

GetSettings loads the persisted settings row. Returns nil (not an error) when no row exists yet — callers merge environment defaults in that case.

func (*Store) GetShare 

func (s *Store) GetShare(ctx context.Context, id int64) (*model.Share, error)

GetShare loads one share with the password decrypted.

func (*Store) GetStepTemplate 

func (s *Store) GetStepTemplate(ctx context.Context, id int64) (*model.StepTemplate, error)

GetStepTemplate loads a template by ID.

func (*Store) InsertNodeMetric 

func (s *Store) InsertNodeMetric(ctx context.Context, nodeID int64, m *model.NodeMetrics) error

InsertNodeMetric appends one resource sample to the node_metrics ring table and prunes rows older than the retention window. The caller (the heartbeat handler) treats the returned error as best-effort: it logs a WARN and continues so a metrics write failure never kills the heartbeat (which would strand the agent without instructions). ts uses datetime('now') so the value matches the "2006-01-02 15:04:05" UTC format the schema and every other timestamp column use — the (node_id, ts) index sorts lexically, so an RFC3339 offset format would break ordering.

func (*Store) InsertStepTemplateIfAbsent 

func (s *Store) InsertStepTemplateIfAbsent(ctx context.Context, t *model.StepTemplate) (*model.StepTemplate, error)

InsertStepTemplateIfAbsent creates a template only when its key is not yet in the store. Boot seeding uses this so edits made through the UI survive controller restarts (re-seeding only adds brand-new templates).

func (*Store) IsPostgres 

func (s *Store) IsPostgres() bool

IsPostgres reports whether the store is backed by Postgres.

func (*Store) JobExistsForEpisode 

func (s *Store) JobExistsForEpisode(ctx context.Context, episodeDir string) (bool, error)

JobExistsForEpisode reports whether any job (non-cancelled) already covers this episode dir.

func (*Store) JobStats 

func (s *Store) JobStats(ctx context.Context, rng time.Duration) (*model.Stats, error)

JobStats computes the fleet-wide aggregate snapshot over the job history for jobs that finished within the given range window (finished_at >= now-range). rng<=0 means "all time" (no finished_at lower bound).

Every aggregate is a SQL query over the existing jobs table + LEFT JOINs to nodes/flows for display names — no new storage, no schema change. The queries are shaped to use the existing indexes:

  • idx_jobs_status(status) backs the status-filtered terminal counts (done/failed/cancelled) and the failures-by-step GROUP BY.
  • idx_jobs_node_status(node_id, status) backs the per-node GROUP BY (node_id, status) — a composite seek, not a full scan.

The finished_at range predicate is applied as a secondary filter after the index seek; at the fleet's volume (thousands of finished jobs, not millions) this is a cheap bounded scan over the index-matched rows. An empty/never-used fleet returns a zeroed Stats shape (non-nil slices marshaled as []) — never an error — so the Stats page renders cleanly on a fresh install.

func (*Store) LatestNodeMetric 

func (s *Store) LatestNodeMetric(ctx context.Context, nodeID int64) (*model.NodeMetrics, error)

LatestNodeMetric returns the most recent metric sample for a node as a NodeMetrics pointer, or nil if the node has never sent metrics. Used by the node list view to surface the last-known sample per node without holding nodes in memory between heartbeats.

func (*Store) ListAPITokens 

func (s *Store) ListAPITokens(ctx context.Context) ([]*model.APIToken, error)

ListAPITokens returns all tokens (hashes omitted via json:"-" anyway).

func (*Store) ListAudit 

func (s *Store) ListAudit(ctx context.Context, limit int) ([]*model.AuditEvent, error)

ListAudit returns the newest audit events (up to limit; 0 = default 200). Always returns a non-nil slice so JSON renders [] not null.

func (*Store) ListFlows 

func (s *Store) ListFlows(ctx context.Context) ([]*model.Flow, error)

ListFlows returns all flows ordered by name.

func (*Store) ListJobs 

func (s *Store) ListJobs(ctx context.Context, status model.JobStatus, limit int) ([]*model.Job, error)

ListJobs returns jobs, optionally filtered by status, newest first.

func (*Store) ListNodeMetrics 

func (s *Store) ListNodeMetrics(ctx context.Context, nodeID int64, since time.Time, limit int) ([]model.NodeMetricSample, error)

ListNodeMetrics returns metric samples for a node newer than since, ordered oldest-first, capped at metricsMaxPoints. When the raw row count for the range exceeds the cap, rows are evenly sampled (every Nth row) so the dashboard receives a bounded, representative payload instead of thousands of points.

func (*Store) ListNodes 

func (s *Store) ListNodes(ctx context.Context) ([]*model.Node, error)

ListNodes returns all nodes ordered by name.

func (*Store) ListPairingCodes 

func (s *Store) ListPairingCodes(ctx context.Context) ([]*model.PairingCode, error)

ListPairingCodes returns active (unused, unexpired) codes for the UI.

func (*Store) ListProvisionRuns 

func (s *Store) ListProvisionRuns(ctx context.Context) ([]*model.ProvisionRun, error)

ListProvisionRuns returns runs newest-first (the UI shows the recent ones).

func (*Store) ListSeries 

func (s *Store) ListSeries(ctx context.Context) ([]*model.Series, error)

ListSeries returns all series ordered by name.

func (*Store) ListShares 

func (s *Store) ListShares(ctx context.Context) ([]*model.Share, error)

ListShares returns all shares ordered by role,kind,name with passwords decrypted (provisioning needs them; the API layer masks).

func (*Store) ListStepTemplates 

func (s *Store) ListStepTemplates(ctx context.Context) ([]*model.StepTemplate, error)

ListStepTemplates returns all templates ordered by key.

func (*Store) MigrateSettingsKey 

func (s *Store) MigrateSettingsKey(ctx context.Context, key string, value string) (bool, error)

MigrateSettingsKey fills a settings key added by a newer build. It only writes when a row exists AND the key is absent (an old-build save); a key already present — including an operator-set blank — is never touched. The row's updated_at is preserved so the patch does not masquerade as a manual edit. Returns whether the row was changed.

func (*Store) MigrateSettingsShares 

func (s *Store) MigrateSettingsShares(ctx context.Context) int

MigrateSettingsShares converts the legacy flat NFS Settings fields into share rows on first boot after upgrade. Runs once per unique (server,path); returns how many rows were created. Never destructive: the Settings fields stay readable for the old code paths until they're fully removed.

func (*Store) NextAssignableJob 

func (s *Store) NextAssignableJob(ctx context.Context) (*model.Job, error)

NextAssignableJob returns the highest-priority pending job whose backoff gate has elapsed (next_retry_at IS NULL OR <= now), ordered by priority DESC then id ASC. Priority DESC dispatches urgent jobs first; id ASC is true FIFO within a priority tier — episodes encode in submission order on an encode farm, not newest-first. Returns nil (not an error) when no job is assignable. ListJobs stays unchanged for the UI (retry-pending jobs remain visible).

func (*Store) NextAssignableJobForNode 

func (s *Store) NextAssignableJobForNode(ctx context.Context, nodeID int64) (*model.Job, error)

NextAssignableJobForNode returns the best pending job for a specific node. Same backoff gate as NextAssignableJob; ordering is priority DESC (operator intent first), then steering — a job whose last failure was on THIS node sorts after an equally-prioritized job that failed elsewhere or never failed — then id ASC (true FIFO within a tier). Why steering: a job that failed from a node-local fault (full disk, dead GPU, corrupt toolchain) re-fails identically on retry; another node gets it first. Steering is only a tie-break WITHIN a priority tier: a high-priority job that failed here still outranks a normal-priority fresh job, and when the steered-away job is the only candidate it is still returned — a single-node farm must never starve.

func (*Store) NodeByName 

func (s *Store) NodeByName(ctx context.Context, name string) (*model.Node, error)

NodeByName loads a node by unique name.

func (*Store) NodeByTokenHash 

func (s *Store) NodeByTokenHash(ctx context.Context, hash string) (*model.Node, error)

NodeByTokenHash finds the node holding this token hash (auth lookup).

func (*Store) PatchPendingJob 

func (s *Store) PatchPendingJob(ctx context.Context, id int64, flowID *int64, priority *int) (bool, error)

PatchPendingJob updates a pending job's flow_id and/or priority in a single guarded UPDATE (WHERE id=? AND status='pending'). This replaces the old two-call SetJobFlow+SetJobPriority sequence in handlePatchJob, which had a race window: between the two writes the job could be assigned, leaving the second write silently no-op'ing (or partially applying flow but not priority, or vice versa). A single atomic UPDATE closes that window.

Either pointer may be nil (skip that field — COALESCE-style conditional SET). Returns changed=true when exactly one row matched (the job was pending and got updated); changed=false when zero rows matched (the job is no longer pending — caller answers 409). SetJobFlow/SetJobPriority are retained for any other callers and the store-level tests.

func (*Store) PeekPairingCode 

func (s *Store) PeekPairingCode(ctx context.Context, codeHash string) (*model.PairingCode, error)

PeekPairingCode validates a code without consuming it: unknown/expired/ used codes are reported distinctly, and transient DB errors propagate as such instead of being disguised as auth failures.

func (*Store) PruneNodeMetrics 

func (s *Store) PruneNodeMetrics(ctx context.Context, nodeID int64, maxAge time.Duration) error

PruneNodeMetrics deletes rows older than maxAge for a single node from node_metrics. The WHERE node_id = ? AND ts < … predicate seeks the composite index idx_metrics_node_ts(node_id, ts) (EXPLAIN QUERY PLAN reports SEARCH, not SCAN) — a ts-only predicate cannot use that index, so the prune is scoped per node rather than a full-table sweep. Uses SQLite's datetime modifier so the comparison stays in the same text format the rows were written in (no cross-format parsing). Idempotent: a no-op when nothing is older than the cutoff.

func (*Store) PruneOldJobs 

func (s *Store) PruneOldJobs(ctx context.Context, days int) (int64, error)

PruneOldJobs deletes terminal job rows finished more than days ago and returns the number deleted. days <= 0 is a hard no-op (0 = retention disabled in settings — a misconfigured row must never wipe history). Only rows with a stamped finished_at qualify: a pending/running job has finished_at NULL and survives no matter how old it is (the queue is never pruned out from under the dispatcher). Deletion is permanent — the row carries the full log and timings, so this is the retention policy's whole purpose (bounding SQLite growth), not a soft archive.

func (*Store) PruneProvisionRuns 

func (s *Store) PruneProvisionRuns(ctx context.Context, keep int) error

PruneProvisionRuns keeps only the most recent keep rows. Each row carries a full run log, so the table must not grow without bound across the controller's lifetime. Finished runs are pruned first (never a live one).

func (*Store) PruneSessions 

func (s *Store) PruneSessions(ctx context.Context) error

PruneSessions removes expired sessions.

func (*Store) ReleaseNode 

func (s *Store) ReleaseNode(ctx context.Context, nodeID int64) error

ReleaseNode recomputes the node's busy/idle flag after a job finishes: idle when a slot is free (active < max_concurrent_jobs), busy while the node is still at capacity. Finishing one of N jobs must not mark a still-full node idle, and freeing the last slot of a capped node must make it available again.

func (*Store) RetryJob 

func (s *Store) RetryJob(ctx context.Context, id int64) (int64, error)

RetryJob re-queues a failed/cancelled job as pending on no node. It returns the number of rows re-queued (0 when the job is not retryable, e.g. still running) so callers can distinguish success from a no-op. Stale v2 capture (full_log, step_timings_json, next_retry_at) is cleared alongside the v1 state so a retry starts clean and is never blocked by a stale backoff gate once the queue gates dispatch on next_retry_at. retry_count is reset to 0 (why: operator intervention refreshes the automatic retry budget — without this, an exhausted job that was manually retried could never auto-retry again, locking it out of the policy the operator just chose to use).

func (*Store) SaveSettings 

func (s *Store) SaveSettings(ctx context.Context, st *model.Settings) error

SaveSettings persists the settings row (single-row upsert on id=1).

func (*Store) ScheduleJobRetry 

func (s *Store) ScheduleJobRetry(ctx context.Context, jobID int64, retryCount int, nextRetryAt time.Time) error

ScheduleJobRetry silently re-queues a failed job for automatic retry: it flips the job back to pending, stamps retry_count, and sets next_retry_at to the computed backoff time. The UPDATE is guarded by status='failed' so it is atomic with the finished state — a job that already left the failed state (manual retry, cancel) matches zero rows and is a no-op (nil error). This prevents a retry from resurrecting a job a concurrent path already moved on.

The clear set mirrors RetryJob: the failed run's stale observability fields (started_at, finished_at, error, exit_code, outputs_json, log_tail, full_log, step_timings_json) are wiped so a re-queued pending job never briefly shows the old run's error/log to the UI. Without this, a just-retried job sits 'pending' carrying the previous failure's error message and log tail until it is re-assigned.

func (*Store) SeedFinishedJob 

func (s *Store) SeedFinishedJob(ctx context.Context, flowID, nodeID int64, status, step, startedAt, finishedAt string) error

SeedFinishedJob inserts a terminal job row with explicit started_at and finished_at timestamps, for tests that need controlled durations and range boundaries. The normal CreateJob/AssignJob/FinishJob path stamps finished_at=datetime('now'), which is uncontrolled; this helper lets a test backdate. It is exported so cross-package API tests can use it too.

func (*Store) SeriesByName 

func (s *Store) SeriesByName(ctx context.Context, name string) (*model.Series, error)

SeriesByName loads a series by exact folder name.

func (*Store) SeriesProgressCounts 

func (s *Store) SeriesProgressCounts(ctx context.Context) (map[string]SeriesProgress, error)

SeriesProgressCounts computes progress counts for every series that has at least one job row, in a single query (no N+1). The "eventually done wins" semantic is applied Go-side: an episode_dir that appears in both the done set and the failed set (a retried-then-succeeded episode) is counted as done only, NOT failed.

The query returns one row per distinct (series, episode_dir, status) combination; Go collapses these into per-series done/failed/active/seen sets with the set difference. Series with zero jobs simply do not appear in the result map (callers treat absence as all-zero). The query is covered by idx_jobs_episode_dir + idx_jobs_status so the cost is bounded.

func (*Store) SessionByTokenHash 

func (s *Store) SessionByTokenHash(ctx context.Context, tokenHash string) (*model.Session, error)

SessionByTokenHash loads an UNEXPIRED session joined with its user.

func (*Store) SetDefaultFlow 

func (s *Store) SetDefaultFlow(ctx context.Context, id int64) error

SetDefaultFlow marks one flow default and clears all others, atomically.

func (*Store) SetNodeGroup 

func (s *Store) SetNodeGroup(ctx context.Context, id int64, group string) error

SetNodeGroup updates ONLY the routing label — field-scoped SQL so a concurrent status/enabled change isn't clobbered. ” = wildcard node.

func (*Store) SetNodeMaxConcurrent 

func (s *Store) SetNodeMaxConcurrent(ctx context.Context, id int64, maxJobs int) error

SetNodeMaxConcurrent updates the per-node job-slot cap, clamped to 1-8 (defense in depth: the API validates too, but a direct store caller must never be able to write 0 slots — that would strand the node).

func (*Store) SetProvisionRunStatus 

func (s *Store) SetProvisionRunStatus(ctx context.Context, id int64, status, errMsg string, finished bool) error

SetProvisionRunStatus updates status + optional error/finished markers.

func (*Store) SetSeriesEnabled 

func (s *Store) SetSeriesEnabled(ctx context.Context, id int64, enabled bool) error

SetSeriesEnabled updates ONLY the enabled flag (same rationale).

func (*Store) SetSeriesFlow 

func (s *Store) SetSeriesFlow(ctx context.Context, id, flowID int64) error

SetSeriesFlow updates ONLY the flow selection — field-scoped SQL so a concurrent enable/disable patch cannot be clobbered by a stale full-row read-modify-write.

func (*Store) SetSeriesNodeGroup 

func (s *Store) SetSeriesNodeGroup(ctx context.Context, id int64, group string) error

SetSeriesNodeGroup updates ONLY the routing label — field-scoped SQL. ” means the series' jobs accept any node.

func (*Store) SetSeriesNotify 

func (s *Store) SetSeriesNotify(ctx context.Context, id int64, notify bool) error

SetSeriesNotify updates ONLY the notify (Discord mute) flag — field-scoped SQL so a concurrent flow/tag patch cannot be clobbered by a stale read-modify-write. A muted series (notify=false) suppresses both the direct alert and the digest buffering for its jobs.

func (*Store) SetSeriesPaused 

func (s *Store) SetSeriesPaused(ctx context.Context, id int64, paused bool) error

SetSeriesPaused updates ONLY the paused flag — field-scoped SQL like the other setters. A paused series holds at BOTH gates: the scanner skips it (no new jobs) and dispatch refuses its queued pending jobs until lifted.

func (*Store) SetSeriesTag 

func (s *Store) SetSeriesTag(ctx context.Context, id int64, tag string) error

SetSeriesTag updates ONLY the quality tag override (same rationale).

func (*Store) SetSeriesWebhook 

func (s *Store) SetSeriesWebhook(ctx context.Context, id int64, webhookURL string) error

SetSeriesWebhook sets the per-series Discord webhook override (” = use the global webhook). Field-scoped like SetSeriesNotify so a concurrent scanner upsert can't clobber it.

func (*Store) ShareForRole 

func (s *Store) ShareForRole(ctx context.Context, role model.ShareRole) (*model.Share, error)

ShareForRole returns the enabled share serving a role, preferring SMB over NFS over S3 for mount-based consumers (deterministic when several exist). Nil,nil when none.

func (*Store) SlideSession 

func (s *Store) SlideSession(ctx context.Context, tokenHash string, expiresAt time.Time) error

SlideSession extends an active session's expiry.

func (*Store) StepTemplateByKey 

func (s *Store) StepTemplateByKey(ctx context.Context, key string) (*model.StepTemplate, error)

StepTemplateByKey loads a template by unique key.

func (*Store) TouchAPIToken 

func (s *Store) TouchAPIToken(ctx context.Context, id int64) error

TouchAPIToken stamps last_used_at (best-effort bookkeeping; callers ignore errors).

func (*Store) TryAdvisoryLock 

func (s *Store) TryAdvisoryLock(ctx context.Context, key int64) (release func(), ok bool, err error)

TryAdvisoryLock attempts to acquire a non-transactional Postgres advisory lock for key. On Postgres it dedicates a *sql.Conn from the pool for the lock's entire lifetime (pgx advisory locks are session-scoped, and database/sql pools connections, so the lock's owning session must be held by ONE stable Conn — acquiring the lock through the pool would release it when the conn returns to the pool). It runs pg_try_advisory_lock($1); on success it returns (release, true, nil) where release is idempotent and releases the lock (pg_try_advisory_unlock on the same conn) then closes the conn. A background keepalive goroutine runs SELECT 1 on the same conn every advisoryLockKeepalive; if the keepalive fails the lock is considered lost — release() is called and the returned release func is a no-op (the conn is already gone). Callers should re-acquire on loss.

On SQLite this returns (noop-release, true, nil): a single instance has no peer to race, so the lock always succeeds and the release does nothing. This keeps the supervisor path identical across backends — the SQLite controller wraps every loop in the supervisor and always "holds" the lock, so its background loops run exactly as before.

ctx is used ONLY for the initial acquire attempt; the keepalive runs under a detached context so cancelling the caller's ctx does not kill the keepalive mid-hold (release() owns the conn's lifetime).

func (*Store) UpdateFlow 

func (s *Store) UpdateFlow(ctx context.Context, f *model.Flow) error

UpdateFlow replaces name/steps of an existing flow.

func (*Store) UpdateJobStatus 

func (s *Store) UpdateJobStatus(ctx context.Context, id int64, status model.JobStatus, step string, progress float64, logTail string) error

UpdateJobStatus records progress from a heartbeat or completion report. Only live jobs (assigned/running) are updatable — a stale heartbeat must not resurrect a terminal job back into the queue.

func (*Store) UpdateNode 

func (s *Store) UpdateNode(ctx context.Context, n *model.Node) error

UpdateNode persists mutable node fields after a heartbeat or UI action.

func (*Store) UpdateSeries 

func (s *Store) UpdateSeries(ctx context.Context, sr *model.Series) error

UpdateSeries persists flow selection, enabled state, and notify flag.

func (*Store) UpdateShare 

func (s *Store) UpdateShare(ctx context.Context, sh *model.Share) error

UpdateShare applies field changes. An empty Password keeps the stored one (the API never round-trips plaintext); to clear it set a sentinel via DeleteShare/recreate — passwords are only ever replaced, not wiped.

func (*Store) UpdateUserPassword 

func (s *Store) UpdateUserPassword(ctx context.Context, userID int64, passwordHash string) error

UpdateUserPassword replaces a user's bcrypt hash.

func (*Store) UpgradeStepTemplateIfFactory 

func (s *Store) UpgradeStepTemplateIfFactory(ctx context.Context, key, oldPowerShell string, t *model.StepTemplate) (bool, error)

UpgradeStepTemplateIfFactory replaces a builtin template's factory version with a new one — but ONLY when the stored script still matches the old factory version byte-for-byte. A template the user edited in the UI never matches, so customizations survive factory upgrades. Returns whether the upgrade was applied.

func (*Store) UpsertSeriesByName 

func (s *Store) UpsertSeriesByName(ctx context.Context, name string) (*model.Series, error)

UpsertSeriesByName returns the series row for name, creating it enabled with no flow override when first seen. Existing rows are returned as-is.

func (*Store) UpsertStepTemplate 

func (s *Store) UpsertStepTemplate(ctx context.Context, t *model.StepTemplate) (*model.StepTemplate, error)

UpsertStepTemplate inserts a template by key or updates it when present. Used by the UI for edits and custom templates; boot seeding uses InsertStepTemplateIfAbsent instead to preserve those edits.

func (*Store) UserByID 

func (s *Store) UserByID(ctx context.Context, id int64) (*model.User, error)

UserByID loads one account by its numeric id (nil when absent).

func (*Store) UserByUsername 

func (s *Store) UserByUsername(ctx context.Context, username string) (*model.User, error)

UserByUsername loads one account (nil when absent).

func (*Store) Users 

func (s *Store) Users(ctx context.Context) ([]*model.User, error)

Users lists accounts (no hashes).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.