diff --git a/internal/app/app_test.go b/internal/app/app_test.go index e3c281f6..e489e6df 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -42,3 +42,14 @@ func TestStartStopRestartIdempotent(t *testing.T) { _ = rt.Stop(ctx) time.Sleep(10 * time.Millisecond) } + +func TestWriterLiveNilRuntime(t *testing.T) { + var rt *Runtime + if rt.WriterLive(context.Background(), "notify") { + t.Fatal("nil runtime must not skip hooks") + } + rt = &Runtime{} + if rt.WriterLive(context.Background(), "notify") { + t.Fatal("unwired runtime must keep hook writers") + } +} diff --git a/internal/app/docnotify/subscriber_test.go b/internal/app/docnotify/subscriber_test.go index 16a6440f..8b568648 100644 --- a/internal/app/docnotify/subscriber_test.go +++ b/internal/app/docnotify/subscriber_test.go @@ -61,6 +61,44 @@ func TestHandleCandidateReady(t *testing.T) { } } +func TestHandleSkipsWhenConsumerLive(t *testing.T) { + n := &stubNotify{available: true} + sub := &Subscriber{ + Notify: n, + Docs: stubDocs{doc: &docgov.DocumentRecord{AuthorID: 9, ProjectID: 4}}, + SkipWrite: func(context.Context) bool { + return true + }, + } + if err := sub.Handle(context.Background(), notification.EventDocAgentCandidateReady, map[string]any{ + "doc_id": "d1", "candidate_id": int64(3), + }); err != nil { + t.Fatal(err) + } + if n.calls != 0 { + t.Fatalf("live consumer must skip hook write, calls=%d", n.calls) + } +} + +func TestHandleWritesWhenNotLive(t *testing.T) { + n := &stubNotify{available: true} + sub := &Subscriber{ + Notify: n, + Docs: stubDocs{doc: &docgov.DocumentRecord{AuthorID: 9, ProjectID: 4}}, + SkipWrite: func(context.Context) bool { + return false + }, + } + if err := sub.Handle(context.Background(), notification.EventDocAgentCandidateReady, map[string]any{ + "doc_id": "d1", "candidate_id": int64(3), + }); err != nil { + t.Fatal(err) + } + if n.calls != 1 { + t.Fatalf("off/shadow must keep hook write, calls=%d", n.calls) + } +} + func TestHandleSkipsUnknownAndNoRecipient(t *testing.T) { n := &stubNotify{available: true} sub := &Subscriber{Notify: n} diff --git a/internal/events/dispatcher_pg_test.go b/internal/events/dispatcher_pg_test.go index 75c923e3..4ed35cf1 100644 --- a/internal/events/dispatcher_pg_test.go +++ b/internal/events/dispatcher_pg_test.go @@ -295,3 +295,32 @@ func TestE0B_ShadowDoesNotAck_PG(t *testing.T) { t.Fatal(err) } } + +func TestE2_WriterLiveModes_PG(t *testing.T) { + pool, ctx := e0bPool(t) + env := fmt.Sprintf("e2-%d", time.Now().UnixNano()) + if events.WriterLive(ctx, pool, env, events.ConsumerNotify) { + t.Fatal("missing row must not be live") + } + if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeOff); err != nil { + t.Fatal(err) + } + if events.WriterLive(ctx, pool, env, events.ConsumerNotify) { + t.Fatal("off must keep hook writers") + } + if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeShadow); err != nil { + t.Fatal(err) + } + if events.WriterLive(ctx, pool, env, events.ConsumerNotify) { + t.Fatal("shadow must keep hook writers") + } + if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeLive); err != nil { + t.Fatal(err) + } + if !events.WriterLive(ctx, pool, env, events.ConsumerNotify) { + t.Fatal("live must stop hook writers") + } + if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeOff); err != nil { + t.Fatal(err) + } +} diff --git a/internal/events/dispatcher_test.go b/internal/events/dispatcher_test.go index 7b7a1a0e..17e9cbea 100644 --- a/internal/events/dispatcher_test.go +++ b/internal/events/dispatcher_test.go @@ -183,6 +183,12 @@ func TestAuthAndIntegrityGoDeadLetter(t *testing.T) { } } +func TestWriterLiveNilPool(t *testing.T) { + if WriterLive(context.Background(), nil, "dev", ConsumerNotify) { + t.Fatal("nil pool must keep in-process writers") + } +} + func TestReplayRequiresScope(t *testing.T) { if err := AuthorizeReplay(ReplayRequest{}); err != ErrReplayScope { t.Fatalf("empty: %v", err)