diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 56d6db2b..f523063e 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -54,13 +54,18 @@ imports: - storage db: [] docduty: [] + app: [] + app/uow: [] docgov: + - app/uow - docduty - docrevision - fsutil docrevision: + - app/uow - fsutil freedoc: + - app/uow - ctxmgr - docrevision - replyexport @@ -116,6 +121,7 @@ imports: storage: [] stream: [] taskrun: + - app/uow - bizdev - ctxmgr - projectworkflow @@ -179,6 +185,12 @@ event_consumers: - server - notification +unwrap_owners: + - docgov + - docrevision + - taskrun + - freedoc + exceptions: - kind: sql_write package: docduty diff --git a/internal/app/uow/pg_uow.go b/internal/app/uow/pg_uow.go new file mode 100644 index 00000000..62d246ee --- /dev/null +++ b/internal/app/uow/pg_uow.go @@ -0,0 +1,172 @@ +package uow + +import ( + "context" + "errors" + "fmt" + "math/rand" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +type pgScope struct { + ctx context.Context + tx pgx.Tx + done bool +} + +func (s *pgScope) Context() context.Context { return s.ctx } + +// AsPG returns the bound pgx.Tx for owner repositories only. +// Application / docflow code must not call this; archtest enforces unwrap owners. +func AsPG(scope TxScope) (pgx.Tx, error) { + s, ok := scope.(*pgScope) + if !ok || s == nil || s.tx == nil { + return nil, fmt.Errorf("uow: scope is not a postgres transaction") + } + if s.done { + return nil, ErrClosedScope + } + return s.tx, nil +} + +type PG struct { + Pool *pgxpool.Pool + Policy RetryPolicy + Now func() time.Time + Sleep func(time.Duration) + Rand *rand.Rand + BeforeBegin func(context.Context) error + AfterBegin func(TxScope) error + BeforeCommit func(TxScope) error +} + +func NewPG(pool *pgxpool.Pool) *PG { + return &PG{Pool: pool, Policy: DefaultRetryPolicy()} +} + +func (p *PG) Within(ctx context.Context, fn func(TxScope) error) error { + if p == nil || p.Pool == nil { + return fmt.Errorf("uow: postgres pool required") + } + if fn == nil { + return fmt.Errorf("uow: command required") + } + policy := p.Policy + if policy.MaxAttempts < 1 { + policy.MaxAttempts = DefaultMaxAttempts + } + if policy.LockTimeout <= 0 { + policy.LockTimeout = DefaultLockTimeout + } + var last error + for attempt := 1; attempt <= policy.MaxAttempts; attempt++ { + last = p.one(ctx, policy.LockTimeout, fn) + if last == nil { + return nil + } + if !IsSerialization(last) || IsNonRetryable(last) { + return last + } + if attempt == policy.MaxAttempts { + return last + } + p.backoff(attempt) + } + return last +} + +func (p *PG) one(ctx context.Context, lockWait time.Duration, fn func(TxScope) error) error { + if p.BeforeBegin != nil { + if err := p.BeforeBegin(ctx); err != nil { + return err + } + } + if dl, ok := ctx.Deadline(); ok { + remain := time.Until(dl) + if remain > 0 && lockWait > remain { + lockWait = remain + } + } + tx, err := p.Pool.Begin(ctx) + if err != nil { + return err + } + scope := &pgScope{ctx: ctx, tx: tx} + defer func() { + scope.done = true + _ = tx.Rollback(ctx) + }() + if _, err := tx.Exec(ctx, fmt.Sprintf("SET LOCAL lock_timeout = '%dms'", lockWait.Milliseconds())); err != nil { + return err + } + if p.AfterBegin != nil { + if err := p.AfterBegin(scope); err != nil { + return err + } + } + if err := fn(scope); err != nil { + return classifyLock(err) + } + if p.BeforeCommit != nil { + if err := p.BeforeCommit(scope); err != nil { + return err + } + } + if err := tx.Commit(ctx); err != nil { + return classifyLock(err) + } + scope.done = true + return nil +} + +func (p *PG) backoff(attempt int) { + sleep := p.Sleep + if sleep == nil { + sleep = time.Sleep + } + n := 10 + if p.Rand != nil { + n = 5 + p.Rand.Intn(20) + } else { + n = 5 + rand.Intn(20) + } + sleep(time.Duration(n*attempt) * time.Millisecond) +} + +func IsSerialization(err error) bool { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + return pgErr.Code == "40001" || pgErr.Code == "40P01" + } + return false +} + +func IsNonRetryable(err error) bool { + return errors.Is(err, ErrVersionConflict) || + errors.Is(err, ErrPermissionDenied) || + errors.Is(err, ErrDigestConflict) +} + +func classifyLock(err error) error { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + switch pgErr.Code { + case "55P03": // lock_not_available + return ErrBusy + case "57014": // query_canceled / statement timeout + if strings.Contains(strings.ToLower(pgErr.Message), "lock") { + return ErrBusy + } + } + } + msg := strings.ToLower(err.Error()) + if strings.Contains(msg, "lock timeout") || strings.Contains(msg, "canceling statement due to lock timeout") { + return ErrBusy + } + return err +} diff --git a/internal/app/uow/pg_uow_test.go b/internal/app/uow/pg_uow_test.go new file mode 100644 index 00000000..53ef3fea --- /dev/null +++ b/internal/app/uow/pg_uow_test.go @@ -0,0 +1,267 @@ +package uow_test + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "os" + "testing" + "time" + + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/db" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/docrevision" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +func testPool(t *testing.T) (*pgxpool.Pool, context.Context) { + t.Helper() + if testing.Short() { + t.Skip("postgresql fault injection is not part of -short") + } + dsn := db.SkipIfNoTestDatabaseURL(t) + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil { + t.Fatalf("migrate: %v", err) + } + return pool, ctx +} + +func init() { + // ResolveMigrationsDir is cwd-relative; tests in this package start under internal/app/uow. + if _, err := os.Stat("deploy/postgres/goose"); err == nil { + return + } + for _, cand := range []string{"../../..", "../../../.."} { + if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() { + _ = os.Chdir(cand) + return + } + } +} + +func seedProject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix string) int64 { + t.Helper() + var projectID int64 + err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status) + VALUES ($1, $2, 'active') + RETURNING id + `, "uow-"+suffix, "uow "+suffix).Scan(&projectID) + if err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM document_change_events WHERE doc_id LIKE $1`, "uow_"+suffix+"%") + _, _ = pool.Exec(context.Background(), `DELETE FROM document_versions WHERE doc_id LIKE $1`, "uow_"+suffix+"%") + _, _ = pool.Exec(context.Background(), `DELETE FROM document_snapshots WHERE doc_id LIKE $1`, "uow_"+suffix+"%") + _, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE project_id=$1`, projectID) + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id=$1`, projectID) + }) + return projectID +} + +func shaBody(body string) string { + sum := sha256.Sum256([]byte(body)) + return hex.EncodeToString(sum[:]) +} + +func writeOwners(ctx context.Context, scope uow.TxScope, projectID int64, docID, body string) error { + if err := (docgov.Repository{}).InsertRoot(ctx, scope, docgov.DocumentRecord{ + DocID: docID, ProjectID: projectID, TaskKey: "intake", Title: docID, + ContentSHA256: shaBody(body), WorkDir: "ws/" + docID, + }); err != nil { + return err + } + sha := shaBody(body) + if err := (docrevision.Repository{}).InsertBlob(ctx, scope, sha, body); err != nil { + return err + } + sid, err := (docrevision.Repository{}).InsertSnapshot(ctx, scope, docrevision.Snapshot{ + DocID: docID, ContentSHA256: sha, SourceKind: docrevision.SourceTemplate, + ActorRef: "template", CreatedAt: time.Now().UTC(), + }) + if err != nil { + return err + } + vid, err := (docrevision.Repository{}).InsertVersion(ctx, scope, docrevision.Version{ + DocID: docID, VersionNo: 1, Status: docrevision.VersionOpen, OriginKind: docrevision.OriginTemplate, + BaselineSnapshotID: sid, HeadSnapshotID: sid, OpenedAt: time.Now().UTC(), + }) + if err != nil { + return err + } + _, err = (docrevision.Repository{}).InsertEvent(ctx, scope, docrevision.ChangeEvent{ + DocID: docID, VersionID: vid, EventNo: 1, ToSnapshotID: sid, + ChangeKind: docrevision.ChangeImport, ActorKind: docrevision.ActorTemplate, + ActorRef: "template", IdempotencyKey: "uow:" + docID, CreatedAt: time.Now().UTC(), + }) + return err +} + +func countDoc(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string) (records, snaps, events int) { + t.Helper() + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM document_records WHERE doc_id=$1`, docID).Scan(&records); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM document_snapshots WHERE doc_id=$1`, docID).Scan(&snaps); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM document_change_events WHERE doc_id=$1`, docID).Scan(&events); err != nil { + t.Fatal(err) + } + return +} + +func TestSecondOwnerFailureRollsBackFirst(t *testing.T) { + pool, ctx := testPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + projectID := seedProject(t, ctx, pool, suffix) + docID := "uow_" + suffix + "_rollback" + work := uow.NewPG(pool) + err := work.Within(ctx, func(scope uow.TxScope) error { + if err := (docgov.Repository{}).InsertRoot(ctx, scope, docgov.DocumentRecord{ + DocID: docID, ProjectID: projectID, Title: docID, TaskKey: "intake", + }); err != nil { + return err + } + return errors.New("ledger write failed") + }) + if err == nil { + t.Fatal("expected failure") + } + r, s, e := countDoc(t, ctx, pool, docID) + if r != 0 || s != 0 || e != 0 { + t.Fatalf("partial commit records=%d snaps=%d events=%d", r, s, e) + } +} + +func TestStagingFailureLeavesNoLedgerOrRoot(t *testing.T) { + pool, ctx := testPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + projectID := seedProject(t, ctx, pool, suffix) + docID := "uow_" + suffix + "_staging" + work := uow.NewPG(pool) + err := work.Within(ctx, func(scope uow.TxScope) error { + if err := writeOwners(ctx, scope, projectID, docID, "# staging\n"); err != nil { + return err + } + return fmt.Errorf("outbox/staging write failed") + }) + if err == nil { + t.Fatal("expected staging failure") + } + r, s, e := countDoc(t, ctx, pool, docID) + if r != 0 || s != 0 || e != 0 { + t.Fatalf("staging failure leaked rows records=%d snaps=%d events=%d", r, s, e) + } +} + +func TestCommitFailureHasNoSideEffects(t *testing.T) { + pool, ctx := testPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + projectID := seedProject(t, ctx, pool, suffix) + docID := "uow_" + suffix + "_commit" + work := uow.NewPG(pool) + work.BeforeCommit = func(uow.TxScope) error { return errors.New("commit failed") } + sideEffect := false + err := work.Within(ctx, func(scope uow.TxScope) error { + return writeOwners(ctx, scope, projectID, docID, "# commit\n") + }) + if err == nil { + sideEffect = true + } + if sideEffect { + t.Fatal("caller must see commit error and skip post-commit work") + } + r, s, e := countDoc(t, ctx, pool, docID) + if r != 0 || s != 0 || e != 0 { + t.Fatalf("commit failure leaked rows records=%d snaps=%d events=%d", r, s, e) + } +} + +func TestSerializationRetriesWholeTransactionOnce(t *testing.T) { + pool, ctx := testPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + projectID := seedProject(t, ctx, pool, suffix) + docID := "uow_" + suffix + "_ser" + work := uow.NewPG(pool) + work.Sleep = func(time.Duration) {} + attempts := 0 + err := work.Within(ctx, func(scope uow.TxScope) error { + attempts++ + body := fmt.Sprintf("# ser %d\n", attempts) + if err := writeOwners(ctx, scope, projectID, docID, body); err != nil { + return err + } + if attempts == 1 { + return &pgconn.PgError{Code: "40001", Message: "could not serialize access"} + } + return nil + }) + if err != nil { + t.Fatal(err) + } + if attempts != 2 { + t.Fatalf("attempts=%d want 2", attempts) + } + r, s, e := countDoc(t, ctx, pool, docID) + if r != 1 || s != 1 || e != 1 { + t.Fatalf("want one business result records=%d snaps=%d events=%d", r, s, e) + } +} + +func TestNonRetryableErrorsAreNotRetried(t *testing.T) { + pool, ctx := testPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + projectID := seedProject(t, ctx, pool, suffix) + work := uow.NewPG(pool) + work.Sleep = func(time.Duration) {} + for _, want := range []error{uow.ErrVersionConflict, uow.ErrPermissionDenied, uow.ErrDigestConflict} { + attempts := 0 + err := work.Within(ctx, func(scope uow.TxScope) error { + attempts++ + _ = (docgov.Repository{}).InsertRoot(ctx, scope, docgov.DocumentRecord{ + DocID: "uow_" + suffix + "_noretry", ProjectID: projectID, Title: "x", TaskKey: "intake", + }) + return want + }) + if !errors.Is(err, want) { + t.Fatalf("err=%v want %v", err, want) + } + if attempts != 1 { + t.Fatalf("%v retried attempts=%d", want, attempts) + } + } + r, _, _ := countDoc(t, ctx, pool, "uow_"+suffix+"_noretry") + if r != 0 { + t.Fatalf("non-retryable write leaked record=%d", r) + } +} + +func TestHappyPathPersistsBothOwners(t *testing.T) { + pool, ctx := testPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + projectID := seedProject(t, ctx, pool, suffix) + docID := "uow_" + suffix + "_ok" + work := uow.NewPG(pool) + if err := work.Within(ctx, func(scope uow.TxScope) error { + return writeOwners(ctx, scope, projectID, docID, "# ok\n") + }); err != nil { + t.Fatal(err) + } + r, s, e := countDoc(t, ctx, pool, docID) + if r != 1 || s != 1 || e != 1 { + t.Fatalf("happy path records=%d snaps=%d events=%d", r, s, e) + } +} diff --git a/internal/app/uow/uow.go b/internal/app/uow/uow.go new file mode 100644 index 00000000..e77267d6 --- /dev/null +++ b/internal/app/uow/uow.go @@ -0,0 +1,41 @@ +package uow + +import ( + "context" + "errors" + "time" +) + +var ( + ErrVersionConflict = errors.New("uow: version conflict") + ErrPermissionDenied = errors.New("uow: permission denied") + ErrDigestConflict = errors.New("uow: digest conflict") + ErrBusy = errors.New("uow: lock busy") + ErrClosedScope = errors.New("uow: transaction scope is closed") + ErrNestedBegin = errors.New("uow: repository must not begin a nested transaction") + ErrPrematureCommit = errors.New("uow: repository must not commit the unit of work") +) + +// TxScope is an opaque transaction handle. Application code cannot Exec SQL through it. +type TxScope interface { + Context() context.Context +} + +// UnitOfWork runs one command in a single database transaction. +type UnitOfWork interface { + Within(ctx context.Context, fn func(TxScope) error) error +} + +const ( + DefaultLockTimeout = 2 * time.Second + DefaultMaxAttempts = 3 +) + +type RetryPolicy struct { + MaxAttempts int + LockTimeout time.Duration +} + +func DefaultRetryPolicy() RetryPolicy { + return RetryPolicy{MaxAttempts: DefaultMaxAttempts, LockTimeout: DefaultLockTimeout} +} diff --git a/internal/app/uow/uow_test.go b/internal/app/uow/uow_test.go new file mode 100644 index 00000000..a36f26e4 --- /dev/null +++ b/internal/app/uow/uow_test.go @@ -0,0 +1,48 @@ +package uow + +import ( + "context" + "errors" + "reflect" + "testing" + + "github.com/jackc/pgx/v5/pgconn" +) + +func TestTxScopeDoesNotExposeExec(t *testing.T) { + typ := reflect.TypeOf((*TxScope)(nil)).Elem() + for _, name := range []string{"Exec", "Begin", "Commit", "Rollback"} { + if _, ok := typ.MethodByName(name); ok { + t.Fatalf("TxScope must not expose %s", name) + } + } + pg := reflect.TypeOf(&pgScope{}) + if _, ok := pg.MethodByName("Exec"); ok { + t.Fatal("pgScope must not expose Exec") + } +} + +type fakeScope struct{} + +func (fakeScope) Context() context.Context { return context.Background() } + +func TestAsPGRejectsOpaqueScope(t *testing.T) { + if _, err := AsPG(fakeScope{}); err == nil { + t.Fatal("expected unwrap failure") + } +} + +func TestSerializationAndNonRetryable(t *testing.T) { + if !IsSerialization(&pgconn.PgError{Code: "40001"}) || !IsSerialization(&pgconn.PgError{Code: "40P01"}) { + t.Fatal("serialization/deadlock must retry") + } + if IsSerialization(errors.New("plain")) { + t.Fatal("plain error must not retry") + } + if !IsNonRetryable(ErrVersionConflict) || !IsNonRetryable(ErrPermissionDenied) || !IsNonRetryable(ErrDigestConflict) { + t.Fatal("typed conflicts must not enter retry") + } + if IsNonRetryable(&pgconn.PgError{Code: "40001"}) { + t.Fatal("serialization is retryable") + } +} diff --git a/internal/archtest/boundaries.go b/internal/archtest/boundaries.go index cf8d91b5..8f575af3 100644 --- a/internal/archtest/boundaries.go +++ b/internal/archtest/boundaries.go @@ -24,6 +24,7 @@ type Boundaries struct { PathOwners map[string][]string `yaml:"path_owners"` EventProducers map[string][]string `yaml:"event_producers"` EventConsumers map[string][]string `yaml:"event_consumers"` + UnwrapOwners []string `yaml:"unwrap_owners"` Exceptions []Exception `yaml:"exceptions"` BaselineCounts map[string]int `yaml:"baseline_counts"` } @@ -93,6 +94,9 @@ func LoadBoundaries(path string) (*Boundaries, error) { if out.EventConsumers == nil { out.EventConsumers = map[string][]string{} } + if out.UnwrapOwners == nil { + out.UnwrapOwners = []string{} + } if out.BaselineCounts == nil { out.BaselineCounts = map[string]int{} } diff --git a/internal/archtest/scan.go b/internal/archtest/scan.go index 6b89c219..8cdd3e79 100644 --- a/internal/archtest/scan.go +++ b/internal/archtest/scan.go @@ -139,6 +139,7 @@ func scanProductionGo(fset *token.FileSet, repoRoot, path string, b *Boundaries) out = append(out, scanSQLWrites(string(src), pkg, rel, b)...) out = append(out, scanPaths(file, pkg, rel, b)...) out = append(out, scanEvents(file, string(src), pkg, rel, b)...) + out = append(out, scanUnwrap(string(src), pkg, rel, b)...) return out, nil } @@ -409,6 +410,22 @@ func scanEvents(file *ast.File, src, pkg, rel string, b *Boundaries) []Finding { return out } +func scanUnwrap(src, pkg, rel string, b *Boundaries) []Finding { + if !strings.Contains(src, "uow.AsPG") { + return nil + } + if allowedHas(b.UnwrapOwners, pkg) { + return nil + } + return []Finding{{ + Kind: KindImport, + Package: pkg, + File: rel, + Symbol: "uow.AsPG", + Detail: "application layer must not unwrap TxScope", + }} +} + func dedupe(in []Finding) []Finding { seen := map[string]bool{} var out []Finding diff --git a/internal/docgov/repository.go b/internal/docgov/repository.go new file mode 100644 index 00000000..8c1c31da --- /dev/null +++ b/internal/docgov/repository.go @@ -0,0 +1,106 @@ +package docgov + +import ( + "context" + "fmt" + "time" + + "chat-agent-system/internal/app/uow" +) + +// Repository is the document_records writer bound to a UnitOfWork scope. +type Repository struct{} + +func (Repository) InsertRoot(ctx context.Context, scope uow.TxScope, rec DocumentRecord) error { + tx, err := uow.AsPG(scope) + if err != nil { + return err + } + if rec.CreatedAt.IsZero() { + rec.CreatedAt = time.Now().UTC() + } + if rec.UpdatedAt.IsZero() { + rec.UpdatedAt = rec.CreatedAt + } + if rec.Stage == "" { + rec.Stage = "V0.5_feasibility" + } + if rec.CurrentState == "" { + rec.CurrentState = StateDrafting + } + if rec.Title == "" { + rec.Title = rec.DocID + } + if rec.TaskKey == "" { + rec.TaskKey = "intake" + } + _, err = tx.Exec(ctx, ` + INSERT INTO document_records ( + doc_id, project_id, task_key, title, stage, version, + current_state, is_locked, is_template_only, content_sha256, work_dir, + author_name, reviewer_name, approver_name, created_at, updated_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)`, + rec.DocID, rec.ProjectID, rec.TaskKey, rec.Title, rec.Stage, rec.Version, + string(rec.CurrentState), rec.IsLocked, rec.IsTemplateOnly, rec.ContentSHA256, rec.WorkDir, + rec.AuthorName, rec.ReviewerName, rec.ApproverName, rec.CreatedAt, rec.UpdatedAt) + if err != nil { + return fmt.Errorf("docgov: insert root: %w", err) + } + return nil +} + +func (Repository) GetForUpdate(ctx context.Context, scope uow.TxScope, docID string) (*DocumentRecord, error) { + tx, err := uow.AsPG(scope) + if err != nil { + return nil, err + } + var rec DocumentRecord + var state string + err = tx.QueryRow(ctx, ` + SELECT doc_id, project_id, task_key, title, stage, version, current_state, + is_locked, is_template_only, content_sha256, work_dir + FROM document_records WHERE doc_id=$1 FOR UPDATE`, docID).Scan( + &rec.DocID, &rec.ProjectID, &rec.TaskKey, &rec.Title, &rec.Stage, &rec.Version, &state, + &rec.IsLocked, &rec.IsTemplateOnly, &rec.ContentSHA256, &rec.WorkDir) + if err != nil { + return nil, fmt.Errorf("docgov: lock document: %w", err) + } + rec.CurrentState = DocumentState(state) + return &rec, nil +} + +func (Repository) UpdateState(ctx context.Context, scope uow.TxScope, rec DocumentRecord) error { + tx, err := uow.AsPG(scope) + if err != nil { + return err + } + _, err = tx.Exec(ctx, ` + UPDATE document_records + SET current_state=$2, is_locked=$3, content_sha256=COALESCE(NULLIF($4,''), content_sha256), + updated_at=$5 + WHERE doc_id=$1`, + rec.DocID, string(rec.CurrentState), rec.IsLocked, rec.ContentSHA256, time.Now().UTC()) + if err != nil { + return fmt.Errorf("docgov: update state: %w", err) + } + return nil +} + +func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID string, versionID, snapshotID int64, revisionNo int, sha string) error { + tx, err := uow.AsPG(scope) + if err != nil { + return err + } + _, err = tx.Exec(ctx, ` + UPDATE document_records + SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, + row_revision=COALESCE(row_revision,0)+1, + content_sha256=COALESCE(NULLIF($5,''), content_sha256), + updated_at=now() + WHERE doc_id=$1`, + docID, versionID, snapshotID, revisionNo, sha) + if err != nil { + return fmt.Errorf("docgov: project head: %w", err) + } + return nil +} diff --git a/internal/docrevision/repository.go b/internal/docrevision/repository.go new file mode 100644 index 00000000..fc8bf45d --- /dev/null +++ b/internal/docrevision/repository.go @@ -0,0 +1,52 @@ +package docrevision + +import ( + "context" + "fmt" + "time" + + "chat-agent-system/internal/app/uow" +) + +// Repository is the ledger writer bound to a UnitOfWork scope. +type Repository struct{} + +func (Repository) InsertBlob(ctx context.Context, scope uow.TxScope, sha, markdown string) error { + tx, err := uow.AsPG(scope) + if err != nil { + return err + } + _, 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`, + sha, markdown, int64(len(markdown)), time.Now().UTC()) + if err != nil { + return fmt.Errorf("docrevision: insert blob: %w", err) + } + return nil +} + +func (Repository) InsertSnapshot(ctx context.Context, scope uow.TxScope, sn Snapshot) (int64, error) { + tx, err := uow.AsPG(scope) + if err != nil { + return 0, err + } + return (&Service{}).insertSnapshotTX(ctx, tx, &sn) +} + +func (Repository) InsertVersion(ctx context.Context, scope uow.TxScope, v Version) (int64, error) { + tx, err := uow.AsPG(scope) + if err != nil { + return 0, err + } + return (&Service{}).insertVersionTX(ctx, tx, &v) +} + +func (Repository) InsertEvent(ctx context.Context, scope uow.TxScope, ev ChangeEvent) (int64, error) { + tx, err := uow.AsPG(scope) + if err != nil { + return 0, err + } + return (&Service{}).insertEventTX(ctx, tx, &ev) +} diff --git a/internal/freedoc/repository.go b/internal/freedoc/repository.go new file mode 100644 index 00000000..96a3053f --- /dev/null +++ b/internal/freedoc/repository.go @@ -0,0 +1,18 @@ +package freedoc + +import ( + "context" + "fmt" + + "chat-agent-system/internal/app/uow" +) + +// Repository is a M3 stub. M1 freeform projection stays on the existing service path. +type Repository struct{} + +func (Repository) ForbiddenInM1(ctx context.Context, scope uow.TxScope) error { + if _, err := uow.AsPG(scope); err != nil { + return err + } + return fmt.Errorf("freedoc: repository writes are not enabled until M3") +} diff --git a/internal/taskrun/repository.go b/internal/taskrun/repository.go new file mode 100644 index 00000000..6bcf13df --- /dev/null +++ b/internal/taskrun/repository.go @@ -0,0 +1,18 @@ +package taskrun + +import ( + "context" + "fmt" + + "chat-agent-system/internal/app/uow" +) + +// Repository is a M3 stub. M1 must not write task_runs through this port. +type Repository struct{} + +func (Repository) ForbiddenInM1(ctx context.Context, scope uow.TxScope) error { + if _, err := uow.AsPG(scope); err != nil { + return err + } + return fmt.Errorf("taskrun: repository writes are not enabled until M3") +}