98 lines
3.5 KiB
Go
98 lines
3.5 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
)
|
||
|
||
// jobs.go: snapshots + durable jobs. A job = chapter×stage; durability is per-chunk
|
||
// (checkpoints, see ledger.go). A snapshot materializes the job context so the
|
||
// request-hash is deterministic for its entire lifetime (§3.1–3.2).
|
||
|
||
// UpsertSnapshot stores a materialized job-context snapshot (idempotent: the
|
||
// snapshot_id is itself a content hash, so re-inserting the same id is a
|
||
// no-op).
|
||
func (s *Store) UpsertSnapshot(snapshotID, briefHash, payloadJSON string) error {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
_, err := s.w.ExecContext(ctx, `
|
||
INSERT INTO snapshots (snapshot_id, brief_hash, payload) VALUES (?, ?, ?)
|
||
ON CONFLICT (snapshot_id) DO NOTHING`,
|
||
snapshotID, briefHash, payloadJSON)
|
||
return err
|
||
}
|
||
|
||
// Job is one chapter×stage unit of work.
|
||
type Job struct {
|
||
ID int64
|
||
BookID string
|
||
Chapter int
|
||
Stage string
|
||
Status string
|
||
SnapshotID string
|
||
}
|
||
|
||
// EnsureJob returns the existing job for (book, chapter, stage) or creates it
|
||
// bound to snapshotID. An existing job KEEPS its original snapshot — Р6:
|
||
// volatile parts are excluded from the resume key, confirming a term between
|
||
// runs must not invalidate already-translated chunks.
|
||
func (s *Store) EnsureJob(bookID string, chapter int, stage, snapshotID string) (*Job, error) {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
if _, err := s.w.ExecContext(ctx, `
|
||
INSERT INTO jobs (book_id, chapter, stage, snapshot_id) VALUES (?, ?, ?, ?)
|
||
ON CONFLICT (book_id, chapter, stage) DO NOTHING`,
|
||
bookID, chapter, stage, snapshotID); err != nil {
|
||
return nil, err
|
||
}
|
||
return s.getJob(ctx, bookID, chapter, stage)
|
||
}
|
||
|
||
func (s *Store) getJob(ctx context.Context, bookID string, chapter int, stage string) (*Job, error) {
|
||
j := Job{BookID: bookID, Chapter: chapter, Stage: stage}
|
||
err := s.r.QueryRowContext(ctx,
|
||
`SELECT id, status, snapshot_id FROM jobs WHERE book_id = ? AND chapter = ? AND stage = ?`,
|
||
bookID, chapter, stage).Scan(&j.ID, &j.Status, &j.SnapshotID)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return nil, nil
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &j, nil
|
||
}
|
||
|
||
// SetJobStatus transitions a job.
|
||
func (s *Store) SetJobStatus(jobID int64, status string) error {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
_, err := s.w.ExecContext(ctx,
|
||
`UPDATE jobs SET status = ?, updated_at = datetime('now') WHERE id = ?`, status, jobID)
|
||
return err
|
||
}
|
||
|
||
// SnapshotPayload returns the stored payload JSON of a snapshot, or "" when the id is unknown (an id
|
||
// written by an older schema, or a row whose snapshot row was never upserted). Read-only; the caller
|
||
// treats an unknown id as "cannot reason about this move" and falls back to the conservative answer.
|
||
func (s *Store) SnapshotPayload(snapshotID string) (string, error) {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
var payload string
|
||
err := s.r.QueryRowContext(ctx, `SELECT payload FROM snapshots WHERE snapshot_id = ?`, snapshotID).Scan(&payload)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return "", nil
|
||
}
|
||
return payload, err
|
||
}
|
||
|
||
// UpdateJobSnapshot re-pins a job to a new snapshot — an EXPLICIT operator
|
||
// command (tmctl --resnapshot): re-translating already-paid-for chunks happens
|
||
// only deliberately, never as a silent side effect of a config edit (Р6).
|
||
func (s *Store) UpdateJobSnapshot(jobID int64, snapshotID string) error {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
_, err := s.w.ExecContext(ctx,
|
||
`UPDATE jobs SET snapshot_id = ?, updated_at = datetime('now') WHERE id = ?`, snapshotID, jobID)
|
||
return err
|
||
}
|