test(docproject): prove projection shadow, uniqueness, and no-replace
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
@@ -0,0 +1,208 @@
|
||||
package docproject_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"chat-agent-system/internal/app/docproject"
|
||||
"chat-agent-system/internal/app/uow"
|
||||
"chat-agent-system/internal/db"
|
||||
"chat-agent-system/internal/docgov"
|
||||
"chat-agent-system/internal/events"
|
||||
|
||||
"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 TestE1C_ProjectionShadowLiveNoReplace_PG(t *testing.T) {
|
||||
pool, ctx := e1cPool(t)
|
||||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
userID, projectID := seedUserProject(t, ctx, pool, suffix)
|
||||
docID := "e1c_" + suffix
|
||||
seedDoc(t, ctx, pool, docID, projectID, userID)
|
||||
snapID, verID := seedVersion(t, ctx, pool, docID, 2)
|
||||
|
||||
env := committedEnv("c1-"+suffix, docID, projectID, 2)
|
||||
appendEnv(t, ctx, pool, env, []string{events.ConsumerProjection})
|
||||
|
||||
d := events.NewDispatcher(pool, "dev", "e1c-w")
|
||||
if err := d.Register(events.ConsumerProjection, (&docproject.EventConsumer{
|
||||
Docs: docgov.NewGovernanceEngine(pool), Env: "dev",
|
||||
}).HandleLease); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerProjection, events.ModeShadow); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil || n < 1 {
|
||||
t.Fatalf("shadow n=%d err=%v", n, err)
|
||||
}
|
||||
var outcomes, head int
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_projection_outcomes WHERE source_event_id=$1`, env.EventID).Scan(&outcomes); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if outcomes != 0 {
|
||||
t.Fatalf("shadow wrote outcomes=%d", outcomes)
|
||||
}
|
||||
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerProjection, events.ModeLive); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_projection_outcomes WHERE doc_id=$1 AND version_no=2`, docID).Scan(&outcomes); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if outcomes != 1 {
|
||||
t.Fatalf("unique projection outcomes=%d", outcomes)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT head_revision_no FROM document_records WHERE doc_id=$1`, docID).Scan(&head); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if head != 2 {
|
||||
t.Fatalf("head_revision_no=%d want 2 snap=%d ver=%d", head, snapID, verID)
|
||||
}
|
||||
|
||||
older := committedEnv("old-"+suffix, docID, projectID, 1)
|
||||
appendEnv(t, ctx, pool, older, []string{events.ConsumerProjection})
|
||||
seedVersion(t, ctx, pool, docID, 1)
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT head_revision_no FROM document_records WHERE doc_id=$1`, docID).Scan(&head); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if head != 2 {
|
||||
t.Fatalf("no-replace failed, head=%d", head)
|
||||
}
|
||||
|
||||
stale := committedEnv("stale-"+suffix, "missing-"+suffix, projectID, 1)
|
||||
appendEnv(t, ctx, pool, stale, []string{events.ConsumerProjection})
|
||||
if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var status string
|
||||
if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, stale.EventID, events.ConsumerProjection).Scan(&status); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if status != "quarantine" {
|
||||
t.Fatalf("missing doc status=%s", status)
|
||||
}
|
||||
|
||||
tx, err := pool.Begin(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
replay := committedEnv("replay-"+suffix, docID, projectID, 3)
|
||||
appendEnv(t, ctx, pool, replay, []string{events.ConsumerProjection})
|
||||
seedVersion(t, ctx, pool, docID, 3)
|
||||
if err := events.SetMode(ctx, pool, "dev", events.ConsumerProjection, events.ModeOff); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
n, err := d.RunOnce(ctx, events.ConsumerProjection)
|
||||
_ = tx.Rollback(ctx)
|
||||
if err != nil || n != 0 {
|
||||
t.Fatalf("rollback off claimed=%d err=%v", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
func e1cPool(t *testing.T) (*pgxpool.Pool, context.Context) {
|
||||
t.Helper()
|
||||
if testing.Short() {
|
||||
t.Skip("e1c")
|
||||
}
|
||||
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 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`, "e1c-"+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`, "e1c-"+suffix, "e1c").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, head_revision_no
|
||||
) VALUES ($1,$2,'intake','t','V0.5',1,'drafting',false,false,$3,'A',now(),now(),0)
|
||||
`, docID, projectID, userID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func seedVersion(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, versionNo int) (snapID, verID int64) {
|
||||
t.Helper()
|
||||
sha := fmt.Sprintf("%064d", versionNo)
|
||||
if _, err := pool.Exec(ctx, `INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size) VALUES ($1,$2,1) ON CONFLICT (content_sha256) DO NOTHING`, sha, "body"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `
|
||||
INSERT INTO document_snapshots (doc_id, content_sha256, block_map, source_kind)
|
||||
VALUES ($1,$2,'[]','agent') RETURNING id
|
||||
`, docID, sha).Scan(&snapID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `
|
||||
INSERT INTO document_versions (doc_id, version_no, status, origin_kind, baseline_snapshot_id, head_snapshot_id, sealed_at)
|
||||
VALUES ($1,$2,'sealed','agent',$3,$3,now()) RETURNING id
|
||||
`, docID, versionNo, snapID).Scan(&verID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return snapID, verID
|
||||
}
|
||||
|
||||
func committedEnv(id, docID string, projectID, ver int64) events.Envelope {
|
||||
body := []byte(fmt.Sprintf(`{"doc_id":%q,"version_no":%d}`, docID, ver))
|
||||
return events.Envelope{
|
||||
EventID: "commit-" + id, EventName: events.EventVersionCommitted, SchemaVersion: events.EnvelopeSchema,
|
||||
Environment: "dev", ProjectID: projectID, AggregateType: "document", AggregateID: docID,
|
||||
AggregateVersion: ver, ActorRef: "user:1", IdempotencyKey: "commit-" + 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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user