weichen
3e01f18756
User authorized lifting CURRENT FAV whole-branch merge protection on 2026-09-23. Keep FAV_DEGRADED_READ/FAV_PUBLISHER_ENABLED default false. Drop duplicate goose 00034 (local_dev already has 00036). Preserve agentx workspace admin routes and pin-export wiring.
290 lines
8.3 KiB
Go
290 lines
8.3 KiB
Go
package docrevision
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
func (s *Service) openTemplatePGLocked(ctx context.Context, docID, canon, sha string, actor ActorSnapshot, projectID int64, workDir string) (*RevisionState, error) {
|
|
blob := &ContentBlob{ContentSHA256: sha, ContentMarkdown: canon, ByteSize: int64(len(canon)), CreatedAt: time.Now().UTC()}
|
|
blocks := BuildBlockMap(canon, nil)
|
|
err := s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
|
|
if err := s.lockOpenVersionTX(ctx, tx, docID); err != nil {
|
|
return err
|
|
}
|
|
if err := s.ensureBlobTX(ctx, tx, blob); err != nil {
|
|
return err
|
|
}
|
|
sn := &Snapshot{
|
|
DocID: docID, ContentSHA256: sha, BlockMap: blocks, SourceKind: SourceTemplate,
|
|
ActorRef: "template", ActorSnapshot: actor, CreatedAt: time.Now().UTC(),
|
|
}
|
|
id, err := s.insertSnapshotTX(ctx, tx, sn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptSnapshotID(sn, id)
|
|
|
|
ver := &Version{
|
|
DocID: docID, VersionNo: 1, Status: VersionOpen, OriginKind: OriginTemplate,
|
|
BaselineSnapshotID: id, HeadSnapshotID: id, OpenedAt: time.Now().UTC(),
|
|
}
|
|
vid, err := s.insertVersionTX(ctx, tx, ver)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptVersionID(ver, vid)
|
|
|
|
ev := &ChangeEvent{
|
|
DocID: docID, VersionID: vid, EventNo: 1, ToSnapshotID: id,
|
|
ChangeKind: ChangeImport, ActorKind: ActorTemplate, ActorRef: "template",
|
|
ActorSnapshot: actor, IdempotencyKey: fmt.Sprintf("template:%s:%s", docID, sha),
|
|
CreatedAt: time.Now().UTC(),
|
|
}
|
|
eid, err := s.insertEventTX(ctx, tx, ev)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptEventID(ev, eid)
|
|
|
|
root := &DocRootPointers{
|
|
DocID: docID, Kind: "controlled", ProjectID: projectID, WorkDir: workDir,
|
|
LedgerVersionID: &vid, HeadSnapshotID: &id, HeadRevisionNo: 0,
|
|
ContentSHA256: sha, LifecycleState: "drafting", RevisionSchemaVer: SchemaVersion,
|
|
}
|
|
s.roots[docID] = root
|
|
s.blobs[sha] = blob
|
|
return s.projectRootPointersTX(ctx, tx, root)
|
|
})
|
|
if err != nil {
|
|
s.forgetDocLocked(docID)
|
|
return nil, err
|
|
}
|
|
return s.mustStateLocked(docID), nil
|
|
}
|
|
|
|
func (s *Service) persistHumanRevisionLocked(ctx context.Context, root *DocRootPointers, ver *Version, snap *Snapshot, ev *ChangeEvent, isBaseline bool, rep *ChangeReport) error {
|
|
if !s.usingPG() {
|
|
if root != nil && snap != nil {
|
|
key := ""
|
|
if ev != nil {
|
|
key = ev.IdempotencyKey
|
|
}
|
|
s.noteContentProjectionLocked(root.DocID, snap.ID, root.HeadRevisionNo, key)
|
|
}
|
|
if rep != nil {
|
|
rep.ToSnapshotID = snap.ID
|
|
s.assignReportIDLocked(rep)
|
|
}
|
|
return nil
|
|
}
|
|
blob := s.blobs[snap.ContentSHA256]
|
|
return s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
|
|
if !isBaseline {
|
|
if err := s.lockOpenVersionTX(ctx, tx, root.DocID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := s.ensureBlobTX(ctx, tx, blob); err != nil {
|
|
return err
|
|
}
|
|
id, err := s.insertSnapshotTX(ctx, tx, snap)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptSnapshotID(snap, id)
|
|
|
|
if isBaseline {
|
|
ver.BaselineSnapshotID = id
|
|
ver.HeadSnapshotID = id
|
|
vid, err := s.insertVersionTX(ctx, tx, ver)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptVersionID(ver, vid)
|
|
root.LedgerVersionID = &vid
|
|
root.HeadSnapshotID = &id
|
|
ev.VersionID = vid
|
|
ev.ToSnapshotID = id
|
|
} else {
|
|
ver.HeadSnapshotID = id
|
|
root.HeadSnapshotID = &id
|
|
if err := s.updateVersionTX(ctx, tx, ver); err != nil {
|
|
return err
|
|
}
|
|
ev.ToSnapshotID = id
|
|
ev.VersionID = ver.ID
|
|
}
|
|
|
|
eid, err := s.insertEventTX(ctx, tx, ev)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptEventID(ev, eid)
|
|
key := ""
|
|
if ev != nil {
|
|
key = ev.IdempotencyKey
|
|
}
|
|
if err := s.insertContentProjectionTX(ctx, tx, root.DocID, snap.ID, root.HeadRevisionNo, key); err != nil {
|
|
return err
|
|
}
|
|
if rep != nil {
|
|
rep.ToSnapshotID = snap.ID
|
|
rid, err := s.insertReportTX(ctx, tx, rep)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptReportID(rep, rid)
|
|
s.reports[rep.ID] = rep
|
|
}
|
|
s.noteContentProjectionLocked(root.DocID, snap.ID, root.HeadRevisionNo, key)
|
|
return s.projectRootPointersTX(ctx, tx, root)
|
|
})
|
|
}
|
|
|
|
func (s *Service) insertContentProjectionTX(ctx context.Context, tx pgx.Tx, docID string, snapshotID int64, revisionNo int, idempotencyKey string) error {
|
|
if docID == "" || snapshotID == 0 {
|
|
return nil
|
|
}
|
|
_, err := tx.Exec(ctx, `
|
|
INSERT INTO document_content_projections (doc_id, snapshot_id, revision_no, idempotency_key, status)
|
|
VALUES ($1, $2, $3, $4, 'pending')
|
|
ON CONFLICT (doc_id, snapshot_id) DO NOTHING`, docID, snapshotID, revisionNo, idempotencyKey)
|
|
return err
|
|
}
|
|
|
|
func (s *Service) persistAgentCandidateLocked(ctx context.Context, outSnap *Snapshot, rep *ChangeReport, c *AgentCandidate, superseded []*ChangeReport) error {
|
|
if !s.usingPG() {
|
|
return nil
|
|
}
|
|
blob := s.blobs[outSnap.ContentSHA256]
|
|
return s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
|
|
if err := s.lockOpenVersionTX(ctx, tx, c.DocID); err != nil {
|
|
return err
|
|
}
|
|
if err := s.ensureBlobTX(ctx, tx, blob); err != nil {
|
|
return err
|
|
}
|
|
sid, err := s.insertSnapshotTX(ctx, tx, outSnap)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptSnapshotID(outSnap, sid)
|
|
c.OutputSnapshotID = &outSnap.ID
|
|
rep.ToSnapshotID = sid
|
|
|
|
rid, err := s.insertReportTX(ctx, tx, rep)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptReportID(rep, rid)
|
|
c.ChangeReportID = &rep.ID
|
|
|
|
cid, err := s.insertCandidateTX(ctx, tx, c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptCandidateID(c, cid)
|
|
for _, old := range superseded {
|
|
if err := s.updateReportFactsTX(ctx, tx, old); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func (s *Service) persistAgentCommitLocked(ctx context.Context, root *DocRootPointers, oldVer, newVer *Version, outSnap *Snapshot, ev *ChangeEvent, c *AgentCandidate) error {
|
|
if !s.usingPG() {
|
|
return nil
|
|
}
|
|
return s.withTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
|
|
if err := s.lockOpenVersionTX(ctx, tx, root.DocID); err != nil {
|
|
return err
|
|
}
|
|
if err := s.updateVersionTX(ctx, tx, oldVer); err != nil {
|
|
return err
|
|
}
|
|
if newVer.ChangeReportID != nil {
|
|
if auth := s.reports[*newVer.ChangeReportID]; auth != nil && auth.Authority == ReportAuthoritative {
|
|
rid, err := s.insertReportTX(ctx, tx, auth)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptReportID(auth, rid)
|
|
newVer.ChangeReportID = &auth.ID
|
|
if c.ChangeReportID != nil {
|
|
if proposed := s.reports[*c.ChangeReportID]; proposed != nil {
|
|
proposed.SuccessorReportID = &auth.ID
|
|
}
|
|
}
|
|
for _, other := range s.reports {
|
|
if other == nil || other.DocID != root.DocID || other.Authority != ReportProposed {
|
|
continue
|
|
}
|
|
if err := s.updateReportFactsTX(ctx, tx, other); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
vid, err := s.insertVersionTX(ctx, tx, newVer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptVersionID(newVer, vid)
|
|
root.LedgerVersionID = &newVer.ID
|
|
hid := outSnap.ID
|
|
root.HeadSnapshotID = &hid
|
|
|
|
ev.VersionID = vid
|
|
ev.ToSnapshotID = outSnap.ID
|
|
eid, err := s.insertEventTX(ctx, tx, ev)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.adoptEventID(ev, eid)
|
|
|
|
if err := s.updateCandidateTX(ctx, tx, c); err != nil {
|
|
return err
|
|
}
|
|
if newVer.ChangeReportID != nil {
|
|
_, _ = tx.Exec(ctx, `UPDATE document_versions SET change_report_id=$2 WHERE id=$1`, vid, *newVer.ChangeReportID)
|
|
}
|
|
if err := s.projectRootPointersTX(ctx, tx, root); err != nil {
|
|
return err
|
|
}
|
|
if s.OnFinalArtifactTX == nil {
|
|
return nil
|
|
}
|
|
eventID := fmt.Sprintf("ev:%d", ev.ID)
|
|
outMD := []byte(nil)
|
|
if b := s.blobs[outSnap.ContentSHA256]; b != nil {
|
|
outMD = []byte(b.ContentMarkdown)
|
|
}
|
|
if err := s.OnFinalArtifactTX(ctx, tx, "agent_commit", root.DocID, root.ProjectID, outSnap.ID, eventID, outMD, nil); err != nil {
|
|
return err
|
|
}
|
|
if oldVer != nil && oldVer.SealedAt != nil {
|
|
inMD := []byte(nil)
|
|
if snap := s.snapshots[oldVer.HeadSnapshotID]; snap != nil {
|
|
if b := s.blobs[snap.ContentSHA256]; b != nil {
|
|
inMD = []byte(b.ContentMarkdown)
|
|
}
|
|
}
|
|
return s.OnFinalArtifactTX(ctx, tx, "version_sealed", root.DocID, root.ProjectID, oldVer.HeadSnapshotID, eventID+":sealed", inMD, oldVer.SealedAt)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
func (s *Service) lockOpenVersionTX(ctx context.Context, tx pgx.Tx, docID string) error {
|
|
if _, err := tx.Exec(ctx, `SELECT id FROM document_versions WHERE doc_id=$1 AND status='open' FOR UPDATE`, docID); err != nil {
|
|
return fmt.Errorf("docrevision: lock open version: %w", err)
|
|
}
|
|
return nil
|
|
}
|