weichenandCursor
2d0fcb94bc
Opening a template stored database IDs without advancing nextSnapID, so the next temporary snapshot could overwrite and then delete the real baseline. Adopt helpers now bump every counter and only drop a map entry when it still points at the same object. Co-authored-by: Cursor <cursoragent@cursor.com>
460 lines
15 KiB
Go
460 lines
15 KiB
Go
package docrevision
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// ErrNeedsContent indicates the document has no server-side body to bootstrap a ledger from.
|
|
var ErrNeedsContent = errors.New("docrevision: needs content to bootstrap")
|
|
|
|
const ErrCodeNeedsContent = "DOCUMENT_REVISION_NEEDS_CONTENT"
|
|
|
|
// BootstrapRequest supplies the fact-source markdown used for first-access ledger creation.
|
|
type BootstrapRequest struct {
|
|
DocID string
|
|
Kind string // controlled|freeform
|
|
Markdown string
|
|
ProjectID int64
|
|
WorkDir string
|
|
LifecycleState string
|
|
Actor ActorSnapshot
|
|
Origin string // template|migration
|
|
}
|
|
|
|
func (s *Service) usingPG() bool {
|
|
return s != nil && s.pool != nil
|
|
}
|
|
|
|
// hydrateDocLocked loads ledger rows for docID from PostgreSQL into memory maps.
|
|
// Caller must hold s.mu.
|
|
func (s *Service) hydrateDocLocked(ctx context.Context, docID string) error {
|
|
if !s.usingPG() || docID == "" {
|
|
return nil
|
|
}
|
|
if root, ok := s.roots[docID]; ok && root.LedgerVersionID != nil {
|
|
return nil
|
|
}
|
|
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT id, doc_id, content_sha256, parent_snapshot_id, block_map, source_kind,
|
|
actor_user_id, actor_ref, actor_snapshot, source_run_id, created_at
|
|
FROM document_snapshots WHERE doc_id = $1 ORDER BY id`, docID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
|
|
type snapRow struct {
|
|
snap Snapshot
|
|
}
|
|
var snaps []snapRow
|
|
shas := map[string]struct{}{}
|
|
for rows.Next() {
|
|
var sn Snapshot
|
|
var blockRaw []byte
|
|
var actorRaw []byte
|
|
if err := rows.Scan(&sn.ID, &sn.DocID, &sn.ContentSHA256, &sn.ParentSnapshotID, &blockRaw,
|
|
&sn.SourceKind, &sn.ActorUserID, &sn.ActorRef, &actorRaw, &sn.SourceRunID, &sn.CreatedAt); err != nil {
|
|
return err
|
|
}
|
|
_ = json.Unmarshal(blockRaw, &sn.BlockMap)
|
|
_ = json.Unmarshal(actorRaw, &sn.ActorSnapshot)
|
|
snaps = append(snaps, snapRow{snap: sn})
|
|
shas[sn.ContentSHA256] = struct{}{}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return err
|
|
}
|
|
if len(snaps) == 0 {
|
|
return nil
|
|
}
|
|
|
|
for sha := range shas {
|
|
var md string
|
|
var byteSize int64
|
|
var created time.Time
|
|
err := s.pool.QueryRow(ctx, `
|
|
SELECT content_markdown, byte_size, created_at FROM document_content_blobs WHERE content_sha256 = $1`, sha).
|
|
Scan(&md, &byteSize, &created)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.blobs[sha] = &ContentBlob{ContentSHA256: sha, ContentMarkdown: md, ByteSize: byteSize, CreatedAt: created}
|
|
}
|
|
for i := range snaps {
|
|
cp := snaps[i].snap
|
|
s.snapshots[cp.ID] = &cp
|
|
if cp.ID > s.nextSnapID {
|
|
s.nextSnapID = cp.ID
|
|
}
|
|
}
|
|
|
|
vrows, err := s.pool.Query(ctx, `
|
|
SELECT id, doc_id, version_no, status, origin_kind, baseline_snapshot_id, head_snapshot_id,
|
|
human_revision_count, source_run_id, change_report_id, opened_at, sealed_at
|
|
FROM document_versions WHERE doc_id = $1 ORDER BY version_no`, docID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer vrows.Close()
|
|
var openVer *Version
|
|
for vrows.Next() {
|
|
var v Version
|
|
if err := vrows.Scan(&v.ID, &v.DocID, &v.VersionNo, &v.Status, &v.OriginKind, &v.BaselineSnapshotID,
|
|
&v.HeadSnapshotID, &v.HumanRevisionCount, &v.SourceRunID, &v.ChangeReportID, &v.OpenedAt, &v.SealedAt); err != nil {
|
|
return err
|
|
}
|
|
cp := v
|
|
s.versions[cp.ID] = &cp
|
|
if cp.ID > s.nextVerID {
|
|
s.nextVerID = cp.ID
|
|
}
|
|
if cp.Status == VersionOpen {
|
|
openVer = &cp
|
|
}
|
|
}
|
|
if err := vrows.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
erows, err := s.pool.Query(ctx, `
|
|
SELECT id, doc_id, version_id, event_no, from_snapshot_id, to_snapshot_id, change_kind, actor_kind,
|
|
actor_user_id, actor_ref, actor_snapshot, change_set, edit_session_id, request_id, idempotency_key, created_at
|
|
FROM document_change_events WHERE doc_id = $1 ORDER BY event_no`, docID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer erows.Close()
|
|
for erows.Next() {
|
|
var ev ChangeEvent
|
|
var actorRaw, changeRaw []byte
|
|
if err := erows.Scan(&ev.ID, &ev.DocID, &ev.VersionID, &ev.EventNo, &ev.FromSnapshotID, &ev.ToSnapshotID,
|
|
&ev.ChangeKind, &ev.ActorKind, &ev.ActorUserID, &ev.ActorRef, &actorRaw, &changeRaw,
|
|
&ev.EditSessionID, &ev.RequestID, &ev.IdempotencyKey, &ev.CreatedAt); err != nil {
|
|
return err
|
|
}
|
|
_ = json.Unmarshal(actorRaw, &ev.ActorSnapshot)
|
|
_ = json.Unmarshal(changeRaw, &ev.ChangeSet)
|
|
cp := ev
|
|
s.events[cp.ID] = &cp
|
|
if cp.IdempotencyKey != "" {
|
|
s.idemp[cp.IdempotencyKey] = cp.ID
|
|
}
|
|
if cp.ID > s.nextEvtID {
|
|
s.nextEvtID = cp.ID
|
|
}
|
|
}
|
|
if err := erows.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
rrows, err := s.pool.Query(ctx, `
|
|
SELECT id, doc_id, from_snapshot_id, to_snapshot_id, schema_version, summary, risk_level, risk_items,
|
|
project_impacts, document_impacts, open_questions, analysis_status, model_snapshot, prompt_version,
|
|
deterministic_facts, corrects_report_id, raw_json, created_at
|
|
FROM document_change_reports WHERE doc_id = $1`, docID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rrows.Close()
|
|
for rrows.Next() {
|
|
var rep ChangeReport
|
|
var sumRaw, riskRaw, projRaw, docRaw, qRaw, factRaw []byte
|
|
if err := rrows.Scan(&rep.ID, &rep.DocID, &rep.FromSnapshotID, &rep.ToSnapshotID, &rep.SchemaVersion,
|
|
&sumRaw, &rep.RiskLevel, &riskRaw, &projRaw, &docRaw, &qRaw, &rep.AnalysisStatus,
|
|
&rep.ModelSnapshot, &rep.PromptVersion, &factRaw, &rep.CorrectsReportID, &rep.RawJSON, &rep.CreatedAt); err != nil {
|
|
return err
|
|
}
|
|
_ = json.Unmarshal(sumRaw, &rep.Summary)
|
|
_ = json.Unmarshal(riskRaw, &rep.RiskItems)
|
|
_ = json.Unmarshal(projRaw, &rep.ProjectImpacts)
|
|
_ = json.Unmarshal(docRaw, &rep.DocumentImpacts)
|
|
_ = json.Unmarshal(qRaw, &rep.OpenQuestions)
|
|
_ = json.Unmarshal(factRaw, &rep.DeterministicFacts)
|
|
cp := rep
|
|
s.reports[cp.ID] = &cp
|
|
if cp.ID > s.nextRepID {
|
|
s.nextRepID = cp.ID
|
|
}
|
|
}
|
|
if err := rrows.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
crows, err := s.pool.Query(ctx, `
|
|
SELECT id, doc_id, input_snapshot_id, output_snapshot_id, run_id, proposal_id, status, authorized_scope,
|
|
quality_result, change_report_id, actor_kind, actor_snapshot, created_at, decided_at
|
|
FROM document_agent_candidates WHERE doc_id = $1`, docID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer crows.Close()
|
|
for crows.Next() {
|
|
var c AgentCandidate
|
|
var actorRaw []byte
|
|
if err := crows.Scan(&c.ID, &c.DocID, &c.InputSnapshotID, &c.OutputSnapshotID, &c.RunID, &c.ProposalID,
|
|
&c.Status, &c.AuthorizedScope, &c.QualityResult, &c.ChangeReportID, &c.ActorKind, &actorRaw,
|
|
&c.CreatedAt, &c.DecidedAt); err != nil {
|
|
return err
|
|
}
|
|
_ = json.Unmarshal(actorRaw, &c.ActorSnapshot)
|
|
cp := c
|
|
s.candidates[cp.ID] = &cp
|
|
if cp.ID > s.nextCandID {
|
|
s.nextCandID = cp.ID
|
|
}
|
|
}
|
|
if err := crows.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
root := &DocRootPointers{DocID: docID, RevisionSchemaVer: SchemaVersion, Kind: "controlled", LifecycleState: "drafting"}
|
|
if openVer != nil {
|
|
vid := openVer.ID
|
|
hid := openVer.HeadSnapshotID
|
|
root.LedgerVersionID = &vid
|
|
root.HeadSnapshotID = &hid
|
|
root.HeadRevisionNo = openVer.HumanRevisionCount
|
|
if head := s.snapshots[openVer.HeadSnapshotID]; head != nil {
|
|
root.ContentSHA256 = head.ContentSHA256
|
|
}
|
|
}
|
|
|
|
var ledgerID, headID *int64
|
|
var headRev int
|
|
var workDir, lifecycle string
|
|
var projectID int64
|
|
err = s.pool.QueryRow(ctx, `
|
|
SELECT ledger_version_id, head_snapshot_id, head_revision_no, COALESCE(work_dir,''), COALESCE(current_state,'drafting'), project_id
|
|
FROM document_records WHERE doc_id = $1`, docID).
|
|
Scan(&ledgerID, &headID, &headRev, &workDir, &lifecycle, &projectID)
|
|
if err == nil {
|
|
root.Kind = "controlled"
|
|
root.WorkDir = workDir
|
|
root.LifecycleState = lifecycle
|
|
root.ProjectID = projectID
|
|
if ledgerID != nil {
|
|
root.LedgerVersionID = ledgerID
|
|
}
|
|
if headID != nil {
|
|
root.HeadSnapshotID = headID
|
|
}
|
|
root.HeadRevisionNo = headRev
|
|
} else if errors.Is(err, pgx.ErrNoRows) {
|
|
var fLedger, fHead *int64
|
|
var fRev int
|
|
var fProj int64
|
|
ferr := s.pool.QueryRow(ctx, `
|
|
SELECT ledger_version_id, head_snapshot_id, head_revision_no, project_id
|
|
FROM freeform_documents WHERE doc_id = $1 AND NOT is_deleted`, docID).
|
|
Scan(&fLedger, &fHead, &fRev, &fProj)
|
|
if ferr == nil {
|
|
root.Kind = "freeform"
|
|
root.LifecycleState = "drafting"
|
|
root.ProjectID = fProj
|
|
if fLedger != nil {
|
|
root.LedgerVersionID = fLedger
|
|
}
|
|
if fHead != nil {
|
|
root.HeadSnapshotID = fHead
|
|
}
|
|
root.HeadRevisionNo = fRev
|
|
} else if !errors.Is(ferr, pgx.ErrNoRows) {
|
|
return ferr
|
|
}
|
|
} else {
|
|
return err
|
|
}
|
|
|
|
if root.LedgerVersionID != nil {
|
|
s.roots[docID] = root
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) ensureBlobTX(ctx context.Context, tx pgx.Tx, blob *ContentBlob) error {
|
|
_, err := tx.Exec(ctx, `
|
|
INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size, created_at)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (content_sha256) DO NOTHING`,
|
|
blob.ContentSHA256, blob.ContentMarkdown, blob.ByteSize, blob.CreatedAt)
|
|
return err
|
|
}
|
|
|
|
func (s *Service) insertSnapshotTX(ctx context.Context, tx pgx.Tx, sn *Snapshot) (int64, error) {
|
|
blockRaw, _ := json.Marshal(sn.BlockMap)
|
|
if blockRaw == nil {
|
|
blockRaw = []byte("[]")
|
|
}
|
|
actorRaw, _ := json.Marshal(sn.ActorSnapshot)
|
|
if actorRaw == nil {
|
|
actorRaw = []byte("{}")
|
|
}
|
|
var id int64
|
|
err := tx.QueryRow(ctx, `
|
|
INSERT INTO document_snapshots (
|
|
doc_id, content_sha256, parent_snapshot_id, block_map, source_kind,
|
|
actor_user_id, actor_ref, actor_snapshot, source_run_id, created_at
|
|
) VALUES ($1,$2,$3,$4::jsonb,$5,$6,$7,$8::jsonb,$9,$10) RETURNING id`,
|
|
sn.DocID, sn.ContentSHA256, sn.ParentSnapshotID, blockRaw, sn.SourceKind,
|
|
sn.ActorUserID, sn.ActorRef, actorRaw, sn.SourceRunID, sn.CreatedAt,
|
|
).Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
func (s *Service) insertVersionTX(ctx context.Context, tx pgx.Tx, v *Version) (int64, error) {
|
|
var id int64
|
|
err := tx.QueryRow(ctx, `
|
|
INSERT INTO document_versions (
|
|
doc_id, version_no, status, origin_kind, baseline_snapshot_id, head_snapshot_id,
|
|
human_revision_count, source_run_id, change_report_id, opened_at, sealed_at
|
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) RETURNING id`,
|
|
v.DocID, v.VersionNo, v.Status, v.OriginKind, v.BaselineSnapshotID, v.HeadSnapshotID,
|
|
v.HumanRevisionCount, v.SourceRunID, v.ChangeReportID, v.OpenedAt, v.SealedAt,
|
|
).Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
func (s *Service) updateVersionTX(ctx context.Context, tx pgx.Tx, v *Version) error {
|
|
_, err := tx.Exec(ctx, `
|
|
UPDATE document_versions SET status=$2, head_snapshot_id=$3, human_revision_count=$4,
|
|
change_report_id=$5, sealed_at=$6 WHERE id=$1`,
|
|
v.ID, v.Status, v.HeadSnapshotID, v.HumanRevisionCount, v.ChangeReportID, v.SealedAt)
|
|
return err
|
|
}
|
|
|
|
func (s *Service) insertEventTX(ctx context.Context, tx pgx.Tx, ev *ChangeEvent) (int64, error) {
|
|
actorRaw, _ := json.Marshal(ev.ActorSnapshot)
|
|
if actorRaw == nil {
|
|
actorRaw = []byte("{}")
|
|
}
|
|
changeRaw, _ := json.Marshal(ev.ChangeSet)
|
|
if changeRaw == nil {
|
|
changeRaw = []byte("[]")
|
|
}
|
|
var id int64
|
|
err := tx.QueryRow(ctx, `
|
|
INSERT INTO document_change_events (
|
|
doc_id, version_id, event_no, from_snapshot_id, to_snapshot_id, change_kind, actor_kind,
|
|
actor_user_id, actor_ref, actor_snapshot, change_set, edit_session_id, request_id, idempotency_key, created_at
|
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,$11::jsonb,$12,$13,$14,$15) RETURNING id`,
|
|
ev.DocID, ev.VersionID, ev.EventNo, ev.FromSnapshotID, ev.ToSnapshotID, ev.ChangeKind, ev.ActorKind,
|
|
ev.ActorUserID, ev.ActorRef, actorRaw, changeRaw, ev.EditSessionID, ev.RequestID, ev.IdempotencyKey, ev.CreatedAt,
|
|
).Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
func (s *Service) insertReportTX(ctx context.Context, tx pgx.Tx, rep *ChangeReport) (int64, error) {
|
|
sumRaw, _ := json.Marshal(rep.Summary)
|
|
if sumRaw == nil {
|
|
sumRaw = []byte("[]")
|
|
}
|
|
riskRaw, _ := json.Marshal(rep.RiskItems)
|
|
if riskRaw == nil {
|
|
riskRaw = []byte("[]")
|
|
}
|
|
projRaw, _ := json.Marshal(rep.ProjectImpacts)
|
|
if projRaw == nil {
|
|
projRaw = []byte("{}")
|
|
}
|
|
docRaw, _ := json.Marshal(rep.DocumentImpacts)
|
|
if docRaw == nil {
|
|
docRaw = []byte("[]")
|
|
}
|
|
qRaw, _ := json.Marshal(rep.OpenQuestions)
|
|
if qRaw == nil {
|
|
qRaw = []byte("[]")
|
|
}
|
|
factRaw, _ := json.Marshal(rep.DeterministicFacts)
|
|
if factRaw == nil {
|
|
factRaw = []byte("{}")
|
|
}
|
|
var id int64
|
|
err := tx.QueryRow(ctx, `
|
|
INSERT INTO document_change_reports (
|
|
doc_id, from_snapshot_id, to_snapshot_id, schema_version, summary, risk_level, risk_items,
|
|
project_impacts, document_impacts, open_questions, analysis_status, model_snapshot, prompt_version,
|
|
deterministic_facts, corrects_report_id, raw_json, created_at
|
|
) VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7::jsonb,$8::jsonb,$9::jsonb,$10::jsonb,$11,$12,$13,$14::jsonb,$15,$16,$17)
|
|
RETURNING id`,
|
|
rep.DocID, rep.FromSnapshotID, rep.ToSnapshotID, rep.SchemaVersion, sumRaw, rep.RiskLevel, riskRaw,
|
|
projRaw, docRaw, qRaw, rep.AnalysisStatus, rep.ModelSnapshot, rep.PromptVersion,
|
|
factRaw, rep.CorrectsReportID, rep.RawJSON, rep.CreatedAt,
|
|
).Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
func (s *Service) insertCandidateTX(ctx context.Context, tx pgx.Tx, c *AgentCandidate) (int64, error) {
|
|
actorRaw, _ := json.Marshal(c.ActorSnapshot)
|
|
if actorRaw == nil {
|
|
actorRaw = []byte("{}")
|
|
}
|
|
var id int64
|
|
err := tx.QueryRow(ctx, `
|
|
INSERT INTO document_agent_candidates (
|
|
doc_id, input_snapshot_id, output_snapshot_id, run_id, proposal_id, status, authorized_scope,
|
|
quality_result, change_report_id, actor_kind, actor_snapshot, created_at, decided_at
|
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::jsonb,$12,$13) RETURNING id`,
|
|
c.DocID, c.InputSnapshotID, c.OutputSnapshotID, c.RunID, c.ProposalID, c.Status, c.AuthorizedScope,
|
|
c.QualityResult, c.ChangeReportID, c.ActorKind, actorRaw, c.CreatedAt, c.DecidedAt,
|
|
).Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
func (s *Service) updateCandidateTX(ctx context.Context, tx pgx.Tx, c *AgentCandidate) error {
|
|
_, err := tx.Exec(ctx, `
|
|
UPDATE document_agent_candidates SET status=$2, decided_at=$3, change_report_id=$4, output_snapshot_id=$5
|
|
WHERE id=$1`, c.ID, c.Status, c.DecidedAt, c.ChangeReportID, c.OutputSnapshotID)
|
|
return err
|
|
}
|
|
|
|
func (s *Service) projectRootPointersTX(ctx context.Context, tx pgx.Tx, root *DocRootPointers) error {
|
|
if root == nil {
|
|
return nil
|
|
}
|
|
switch root.Kind {
|
|
case "freeform":
|
|
_, err := tx.Exec(ctx, `
|
|
UPDATE freeform_documents
|
|
SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4,
|
|
revision_schema_version=$5, row_revision=COALESCE(row_revision,0)+1
|
|
WHERE doc_id=$1`,
|
|
root.DocID, root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer)
|
|
return err
|
|
default:
|
|
_, err := tx.Exec(ctx, `
|
|
UPDATE document_records
|
|
SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4,
|
|
revision_schema_version=$5, row_revision=COALESCE(row_revision,0)+1,
|
|
content_sha256=COALESCE(NULLIF($6,''), content_sha256)
|
|
WHERE doc_id=$1`,
|
|
root.DocID, root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo,
|
|
root.RevisionSchemaVer, root.ContentSHA256)
|
|
return err
|
|
}
|
|
}
|
|
|
|
func (s *Service) withTx(ctx context.Context, fn func(pgx.Tx) error) error {
|
|
if !s.usingPG() {
|
|
return nil
|
|
}
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
if err := fn(tx); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
// Pool exposes the underlying pool for integration tests.
|
|
func (s *Service) Pool() *pgxpool.Pool { return s.pool }
|