Documentation
¶
Overview ¶
Package scriptstore is the PostgreSQL implementation of the managed-script store contract (pkg/script). It is built only by internal/platform/ scriptlayer, which is why it lives under internal/ rather than beside the domain: an implementation seam with one composition-root caller is not part of the module's supported import surface (docs/library/stability.md).
The layout follows pkg/prompt/postgres closely — a column list defined once so the scan order cannot drift from the query, a withTx helper, and version writes transactional with the scripts row they touch.
Index ¶
- Constants
- func NewDiscoveryStore(db *sql.DB) script.Store
- type Store
- func (s *Store) AdvanceSchedule(ctx context.Context, adv script.ScheduleAdvance) (bool, error)
- func (s *Store) Claim(ctx context.Context, worker string, lease time.Duration) (*script.Run, error)
- func (s *Store) Contract(ctx context.Context, id string) (*script.Contract, error)
- func (s *Store) Create(ctx context.Context, sc *script.Script, author script.Author) error
- func (s *Store) Delete(ctx context.Context, id string) error
- func (s *Store) DueSchedules(ctx context.Context, now time.Time, limit int) ([]script.Schedule, error)
- func (s *Store) Enqueue(ctx context.Context, r *script.Run) error
- func (s *Store) Finish(ctx context.Context, lease script.RunLease, result script.RunResult) error
- func (s *Store) GetByID(ctx context.Context, id string) (*script.Script, error)
- func (s *Store) GetByName(ctx context.Context, ownerEmail, name string) (*script.Script, error)
- func (s *Store) GetRun(ctx context.Context, id string) (*script.Run, error)
- func (s *Store) GetSchedule(ctx context.Context, scriptID string) (*script.Schedule, error)
- func (s *Store) GetVersion(ctx context.Context, scriptID string, version int) (*script.Version, error)
- func (s *Store) GetVersionByID(ctx context.Context, id string) (*script.Version, error)
- func (s *Store) LatestDryRun(ctx context.Context, scriptID string, sourceSHA256 []byte) (*script.DryRun, error)
- func (s *Store) LatestRuns(ctx context.Context, scriptIDs []string) (map[string]script.Run, error)
- func (s *Store) List(ctx context.Context, filter script.ListFilter) ([]script.Script, error)
- func (s *Store) ListRuns(ctx context.Context, filter script.RunFilter) ([]script.Run, error)
- func (s *Store) ListSchedules(ctx context.Context, filter script.ScheduleFilter) ([]script.Schedule, error)
- func (s *Store) ListVersions(ctx context.Context, scriptID string) ([]script.Version, error)
- func (s *Store) MaterializeRun(ctx context.Context, r *script.Run) (script.Materialization, error)
- func (s *Store) PurgeRuns(ctx context.Context, retention time.Duration) (int64, error)
- func (s *Store) RecordDryRun(ctx context.Context, d *script.DryRun) error
- func (s *Store) RecordOutput(ctx context.Context, lease script.RunLease, out script.RunOutput) error
- func (s *Store) Retry(ctx context.Context, lease script.RunLease, cause string, ...) error
- func (s *Store) Search(ctx context.Context, q script.SearchQuery) ([]script.ScoredScript, error)
- func (s *Store) SetSchedule(ctx context.Context, sched *script.Schedule) error
- func (s *Store) SetScheduleEnabled(ctx context.Context, scriptID string, enabled bool, actor string) error
- func (s *Store) Transfer(ctx context.Context, id, newOwnerEmail string, author script.Author) error
- func (s *Store) Update(ctx context.Context, sc *script.Script) error
- func (s *Store) UpdateWithVersion(ctx context.Context, sc *script.Script, author script.Author) error
Constants ¶
const NotifyChannel = "script_runs"
NotifyChannel is the pg_notify channel a producer fires so a run worker wakes without waiting for its poll tick. Enqueue fires it best-effort; the worker polls regardless, so a missed notification costs latency and nothing else.
Variables ¶
This section is empty.
Functions ¶
func NewDiscoveryStore ¶
NewDiscoveryStore returns the store the search federation reads to rank scripts and to resolve an mcp:script:<id> reference (#1302), or nil when the deployment has no database and therefore no scripts to find.
It exists so the composition root asks for the capability in one expression. Discovery builds its own handle rather than borrowing the one the script feature assembles, because search is federated during initialization while the feature is wired at Start; the store is a stateless reader over the same *sql.DB — the tool layer and the run worker each construct their own for the same reason — so this is a second handle on one database, not a second authority.
Types ¶
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store implements script.Store and script.VersionStore using PostgreSQL.
func New ¶
func New(db *sql.DB, opts ...indexjobs.StoreOption) *Store
New creates a PostgreSQL script store over db. Pass indexjobs.WithProducer to bind the write-path index-job producer; without it the write path enqueues nothing and the reconciler is the only route to the index.
func (*Store) AdvanceSchedule ¶
AdvanceSchedule moves a schedule forward, only if it is still where the caller found it.
The From guard is what keeps two replicas that walked the same fire from double-counting the misses or moving the schedule twice: the second UPDATE matches no row and the caller learns it lost the race. It is an efficiency measure, not the single-fire guarantee — that one belongs to the unique index on the run — which is why losing here is silent.
func (*Store) Claim ¶
Claim takes the next due run for worker and holds it for lease.
The UPDATE is the claim: one statement that selects the oldest due row with FOR UPDATE SKIP LOCKED, marks it running, increments the attempt, and stamps the lease. Concurrent workers on other replicas skip each other's locked rows rather than blocking, and a run whose worker died is picked up by the next claim once its lease expires.
func (*Store) Contract ¶
Contract composes the contract document for one script: the live record and its parameter contract, the cadence when it has one, and the last successful run with what it produced.
Returns nil, nil when no such script exists. A missing schedule is not an error (most scripts have none), and neither is a script that has never completed a run: both are ordinary states the document reports as absent.
func (*Store) Create ¶
Create persists a new script and its v1 snapshot in one transaction, so a script never exists without the version history that explains it.
func (*Store) DueSchedules ¶
func (s *Store) DueSchedules(ctx context.Context, now time.Time, limit int) ([]script.Schedule, error)
DueSchedules returns enabled schedules whose next fire has arrived.
now is passed in rather than read from the database clock so a caller can walk a schedule deterministically in a test; the correctness of a fire does not depend on which clock decided it was due, only on the unique index the resulting insert conflicts against.
func (*Store) Enqueue ¶
Enqueue inserts a pending run and fires a best-effort wakeup so a worker claims it without waiting for the next poll.
The run id is supplied by the caller rather than generated here: it is also the run's session id, minted before the run exists so every audit row the run produces carries it.
func (*Store) GetByName ¶ added in v1.123.0
GetByName retrieves one owner's script by name. An empty owner matches nothing: the ownerless rows a transfer exists to adopt are addressable by id alone, never by a name lookup an unidentified caller could make.
func (*Store) GetSchedule ¶
GetSchedule returns one script's schedule.
func (*Store) GetVersion ¶
func (s *Store) GetVersion(ctx context.Context, scriptID string, version int) (*script.Version, error)
GetVersion returns one version with its full source, or nil, nil when the script has no such version.
func (*Store) GetVersionByID ¶
GetVersionByID returns one version by id, or nil, nil when no such version exists. It is the runner's read: the execution gate stores an id, and only an id identifies one immutable snapshot for the life of the script.
func (*Store) LatestDryRun ¶ added in v1.122.0
func (s *Store) LatestDryRun(ctx context.Context, scriptID string, sourceSHA256 []byte) (*script.DryRun, error)
LatestDryRun returns the newest account of one script's exact source.
Nil is the ordinary answer rather than an error: most versions were never dry-run, and the reviewer's question is "did anyone run this", whose honest negative is a nil.
func (*Store) LatestRuns ¶
LatestRuns returns the most recent run of each named script, keyed by script id, omitting the scripts that have never been run.
A listing that shows one row per script needs each script's last run, and asking for it script by script is a query per row. This is that answer in one query. It orders on creation rather than on completion — unlike the contract's last SUCCESSFUL run, which answers "what did this produce" and so must be a finished one — because a listing reports the state of the automation: a run that is pending or failed right now is the answer to "how is this going", and ordering by finished_at would hide it behind an older success.
func (*Store) ListSchedules ¶
func (s *Store) ListSchedules(ctx context.Context, filter script.ScheduleFilter) ([]script.Schedule, error)
ListSchedules returns schedules matching the filter, newest first.
func (*Store) ListVersions ¶
ListVersions returns every version of the script, newest first.
func (*Store) MaterializeRun ¶
MaterializeRun inserts one scheduled run and reports what happened.
The three outcomes are decided by the two unique indexes, not by the read that precedes them:
- the insert lands: this caller materialized the fire;
- it conflicts and a row for (schedule, fire time) exists: another replica materialized the same fire, which is the normal outcome of racing materializers and not a fault;
- it conflicts and no such row exists: the conflict was the one-open-run index, so the previous run is still going and the overlap policy applies. The skip is then recorded as its own terminal row, through the same conflict-tolerant insert, so two replicas racing to record it also produce exactly one.
func (*Store) PurgeRuns ¶
PurgeRuns deletes terminal runs older than retention.
Only terminal rows are swept, which now includes the skipped-overlap rows a schedule records: a skip is history the same way a failure is, and it carries a finished_at from the moment it exists so it ages out on the same clock. A pending or running row is live work, and a retention pass that could delete it would silently drop a run somebody is waiting on.
func (*Store) RecordDryRun ¶ added in v1.122.0
RecordDryRun stores one account of a draft execution and trims the author's older accounts of the same script.
The trim is in the same statement batch rather than in a sweeper because there is no other producer: an account is written by exactly one path, and bounding it where it is written means the table cannot grow while a sweeper is misconfigured or absent.
func (*Store) RecordOutput ¶
func (s *Store) RecordOutput(ctx context.Context, lease script.RunLease, out script.RunOutput) error
RecordOutput appends one persisted output to the claimed run.
It is called as each output lands rather than once at the end, because the row is what a reclaimed run reads to know what it already wrote. The append is done in SQL (|| on the JSONB array) rather than read-modify-write in Go, so two writes cannot lose one another.
func (*Store) Retry ¶
func (s *Store) Retry(ctx context.Context, lease script.RunLease, cause string, backoff time.Duration) error
Retry returns the claimed run to pending, due after backoff. It is for infrastructure failures only: a script error is deterministic and the same source on the same inputs fails the same way, so the worker never routes one here.
func (*Store) Search ¶
func (s *Store) Search(ctx context.Context, q script.SearchQuery) ([]script.ScoredScript, error)
Search ranks scripts by relevance to the query within the caller's visibility. Visibility is applied in SQL, before ranking, so a script the caller cannot see never reaches the ranker; a script is its owner's, so that predicate is the caller's own address.
A non-nil q.Embedding selects hybrid (semantic + lexical) ranking over the vectors the indexjobs scripts consumer writes, so a script is found by what it does and not only by the words it was named with (#1370). A nil embedding selects the lexical-only path, which is exactly the behavior a deployment with no embedding provider has always had.
func (*Store) SetSchedule ¶
SetSchedule creates or replaces one script's schedule.
The upsert is keyed on script_id rather than on the schedule id because the script is what a caller names: "set the schedule of daily-sales" must not depend on whether one already exists. Replacing keeps the id and the creation stamp, so a schedule's identity — and the runs that point at it — survive an edit of its cadence.
func (*Store) SetScheduleEnabled ¶
func (s *Store) SetScheduleEnabled(ctx context.Context, scriptID string, enabled bool, actor string) error
SetScheduleEnabled turns a schedule on or off.
Enabling does not move next_run_at. A schedule re-enabled after a pause is therefore due for whatever fire it was parked on, which the misfire policy then collapses to one run — the same treatment downtime gets, which is what a pause is.
func (*Store) Transfer ¶ added in v1.123.0
Transfer moves a script to a new owner and records the move as a new applied version authored by the administrator making it.
The version is written unconditionally, unlike an edit, which snapshots only when the substance moved. Nothing about the code changes here — what changes is the authority a run presents, since a run carries the roles captured on the version it executes. Writing the snapshot is therefore the transfer: it is what makes the script run as its new owner rather than as the person who no longer has it, and it leaves the hand-over in the history where the next reader of a run can see where that authority came from.
A collision with a script the new owner already keeps under the same name is reported as script.ErrNameTaken: names are unique within an owner, so the receiving side decides whether the transfer is possible.
func (*Store) Update ¶
Update writes the live script row. It does not touch version history; use UpdateWithVersion (through script.ApplyEdit) for edits that must be snapshotted.
func (*Store) UpdateWithVersion ¶
func (s *Store) UpdateWithVersion(ctx context.Context, sc *script.Script, author script.Author) error
UpdateWithVersion persists sc like Update and, when any versioned snapshot field changed against the stored row, records a new applied version authored by author and advances sc.Version to it.
Source Files