| 1 | package draftstate |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "database/sql" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "time" |
| 12 | |
| 13 | filelock "reasonix/internal/identitylock" |
| 14 | ) |
| 15 | |
| 16 | const operationColumns = `id,draft_id,workspace_id,draft_revision,session_id,topic_id,submission_id,fingerprint,request_json,phase,error,created_at,updated_at,request_id,source_digest,operation_revision,execution_json` |
| 17 | |
| 18 | func rollbackTransaction(tx *sql.Tx) { |
| 19 | _ = tx.Rollback() |
| 20 | } |
| 21 | |
| 22 | // State reads both sides of the draft-operation binding in one transaction. |
| 23 | func (s *Store) State(ctx context.Context, id string) (Draft, *Operation, error) { |
| 24 | var draft Draft |
| 25 | var operation *Operation |
| 26 | err := s.withDB(ctx, func(db *sql.DB) error { |
| 27 | tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) |
| 28 | if err != nil { |
| 29 | return err |
| 30 | } |
| 31 | defer rollbackTransaction(tx) |
| 32 | draft, err = scanDraft(tx.QueryRowContext(ctx, `SELECT `+draftColumns+` FROM drafts WHERE id=?`, id)) |
| 33 | if err != nil { |
| 34 | return err |
| 35 | } |
| 36 | op, err := scanOperation(tx.QueryRowContext(ctx, `SELECT `+operationColumns+` FROM operations WHERE draft_id=? ORDER BY rowid DESC LIMIT 1`, id)) |
| 37 | if err == nil { |
| 38 | operation = &op |
| 39 | } else if !errors.Is(err, ErrOperationNotFound) { |
| 40 | return err |
| 41 | } |
| 42 | return tx.Commit() |
| 43 | }) |
| 44 | return draft, operation, err |
| 45 | } |
| 46 | |
| 47 | func (s *Store) RequestOperation(ctx context.Context, draftID, requestID string) (Operation, error) { |
| 48 | var op Operation |
| 49 | err := s.withDB(ctx, func(db *sql.DB) error { |
| 50 | var err error |
| 51 | op, err = scanOperation(db.QueryRowContext(ctx, `SELECT `+operationColumns+` FROM operations WHERE draft_id=? AND request_id=? ORDER BY rowid DESC LIMIT 1`, draftID, requestID)) |
| 52 | return err |
| 53 | }) |
| 54 | return op, err |
| 55 | } |
| 56 | |
| 57 | // WorkerLease excludes live creation workers across windows and processes. |
| 58 | // It is separate from the Session writer lease and never waits for its owner. |
| 59 | func (s *Store) WorkerLease(sessionID string) (func(), error) { |
| 60 | sum := sha256.Sum256([]byte(sessionID)) |
| 61 | directory := filepath.Join(filepath.Dir(s.path), "draft-workers") |
| 62 | if err := os.MkdirAll(directory, 0o700); err != nil { |
| 63 | return nil, err |
| 64 | } |
| 65 | return filelock.TryAcquire(filepath.Join(directory, fmt.Sprintf("%x.lock", sum))) |
| 66 | } |
| 67 | |
| 68 | // PublicationLease serializes cancellation with the short compare/publish |
| 69 | // boundary. Controller construction never holds this lock or a DB transaction. |
| 70 | func (s *Store) PublicationLease(ctx context.Context, operationID string) (func(), error) { |
| 71 | sum := sha256.Sum256([]byte(operationID)) |
| 72 | directory := filepath.Join(filepath.Dir(s.path), "draft-workers") |
| 73 | if err := os.MkdirAll(directory, 0o700); err != nil { |
| 74 | return nil, err |
| 75 | } |
| 76 | bounded, cancel := context.WithTimeout(ctx, 5*time.Second) |
| 77 | defer cancel() |
| 78 | return filelock.Acquire(bounded, filepath.Join(directory, fmt.Sprintf("%x.publish.lock", sum))) |
| 79 | } |
| 80 | |
| 81 | func (s *Store) ResumeOperation(ctx context.Context, id string, revision uint64) (Operation, error) { |
| 82 | var op Operation |
| 83 | err := s.withDB(ctx, func(db *sql.DB) error { |
| 84 | result, err := db.ExecContext(ctx, `UPDATE operations SET phase='reserved',error='',operation_revision=operation_revision+1 WHERE id=? AND operation_revision=? AND phase IN ('resume_required','runtime_failed')`, id, revision) |
| 85 | if err != nil { |
| 86 | return err |
| 87 | } |
| 88 | n, err := result.RowsAffected() |
| 89 | if err != nil { |
| 90 | return err |
| 91 | } |
| 92 | if n != 1 { |
| 93 | return ErrOperationConflict |
| 94 | } |
| 95 | op, err = scanOperation(db.QueryRowContext(ctx, `SELECT `+operationColumns+` FROM operations WHERE id=?`, id)) |
| 96 | return err |
| 97 | }) |
| 98 | return op, err |
| 99 | } |
| 100 |