Agent-System/internal/docrevision/pg_mutate.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(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(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(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(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)
		}
		return s.projectRootPointersTX(ctx, tx, root)
	})
}

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
}