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) ApproveVersion(ctx context.Context, scriptID string, version int, approver string, ...) (*script.Version, 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) CreateDraftVersion(ctx context.Context, scriptID string, proposed *script.Script, ...) (int, 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) Get(ctx context.Context, name string) (*script.Script, error)
- func (s *Store) GetByID(ctx context.Context, id string) (*script.Script, error)
- func (s *Store) GetPersonal(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) 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) ListPendingReviews(ctx context.Context) ([]script.PendingReview, 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) RecordOutput(ctx context.Context, lease script.RunLease, out script.RunOutput) error
- func (s *Store) RejectVersion(ctx context.Context, scriptID string, version int) 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) 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 (*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) ApproveVersion ¶
func (s *Store) ApproveVersion(ctx context.Context, scriptID string, version int, approver string, grants script.Grants) (*script.Version, error)
ApproveVersion is the only write in this package that may set scripts.approved_version_id, and therefore the only thing that can make a script executable by the platform.
Everything it does happens under the script row lock, in one transaction:
- The grant's roles are REPLACED with the version's own author roles. A caller cannot pass roles in — approval records the authority the author held, so approving can narrow what a script reaches but can never hand it authority its author did not have.
- The approved version's snapshot is applied to the live row and every pending draft is superseded, so the code being served and the code being executed are the same code. That holds for a draft being promoted and equally for an earlier version being approved back into service: a rollback that moved the execution pointer while the live row kept serving the newer source would leave a script whose readable code is not the code that runs.
- The script's execution pointer moves to the approved version, and a script still in its authoring state becomes active.
Re-approving a version rebinds its grant and re-stamps the approval; that is the deliberate act the "widening requires re-approval" rule asks for.
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, the approved version behind the execution gate, 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) CreateDraftVersion ¶
func (s *Store) CreateDraftVersion(ctx context.Context, scriptID string, proposed *script.Script, author script.Author) (int, error)
CreateDraftVersion snapshots proposed's versioned fields as a new draft version of the script without touching the live row, returning the new version number. The approved version keeps executing until the draft is approved.
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) GetPersonal ¶
GetPersonal retrieves a personal script by owner and name.
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) 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) ListPendingReviews ¶
ListPendingReviews returns every version awaiting approval, oldest first.
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) 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) RejectVersion ¶
RejectVersion marks a pending draft rejected.
The status predicate is in the UPDATE rather than in a read before it: a draft that was approved, superseded, or already rejected between the read and the write must not be re-labeled, and an affected-row count is the only answer that cannot race.
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; see script.SearchQuery for why the persona arm scopes on membership rather than on the acting persona.
Ranking is lexical only. Scripts carry no embedding: the corpus is small and its searchable text is a name, a sentence, and a parameter list, so the hybrid machinery the prompt library needs would be cost without a gain.
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) 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. The review gate is re-validated under the row lock (see requireUngated).
Source Files