From 92f31ffa6af0bb75f844637bc60ee7d3f39d5d48 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 09:59:55 +0000 Subject: [PATCH] feat(events): apply version-committed projections only forward Co-authored-by: Cursor --- .doc/system-design/module-boundaries.yaml | 7 ++ cmd/events/main.go | 4 + internal/app/docproject/consumer.go | 96 +++++++++++++++++++++++ internal/docgov/repository.go | 22 ++++++ internal/events/registry.go | 28 +++---- 5 files changed, 143 insertions(+), 14 deletions(-) create mode 100644 internal/app/docproject/consumer.go diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 06031c3c..d4fb184c 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -62,6 +62,7 @@ imports: - app/docadvance - app/docflow - app/docnotify + - app/docproject - app/stageflow - app/uow - artifactvault @@ -96,6 +97,9 @@ imports: - notification - projectinstance - projectworkflow + app/docproject: + - docgov + - events app/docflow: - app/uow - docgov @@ -255,6 +259,8 @@ table_owners: - events event_reconciliations: - events + event_projection_outcomes: + - app/docproject notifications: - notification stage_advance_requests: @@ -308,6 +314,7 @@ event_producers: - docrevision - app/docflow - events + - app/docproject document.downstream_impact_detected: - docrevision - app/docflow diff --git a/cmd/events/main.go b/cmd/events/main.go index da67da6c..3102625c 100644 --- a/cmd/events/main.go +++ b/cmd/events/main.go @@ -13,6 +13,7 @@ import ( "chat-agent-system/internal/app/docadvance" "chat-agent-system/internal/app/docnotify" + "chat-agent-system/internal/app/docproject" "chat-agent-system/internal/config" "chat-agent-system/internal/db" "chat-agent-system/internal/docgov" @@ -77,6 +78,9 @@ func main() { _ = disp.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{ Inst: instSvc, Docs: gov, PW: pwSvc, Notify: notify, Env: cfg.Env, }).HandleLease) + _ = disp.Register(events.ConsumerProjection, (&docproject.EventConsumer{ + Docs: gov, Env: cfg.Env, + }).HandleLease) mux := http.NewServeMux() mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "env": cfg.Env, "owner": owner}) diff --git a/internal/app/docproject/consumer.go b/internal/app/docproject/consumer.go new file mode 100644 index 00000000..a9766814 --- /dev/null +++ b/internal/app/docproject/consumer.go @@ -0,0 +1,96 @@ +package docproject + +import ( + "context" + "encoding/json" + "errors" + + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/events" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +// EventConsumer applies document.agent_version_committed onto ledger head pointers. +// Closures/producer already project in the approve/commit UoW; this path is idempotent no-replace. +type EventConsumer struct { + Docs DocumentLookup + Env string +} + +type DocumentLookup interface { + GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error) +} + +func (c *EventConsumer) HandleLease(ctx context.Context, lease events.Lease, tx pgx.Tx) error { + if c == nil { + return events.ErrUnknownConsumer + } + principal := events.ServicePrincipal{ + ID: "projection-" + c.Env, Environment: c.Env, Consumer: events.ConsumerProjection, + Actions: []string{"projection.apply"}, + } + if err := events.AuthorizePrincipal(principal, lease.Environment, lease.ProjectID, "projection.apply"); err != nil { + return events.ErrQuarantine + } + if lease.EventName != events.EventVersionCommitted { + return nil + } + docID := lease.Envelope.AggregateID + versionNo := int(lease.Envelope.AggregateVersion) + var payload map[string]any + _ = json.Unmarshal(lease.Envelope.Payload, &payload) + if payload != nil { + if id, ok := payload["doc_id"].(string); ok && id != "" { + docID = id + } + if v, ok := payload["version_no"].(float64); ok { + versionNo = int(v) + } + } + if c.Docs == nil || docID == "" { + return events.ErrQuarantine + } + doc, err := c.Docs.GetDocument(ctx, docID) + if err != nil || doc == nil { + return events.ErrQuarantine + } + if doc.ProjectID != lease.ProjectID { + return events.ErrQuarantine + } + var versionID, snapshotID int64 + var sha string + err = tx.QueryRow(ctx, ` + SELECT v.id, COALESCE(v.head_snapshot_id,0), COALESCE(s.content_sha256,'') + FROM document_versions v + LEFT JOIN document_snapshots s ON s.id = v.head_snapshot_id + WHERE v.doc_id=$1 AND v.version_no=$2 + `, docID, versionNo).Scan(&versionID, &snapshotID, &sha) + if err != nil { + return events.ErrQuarantine + } + tag, err := tx.Exec(ctx, ` + INSERT INTO event_projection_outcomes (source_event_id, doc_id, version_no, snapshot_id, applied) + VALUES ($1,$2,$3,$4,false) + ON CONFLICT (source_event_id) DO NOTHING + `, lease.EventID, docID, versionNo, snapshotID) + if err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "23505" { + return nil + } + return err + } + if tag.RowsAffected() == 0 { + return nil + } + applied, err := (docgov.Repository{}).ProjectHeadNoReplace(ctx, tx, docID, versionID, snapshotID, versionNo, sha) + if err != nil { + return err + } + _, err = tx.Exec(ctx, ` + UPDATE event_projection_outcomes SET applied=$2, snapshot_id=$3 WHERE source_event_id=$1 + `, lease.EventID, applied, snapshotID) + return err +} diff --git a/internal/docgov/repository.go b/internal/docgov/repository.go index e14ae50a..a6f8c9f0 100644 --- a/internal/docgov/repository.go +++ b/internal/docgov/repository.go @@ -6,6 +6,8 @@ import ( "time" "chat-agent-system/internal/app/uow" + + "github.com/jackc/pgx/v5" ) // Repository is the document_records writer bound to a UnitOfWork scope. @@ -106,3 +108,23 @@ func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID stri } return nil } + +// ProjectHeadNoReplace moves ledger pointers only forward. Equal or newer heads stay. +func (Repository) ProjectHeadNoReplace(ctx context.Context, tx pgx.Tx, docID string, versionID, snapshotID int64, revisionNo int, sha string) (bool, error) { + if tx == nil || docID == "" { + return false, fmt.Errorf("docgov: project head requires tx and doc_id") + } + tag, 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 + AND COALESCE(head_revision_no, 0) < $4 + `, docID, versionID, snapshotID, revisionNo, sha) + if err != nil { + return false, fmt.Errorf("docgov: project head no-replace: %w", err) + } + return tag.RowsAffected() == 1, nil +} diff --git a/internal/events/registry.go b/internal/events/registry.go index ba25ea57..9028f1d5 100644 --- a/internal/events/registry.go +++ b/internal/events/registry.go @@ -12,24 +12,24 @@ const ( ConsumerWorkflowAdvance = "workflow_advance" ConsumerFAV = "fav" - EventCandidateReady = "document.agent_candidate_ready" - EventVersionCommitted = "document.agent_version_committed" - EventDownstreamImpact = "document.downstream_impact_detected" + EventCandidateReady = "document.agent_candidate_ready" + EventVersionCommitted = "document.agent_version_committed" + EventDownstreamImpact = "document.downstream_impact_detected" EventChangeReportDegraded = "document.change_report_degraded" - EventApproved = "document.approved" - EventTaskRunCompleted = "taskrun.completed" - EventTaskRunCancelled = "taskrun.cancelled" - EventFAVArtifactStaged = "fav.artifact_staged" + EventApproved = "document.approved" + EventTaskRunCompleted = "taskrun.completed" + EventTaskRunCancelled = "taskrun.cancelled" + EventFAVArtifactStaged = "fav.artifact_staged" ) // Contract is the static allowlist row for one event name. type Contract struct { - EventName string - SchemaVersion int - Enabled bool - Producers []string - Consumers []string - PayloadFields []string + EventName string + SchemaVersion int + Enabled bool + Producers []string + Consumers []string + PayloadFields []string } // StaticContracts is the runtime copy of module-boundaries event sections. @@ -37,7 +37,7 @@ type Contract struct { func StaticContracts() []Contract { return []Contract{ {EventName: EventCandidateReady, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "candidate_id", "risk"}}, - {EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"doc_id", "version_no"}}, + {EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events", "app/docproject"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"doc_id", "version_no"}}, {EventName: EventDownstreamImpact, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "impacted_doc_id", "version_no"}}, {EventName: EventChangeReportDegraded, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "report_id"}}, {EventName: EventApproved, SchemaVersion: 1, Enabled: true, Producers: []string{"app/docflow", "events", "app/docadvance"}, Consumers: []string{ConsumerWorkflowAdvance, ConsumerNotify}, PayloadFields: []string{"doc_id", "project_id", "task_key"}},