From a5e46eeb5fcd9c49d22e64767aca6399822f703e Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 09:26:36 +0000 Subject: [PATCH] test(docnotify): prove notify shadow, live uniqueness, and rollback on isolated PG Co-authored-by: Cursor --- internal/app/docnotify/e1a_pg_test.go | 349 ++++++++++++++++++++++++++ 1 file changed, 349 insertions(+) create mode 100644 internal/app/docnotify/e1a_pg_test.go diff --git a/internal/app/docnotify/e1a_pg_test.go b/internal/app/docnotify/e1a_pg_test.go new file mode 100644 index 00000000..7a0d4536 --- /dev/null +++ b/internal/app/docnotify/e1a_pg_test.go @@ -0,0 +1,349 @@ +package docnotify_test + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "chat-agent-system/internal/app/docnotify" + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/db" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/events" + "chat-agent-system/internal/notification" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func init() { + 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 TestIntendedNotifyDigestStable(t *testing.T) { + a := docnotify.IntendedNotifyDigest("e1", notification.TypeDocApproved, 9) + b := docnotify.IntendedNotifyDigest("e1", notification.TypeDocApproved, 9) + if a == "" || a != b { + t.Fatalf("%s %s", a, b) + } + want := events.VisibleOutcomeDigest("e1", notification.TypeDocApproved, 9) + if a != want { + t.Fatalf("normalizer drift: %s vs %s", a, want) + } +} + +func TestE1A_NotifyShadowLiveRollback_PG(t *testing.T) { + pool, ctx := e1aPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + userID, projectID := seedUserProject(t, ctx, pool, suffix) + docID := "e1a_" + suffix + seedDoc(t, ctx, pool, docID, projectID, userID) + + env := approvedEnv(suffix, docID, projectID, 1) + appendEnv(t, ctx, pool, env, []string{events.ConsumerNotify, events.ConsumerWorkflowAdvance}) + + d := newNotifyDispatcher(t, pool, "e1a-w") + if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil { + t.Fatal(err) + } + _, revShadow, err := events.LoadMode(ctx, pool, "dev", events.ConsumerNotify) + if err != nil || revShadow < 1 { + t.Fatalf("shadow revision=%d err=%v", revShadow, err) + } + if n, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil || n < 1 { + t.Fatalf("shadow n=%d err=%v", n, err) + } + var notes, recon int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, env.EventID).Scan(¬es); err != nil { + t.Fatal(err) + } + if notes != 0 { + t.Fatalf("shadow wrote notifications=%d", notes) + } + var intended string + if err := pool.QueryRow(ctx, `SELECT COUNT(*), MAX(intended_digest) FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, env.EventID).Scan(&recon, &intended); err != nil { + t.Fatal(err) + } + if recon < 1 { + t.Fatal("missing shadow recon") + } + want := docnotify.IntendedNotifyDigest(env.EventID, notification.TypeDocApproved, userID) + if intended != want { + t.Fatalf("intended digest %s want %s", intended, want) + } + + if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil { + t.Fatal(err) + } + _, revLive, err := events.LoadMode(ctx, pool, "dev", events.ConsumerNotify) + if err != nil || revLive <= revShadow { + t.Fatalf("live revision %d must exceed shadow %d", revLive, revShadow) + } + if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil { + t.Fatal(err) + } + if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1 AND user_id=$2 AND type=$3`, + env.EventID, userID, notification.TypeDocApproved).Scan(¬es); err != nil { + t.Fatal(err) + } + if notes != 1 { + t.Fatalf("live visible notifications=%d", notes) + } + liveHash, err := hashCommitted(ctx, pool, env.EventID) + if err != nil { + t.Fatal(err) + } + if liveHash != intended { + t.Fatalf("live hash %s != shadow intended %s", liveHash, intended) + } + + var attempts, receipts int + if err := pool.QueryRow(ctx, ` + SELECT + (SELECT COUNT(*) FROM event_delivery_attempts WHERE event_id=$1 AND consumer=$2), + (SELECT COUNT(*) FROM notifications WHERE source_event_id=$1) + `, env.EventID, events.ConsumerNotify).Scan(&attempts, &receipts); err != nil { + t.Fatal(err) + } + if attempts < 1 || receipts != 1 { + t.Fatalf("correlation attempts=%d receipts=%d", attempts, receipts) + } + + if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil { + t.Fatal(err) + } + n, err := d.RunOnce(ctx, events.ConsumerNotify) + if err != nil || n != 0 { + t.Fatalf("rollback off claimed=%d err=%v", n, err) + } +} + +func TestE1A_NotifyFixtureThirtyAuthzAndRollbackTx_PG(t *testing.T) { + pool, ctx := e1aPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + userID, projectID := seedUserProject(t, ctx, pool, suffix) + d := newNotifyDispatcher(t, pool, "e1a-n30") + d.Batch = 64 + + const nFix = 30 + ids := make([]string, 0, nFix) + for i := 0; i < nFix; i++ { + docID := fmt.Sprintf("e1a30_%s_%d", suffix, i) + seedDoc(t, ctx, pool, docID, projectID, userID) + env := approvedEnv(fmt.Sprintf("%s-%d", suffix, i), docID, projectID, int64(i+1)) + appendEnv(t, ctx, pool, env, []string{events.ConsumerNotify}) + ids = append(ids, env.EventID) + } + + if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil { + t.Fatal(err) + } + var scanned int + for i := 0; i < 8; i++ { + n, err := d.RunOnce(ctx, events.ConsumerNotify) + if err != nil { + t.Fatalf("shadow err=%v", err) + } + scanned += n + var reconNow int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE mode='shadow' AND event_id LIKE $1`, "%"+suffix+"%").Scan(&reconNow); err != nil { + t.Fatal(err) + } + if reconNow >= nFix { + break + } + if n == 0 { + t.Fatalf("shadow stalled at recon=%d scanned=%d", reconNow, scanned) + } + } + var shadowNotes int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1`, "%"+suffix+"%").Scan(&shadowNotes); err != nil { + t.Fatal(err) + } + if shadowNotes != 0 { + t.Fatalf("fixture shadow wrote %d notifications", shadowNotes) + } + var recon int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE mode='shadow' AND event_id LIKE $1`, "%"+suffix+"%").Scan(&recon); err != nil { + t.Fatal(err) + } + if recon < nFix { + t.Fatalf("fixture shadow recon=%d want >=%d", recon, nFix) + } + for _, id := range ids { + want := docnotify.IntendedNotifyDigest(id, notification.TypeDocApproved, userID) + var got string + if err := pool.QueryRow(ctx, `SELECT intended_digest FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, id).Scan(&got); err != nil { + t.Fatal(err) + } + if got != want { + t.Fatalf("event %s intended %s want %s", id, got, want) + } + } + + d.Batch = 64 + if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil { + t.Fatal(err) + } + for i := 0; i < 8; i++ { + n, err := d.RunOnce(ctx, events.ConsumerNotify) + if err != nil { + t.Fatal(err) + } + var liveNow int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1 AND type=$2`, "%"+suffix+"%", notification.TypeDocApproved).Scan(&liveNow); err != nil { + t.Fatal(err) + } + if liveNow >= nFix { + break + } + if n == 0 && i > 0 { + t.Fatalf("live stalled at notifications=%d", liveNow) + } + } + var liveNotes int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1 AND type=$2`, "%"+suffix+"%", notification.TypeDocApproved).Scan(&liveNotes); err != nil { + t.Fatal(err) + } + if liveNotes != nFix { + t.Fatalf("live unique outcomes=%d want %d", liveNotes, nFix) + } + + // same-tx rollback: handle+ack then rollback leaves no extra row + missingDoc := approvedEnv("missing-"+suffix, "no-such-doc-"+suffix, projectID, 99) + appendEnv(t, ctx, pool, missingDoc, []string{events.ConsumerNotify}) + if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil { + t.Fatal(err) + } + var qStatus string + if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, missingDoc.EventID, events.ConsumerNotify).Scan(&qStatus); err != nil { + t.Fatal(err) + } + if qStatus != "quarantine" { + t.Fatalf("missing doc status=%s want quarantine", qStatus) + } + var qNotes int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, missingDoc.EventID).Scan(&qNotes); err != nil { + t.Fatal(err) + } + if qNotes != 0 { + t.Fatalf("quarantine leaked notifications=%d", qNotes) + } + + tx, err := pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + if err := (&docnotify.EventConsumer{ + Notify: notification.NewService(pool), Docs: docgov.NewGovernanceEngine(pool), Env: "dev", + }).HandleLease(ctx, events.Lease{ + EventID: ids[0], EventName: events.EventApproved, ProjectID: projectID, Environment: "dev", + Envelope: events.Envelope{EventID: ids[0], EventName: events.EventApproved, ProjectID: projectID, AggregateID: "e1a30_" + suffix + "_0", Payload: []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d}`, "e1a30_"+suffix+"_0", projectID))}, + }, tx); err != nil { + t.Fatal(err) + } + _ = tx.Rollback(ctx) + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, ids[0]).Scan(&liveNotes); err != nil { + t.Fatal(err) + } + if liveNotes != 1 { + t.Fatalf("rollback of extra handle changed committed count=%d", liveNotes) + } + + if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil { + t.Fatal(err) + } +} + +func e1aPool(t *testing.T) (*pgxpool.Pool, context.Context) { + t.Helper() + if testing.Short() { + t.Skip("e1a") + } + dsn := db.SkipIfNoTestDatabaseURL(t) + db.RefuseSharedEnvDSN(t, dsn) + 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 newNotifyDispatcher(t *testing.T, pool *pgxpool.Pool, owner string) *events.Dispatcher { + t.Helper() + d := events.NewDispatcher(pool, "dev", owner) + if err := d.Register(events.ConsumerNotify, (&docnotify.EventConsumer{ + Notify: notification.NewService(pool), Docs: docgov.NewGovernanceEngine(pool), Env: "dev", + }).HandleLease); err != nil { + t.Fatal(err) + } + return d +} + +func seedUserProject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix string) (userID, projectID int64) { + t.Helper() + if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'A','engineer') RETURNING id`, "e1a-"+suffix).Scan(&userID); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `INSERT INTO projects (code, name, status) VALUES ($1,$2,'active') RETURNING id`, "e1a-"+suffix, "e1a").Scan(&projectID); err != nil { + t.Fatal(err) + } + return userID, projectID +} + +func seedDoc(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, projectID, userID int64) { + t.Helper() + if _, err := pool.Exec(ctx, ` + INSERT INTO document_records ( + doc_id, project_id, task_key, title, stage, version, current_state, is_locked, is_template_only, + author_id, author_name, created_at, updated_at + ) VALUES ($1,$2,'intake','t','V0.5',1,'approved',true,false,$3,'A',now(),now()) + `, docID, projectID, userID); err != nil { + t.Fatal(err) + } +} + +func approvedEnv(id, docID string, projectID, ver int64) events.Envelope { + body := []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d,"task_key":"intake"}`, docID, projectID)) + return events.Envelope{ + EventID: "approved-" + id, EventName: events.EventApproved, SchemaVersion: events.EnvelopeSchema, + Environment: "dev", ProjectID: projectID, AggregateType: "document_approval", AggregateID: docID, + AggregateVersion: ver, ActorRef: "user:3", IdempotencyKey: "approved-" + id, + PayloadDigest: events.Digest(body), Producer: "docflow", OccurredAt: time.Now().UTC(), Payload: body, + } +} + +func appendEnv(t *testing.T, ctx context.Context, pool *pgxpool.Pool, env events.Envelope, consumers []string) { + t.Helper() + if err := uow.NewPG(pool).Within(ctx, func(scope uow.TxScope) error { + return (events.PGSink{}).Append(uow.WithScope(ctx, scope), env, consumers) + }); err != nil { + t.Fatal(err) + } +} + +func hashCommitted(ctx context.Context, pool *pgxpool.Pool, eventID string) (string, error) { + tx, err := pool.Begin(ctx) + if err != nil { + return "", err + } + defer func() { _ = tx.Rollback(ctx) }() + return events.HashVisibleNotifications(ctx, tx, eventID) +}