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
		}
		sn.ID = id
		s.snapshots[id] = sn

		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
		}
		ver.ID = vid
		s.versions[vid] = ver

		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
		}
		ev.ID = eid
		s.events[eid] = ev
		s.idemp[ev.IdempotencyKey] = 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) error {
	if !s.usingPG() {
		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
		}
		oldSnapID := snap.ID
		id, err := s.insertSnapshotTX(ctx, tx, snap)
		if err != nil {
			return err
		}
		if oldSnapID != 0 && oldSnapID != id {
			delete(s.snapshots, oldSnapID)
		}
		snap.ID = id
		s.snapshots[id] = snap

		if isBaseline {
			ver.BaselineSnapshotID = id
			ver.HeadSnapshotID = id
			oldVerID := ver.ID
			vid, err := s.insertVersionTX(ctx, tx, ver)
			if err != nil {
				return err
			}
			if oldVerID != 0 && oldVerID != vid {
				delete(s.versions, oldVerID)
			}
			ver.ID = vid
			s.versions[vid] = ver
			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
		}

		oldEv := ev.ID
		eid, err := s.insertEventTX(ctx, tx, ev)
		if err != nil {
			return err
		}
		if oldEv != 0 && oldEv != eid {
			delete(s.events, oldEv)
		}
		ev.ID = eid
		s.events[eid] = ev
		if ev.IdempotencyKey != "" {
			s.idemp[ev.IdempotencyKey] = eid
		}
		return s.projectRootPointersTX(ctx, tx, root)
	})
}

func (s *Service) persistAgentCandidateLocked(ctx context.Context, outSnap *Snapshot, rep *ChangeReport, c *AgentCandidate) 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
		}
		oldSnap := outSnap.ID
		sid, err := s.insertSnapshotTX(ctx, tx, outSnap)
		if err != nil {
			return err
		}
		if oldSnap != 0 && oldSnap != sid {
			delete(s.snapshots, oldSnap)
		}
		outSnap.ID = sid
		s.snapshots[sid] = outSnap
		c.OutputSnapshotID = &sid
		rep.ToSnapshotID = sid

		oldRep := rep.ID
		rid, err := s.insertReportTX(ctx, tx, rep)
		if err != nil {
			return err
		}
		if oldRep != 0 && oldRep != rid {
			delete(s.reports, oldRep)
		}
		rep.ID = rid
		s.reports[rid] = rep
		c.ChangeReportID = &rid

		oldC := c.ID
		cid, err := s.insertCandidateTX(ctx, tx, c)
		if err != nil {
			return err
		}
		if oldC != 0 && oldC != cid {
			delete(s.candidates, oldC)
		}
		c.ID = cid
		s.candidates[cid] = c
		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
		}
		oldVerID := newVer.ID
		vid, err := s.insertVersionTX(ctx, tx, newVer)
		if err != nil {
			return err
		}
		if oldVerID != 0 && oldVerID != vid {
			delete(s.versions, oldVerID)
		}
		newVer.ID = vid
		s.versions[vid] = newVer
		root.LedgerVersionID = &vid
		hid := outSnap.ID
		root.HeadSnapshotID = &hid

		ev.VersionID = vid
		ev.ToSnapshotID = outSnap.ID
		oldEv := ev.ID
		eid, err := s.insertEventTX(ctx, tx, ev)
		if err != nil {
			return err
		}
		if oldEv != 0 {
			delete(s.events, oldEv)
		}
		ev.ID = eid
		s.events[eid] = ev

		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
}