84 lines
3.2 KiB
Go
84 lines
3.2 KiB
Go
package store
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
)
|
||
|
||
// jobs.go: snapshots + durable jobs. Джоба = глава×стадия; сохранность — чанк
|
||
// (checkpoints, см. ledger.go). Snapshot материализует контекст джобы, чтобы
|
||
// request-hash был детерминирован на всю её жизнь (§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: из
|
||
// ключа resume исключаются волатильные части, подтверждение термина между
|
||
// запусками не должно инвалидировать уже переведённые чанки.
|
||
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
|
||
}
|
||
|
||
// UpdateJobSnapshot re-pins a job to a new snapshot — ЯВНАЯ команда оператора
|
||
// (tmctl --resnapshot): пере-перевод уже оплаченных чанков случается только
|
||
// осознанно, никогда как тихий побочный эффект правки конфига (Р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
|
||
}
|