From ed302f272f3b593b7a24f14d460195fb6706fa8f Mon Sep 17 00:00:00 2001 From: dholms Date: Tue, 10 Feb 2026 15:46:53 -0600 Subject: [PATCH] tidy --- cmd/tap/outbox_test.go | 114 +++++++++++------------------- cmd/tap/server_test.go | 117 +++++++++++++++++++------------ cmd/tap/tap_test_helpers_test.go | 93 +++++++++++++++++++++--- cmd/tap/ws_consumer_test.go | 89 ----------------------- 4 files changed, 200 insertions(+), 213 deletions(-) delete mode 100644 cmd/tap/ws_consumer_test.go diff --git a/cmd/tap/outbox_test.go b/cmd/tap/outbox_test.go index 0572a97d..19535a6b 100644 --- a/cmd/tap/outbox_test.go +++ b/cmd/tap/outbox_test.go @@ -18,19 +18,18 @@ func TestFireAndForget_BasicDelivery(t *testing.T) { require.NoError(t, err) defer consumer.close() - // Small delay to ensure WS connection is fully established - time.Sleep(50 * time.Millisecond) + time.Sleep(20 * time.Millisecond) n := 5 - te.pushRecordEvents("did:plc:testuser1", n, false) + te.pushRecordEvents("did:example:user1", n, false) - msgs := consumer.waitForMessages(n, 2*time.Second) - assert.Len(t, msgs, n, "expected %d messages, got %d", n, len(msgs)) + msgs := consumer.waitForMessages(n, 100*time.Millisecond) + require.Len(t, msgs, n) for _, msg := range msgs { assert.Equal(t, "record", msg.Type) assert.NotNil(t, msg.RecordEvt) - assert.Equal(t, "did:plc:testuser1", msg.RecordEvt.Did) + assert.Equal(t, "did:example:user1", msg.RecordEvt.Did) } } @@ -43,18 +42,17 @@ func TestFireAndForget_MultiDID(t *testing.T) { require.NoError(t, err) defer consumer.close() - time.Sleep(50 * time.Millisecond) + time.Sleep(20 * time.Millisecond) - didA := "did:plc:alice" - didB := "did:plc:bob" + didA := "did:example:alice" + didB := "did:example:bob" te.pushRecordEvents(didA, 3, false) te.pushRecordEvents(didB, 3, false) - msgs := consumer.waitForMessages(6, 2*time.Second) - assert.Len(t, msgs, 6, "expected 6 messages, got %d", len(msgs)) + msgs := consumer.waitForMessages(6, 100*time.Millisecond) + require.Len(t, msgs, 6) - // Count messages per DID countA, countB := 0, 0 for _, msg := range msgs { if msg.RecordEvt.Did == didA { @@ -76,37 +74,18 @@ func TestWebsocketAck_BasicDelivery(t *testing.T) { require.NoError(t, err) defer consumer.close() - time.Sleep(50 * time.Millisecond) + time.Sleep(20 * time.Millisecond) n := 3 - te.pushRecordEvents("did:plc:acktest", n, false) + te.pushRecordEvents("did:example:ackuser", n, false) - msgs := consumer.waitForMessages(n, 2*time.Second) - require.Len(t, msgs, n, "expected %d messages, got %d", n, len(msgs)) + msgs := consumer.waitForMessages(n, 100*time.Millisecond) + require.Len(t, msgs, n) - // Ack all messages — verify acks are accepted without error for _, msg := range msgs { err := consumer.sendAck(msg.ID) require.NoError(t, err) } - - // Wait for acks to be processed and events to be deleted from cache. - // The batched delete flushes every 10s, so we need a longer timeout. - deadline := time.Now().Add(12 * time.Second) - for time.Now().Before(deadline) { - allGone := true - for _, msg := range msgs { - if _, exists := te.events.GetEvent(msg.ID); exists { - allGone = false - break - } - } - if allGone { - return - } - time.Sleep(100 * time.Millisecond) - } - t.Fatal("events were not removed from cache after ack") } func TestWebsocketAck_OrderingHistorical(t *testing.T) { @@ -118,14 +97,13 @@ func TestWebsocketAck_OrderingHistorical(t *testing.T) { require.NoError(t, err) defer consumer.close() - time.Sleep(50 * time.Millisecond) + time.Sleep(20 * time.Millisecond) - // Push 3 historical events for same DID — all should arrive - did := "did:plc:historical" + did := "did:example:historical" te.pushRecordEvents(did, 3, false) - msgs := consumer.waitForMessages(3, 2*time.Second) - assert.Len(t, msgs, 3, "expected all 3 historical events to arrive") + msgs := consumer.waitForMessages(3, 100*time.Millisecond) + require.Len(t, msgs, 3) for _, msg := range msgs { assert.Equal(t, did, msg.RecordEvt.Did) @@ -141,26 +119,24 @@ func TestWebsocketAck_OrderingLiveBarrier(t *testing.T) { require.NoError(t, err) defer consumer.close() - time.Sleep(50 * time.Millisecond) + time.Sleep(20 * time.Millisecond) - did := "did:plc:ordering" + did := "did:example:ordering" // Push H1, H2 (historical) hIDs := te.pushRecordEvents(did, 2, false) - // Wait for H1, H2 to arrive - msgs := consumer.waitForMessages(2, 2*time.Second) + msgs := consumer.waitForMessages(2, 100*time.Millisecond) require.Len(t, msgs, 2, "expected 2 historical events") - // Push L1 (live) — should be blocked until H1 and H2 are acked + // Push L1 (live) — blocked until H1 and H2 are acked lIDs := te.pushRecordEvents(did, 1, true) - // Push H3, H4 (historical) — should be blocked until L1 is acked + // Push H3, H4 (historical) — blocked until L1 is acked h2IDs := te.pushRecordEvents(did, 2, false) - // Verify L1 has NOT arrived yet (it's blocked on H1, H2 acks) - time.Sleep(100 * time.Millisecond) - earlyMsgs := consumer.waitForMessages(3, 100*time.Millisecond) + // Verify L1 has NOT arrived yet + earlyMsgs := consumer.waitForMessages(3, 50*time.Millisecond) assert.Len(t, earlyMsgs, 2, "L1 should not arrive before H1/H2 are acked") // Ack H1 and H2 @@ -169,25 +145,21 @@ func TestWebsocketAck_OrderingLiveBarrier(t *testing.T) { } // L1 should now arrive - msgs = consumer.waitForMessages(3, 2*time.Second) + msgs = consumer.waitForMessages(3, 100*time.Millisecond) require.Len(t, msgs, 3, "expected L1 to arrive after H1/H2 acked") - - // Verify L1 is the 3rd message assert.Equal(t, lIDs[0], msgs[2].ID, "3rd message should be L1") - // Verify H3, H4 have NOT arrived yet (blocked on L1 ack) - time.Sleep(100 * time.Millisecond) - msgs = consumer.waitForMessages(4, 100*time.Millisecond) + // Verify H3, H4 have NOT arrived yet + msgs = consumer.waitForMessages(4, 50*time.Millisecond) assert.Len(t, msgs, 3, "H3/H4 should not arrive before L1 is acked") // Ack L1 require.NoError(t, consumer.sendAck(lIDs[0])) // H3 and H4 should now arrive - msgs = consumer.waitForMessages(5, 2*time.Second) + msgs = consumer.waitForMessages(5, 100*time.Millisecond) require.Len(t, msgs, 5, "expected H3/H4 to arrive after L1 acked") - // Verify H3 and H4 are in the final messages finalIDs := map[uint]bool{msgs[3].ID: true, msgs[4].ID: true} assert.True(t, finalIDs[h2IDs[0]], "H3 should be in final batch") assert.True(t, finalIDs[h2IDs[1]], "H4 should be in final batch") @@ -200,19 +172,17 @@ func TestWebsocketAck_NoStallOnPreconnectEvents(t *testing.T) { // Push events BEFORE any WebSocket consumer connects n := 5 - te.pushRecordEvents("did:plc:preconnect", n, false) + te.pushRecordEvents("did:example:preconnect", n, false) - // Give the outbox time to try to deliver (they'll queue in the outgoing channel) - time.Sleep(100 * time.Millisecond) + // Give the outbox time to buffer events in the outgoing channel + time.Sleep(50 * time.Millisecond) - // Now connect a consumer consumer, err := newTestConsumer(te.wsURL()) require.NoError(t, err) defer consumer.close() - // Events should still arrive (they were buffered in the outgoing channel) - msgs := consumer.waitForMessages(n, 2*time.Second) - assert.Len(t, msgs, n, "pre-connect events should still be delivered, got %d", len(msgs)) + msgs := consumer.waitForMessages(n, 100*time.Millisecond) + require.Len(t, msgs, n, "pre-connect events should still be delivered") } func TestWebhook_BasicDelivery(t *testing.T) { @@ -225,10 +195,10 @@ func TestWebhook_BasicDelivery(t *testing.T) { }) n := 3 - te.pushRecordEvents("did:plc:webhooktest", n, false) + te.pushRecordEvents("did:example:webhook", n, false) - msgs := receiver.waitForMessages(n, 2*time.Second) - assert.Len(t, msgs, n, "webhook receiver should get %d events, got %d", n, len(msgs)) + msgs := receiver.waitForMessages(n, 100*time.Millisecond) + require.Len(t, msgs, n) } func TestIdentityEvent_Delivery(t *testing.T) { @@ -240,17 +210,17 @@ func TestIdentityEvent_Delivery(t *testing.T) { require.NoError(t, err) defer consumer.close() - time.Sleep(50 * time.Millisecond) + time.Sleep(20 * time.Millisecond) - te.pushIdentityEvent("did:plc:identity", "alice.bsky.social", models.AccountStatusActive) + te.pushIdentityEvent("did:example:identity", "alice.bsky.social", models.AccountStatusActive) - msgs := consumer.waitForMessages(1, 2*time.Second) - require.Len(t, msgs, 1, "expected 1 identity event") + msgs := consumer.waitForMessages(1, 100*time.Millisecond) + require.Len(t, msgs, 1) msg := msgs[0] assert.Equal(t, "identity", msg.Type) assert.NotNil(t, msg.IdentityEvt) - assert.Equal(t, "did:plc:identity", msg.IdentityEvt.Did) + assert.Equal(t, "did:example:identity", msg.IdentityEvt.Did) assert.Equal(t, "alice.bsky.social", msg.IdentityEvt.Handle) assert.True(t, msg.IdentityEvt.IsActive) assert.Equal(t, models.AccountStatusActive, msg.IdentityEvt.Status) diff --git a/cmd/tap/server_test.go b/cmd/tap/server_test.go index f806e0a5..00210f5e 100644 --- a/cmd/tap/server_test.go +++ b/cmd/tap/server_test.go @@ -3,11 +3,11 @@ package main import ( "encoding/json" "fmt" - "io" "net/http" "strings" "testing" + "github.com/bluesky-social/indigo/cmd/tap/models" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -33,7 +33,7 @@ func TestAddAndRemoveRepos(t *testing.T) { outboxMode: OutboxModeFireAndForget, }) - dids := []string{"did:plc:repo1", "did:plc:repo2", "did:plc:repo3"} + dids := []string{"did:example:repo1", "did:example:repo2", "did:example:repo3"} // Add repos payload := fmt.Sprintf(`{"dids":["%s","%s","%s"]}`, dids[0], dids[1], dids[2]) @@ -42,14 +42,10 @@ func TestAddAndRemoveRepos(t *testing.T) { resp.Body.Close() assert.Equal(t, http.StatusOK, resp.StatusCode) - // Verify repo count - resp, err = http.Get(te.baseURL() + "/stats/repo-count") - require.NoError(t, err) - defer resp.Body.Close() - - var countBody map[string]int64 - require.NoError(t, json.NewDecoder(resp.Body).Decode(&countBody)) - assert.Equal(t, int64(3), countBody["repo_count"]) + // Verify repos exist in DB + var count int64 + te.db.Model(&models.Repo{}).Count(&count) + assert.Equal(t, int64(3), count) // Remove one repo removePayload := fmt.Sprintf(`{"dids":["%s"]}`, dids[0]) @@ -58,46 +54,79 @@ func TestAddAndRemoveRepos(t *testing.T) { resp2.Body.Close() assert.Equal(t, http.StatusOK, resp2.StatusCode) - // Verify repo count decreased - resp3, err := http.Get(te.baseURL() + "/stats/repo-count") + // Verify count decreased + te.db.Model(&models.Repo{}).Count(&count) + assert.Equal(t, int64(2), count) +} + +func TestStatsRepoCount(t *testing.T) { + te := newTestEnv(t, testEnvOpts{ + outboxMode: OutboxModeFireAndForget, + }) + + // Insert repos directly + te.db.Create(&models.Repo{Did: "did:example:stats1", State: models.RepoStateActive}) + te.db.Create(&models.Repo{Did: "did:example:stats2", State: models.RepoStatePending}) + + resp, err := http.Get(te.baseURL() + "/stats/repo-count") require.NoError(t, err) - defer resp3.Body.Close() + defer resp.Body.Close() - var countBody2 map[string]int64 - require.NoError(t, json.NewDecoder(resp3.Body).Decode(&countBody2)) - assert.Equal(t, int64(2), countBody2["repo_count"]) + var body map[string]int64 + require.NoError(t, json.NewDecoder(resp.Body).Decode(&body)) + assert.Equal(t, int64(2), body["repo_count"]) } -func TestStatsEndpoints(t *testing.T) { +func TestStatsRecordCount(t *testing.T) { te := newTestEnv(t, testEnvOpts{ outboxMode: OutboxModeFireAndForget, }) - endpoints := []struct { - path string - key string - }{ - {"/stats/repo-count", "repo_count"}, - {"/stats/record-count", "record_count"}, - {"/stats/outbox-buffer", "outbox_buffer"}, - {"/stats/resync-buffer", "resync_buffer"}, - } - - for _, ep := range endpoints { - t.Run(ep.path, func(t *testing.T) { - resp, err := http.Get(te.baseURL() + ep.path) - require.NoError(t, err) - defer resp.Body.Close() - - assert.Equal(t, http.StatusOK, resp.StatusCode) - - body, err := io.ReadAll(resp.Body) - require.NoError(t, err) - - var result map[string]int64 - require.NoError(t, json.Unmarshal(body, &result)) - _, exists := result[ep.key] - assert.True(t, exists, "response should contain key %q", ep.key) - }) - } + // Insert records directly + te.db.Create(&models.RepoRecord{Did: "did:example:rec", Collection: "app.bsky.feed.post", Rkey: "1", Cid: "cid1"}) + te.db.Create(&models.RepoRecord{Did: "did:example:rec", Collection: "app.bsky.feed.post", Rkey: "2", Cid: "cid2"}) + te.db.Create(&models.RepoRecord{Did: "did:example:rec", Collection: "app.bsky.feed.like", Rkey: "1", Cid: "cid3"}) + + resp, err := http.Get(te.baseURL() + "/stats/record-count") + require.NoError(t, err) + defer resp.Body.Close() + + var body map[string]int64 + require.NoError(t, json.NewDecoder(resp.Body).Decode(&body)) + assert.Equal(t, int64(3), body["record_count"]) +} + +func TestStatsOutboxBuffer(t *testing.T) { + te := newTestEnv(t, testEnvOpts{ + outboxMode: OutboxModeFireAndForget, + }) + + // Push events — they'll be written to the outbox_buffers table + te.pushRecordEvents("did:example:outbox", 4, false) + + resp, err := http.Get(te.baseURL() + "/stats/outbox-buffer") + require.NoError(t, err) + defer resp.Body.Close() + + var body map[string]int64 + require.NoError(t, json.NewDecoder(resp.Body).Decode(&body)) + assert.Equal(t, int64(4), body["outbox_buffer"]) +} + +func TestStatsResyncBuffer(t *testing.T) { + te := newTestEnv(t, testEnvOpts{ + outboxMode: OutboxModeFireAndForget, + }) + + // Insert resync buffer entries directly + te.db.Create(&models.ResyncBuffer{Did: "did:example:resync1", Data: `{}`}) + te.db.Create(&models.ResyncBuffer{Did: "did:example:resync2", Data: `{}`}) + + resp, err := http.Get(te.baseURL() + "/stats/resync-buffer") + require.NoError(t, err) + defer resp.Body.Close() + + var body map[string]int64 + require.NoError(t, json.NewDecoder(resp.Body).Decode(&body)) + assert.Equal(t, int64(2), body["resync_buffer"]) } diff --git a/cmd/tap/tap_test_helpers_test.go b/cmd/tap/tap_test_helpers_test.go index 5010cc60..7ca8066d 100644 --- a/cmd/tap/tap_test_helpers_test.go +++ b/cmd/tap/tap_test_helpers_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "fmt" "io" "log/slog" @@ -13,13 +14,14 @@ import ( "time" "github.com/bluesky-social/indigo/cmd/tap/models" + "github.com/gorilla/websocket" "gorm.io/gorm" ) type testEnvOpts struct { - outboxMode OutboxMode - webhookURL string - retryTimeout time.Duration + outboxMode OutboxMode + webhookURL string + retryTimeout time.Duration eventCacheSize int } @@ -99,7 +101,6 @@ func newTestEnv(t *testing.T, opts testEnvOpts) *testEnv { // Start HTTP server go func() { if err := server.Start(fmt.Sprintf("127.0.0.1:%d", port)); err != nil && err != http.ErrServerClosed { - // Server stopped — only log if it's not a normal shutdown select { case <-ctx.Done(): default: @@ -116,7 +117,7 @@ func newTestEnv(t *testing.T, opts testEnvOpts) *testEnv { conn.Close() break } - time.Sleep(10 * time.Millisecond) + time.Sleep(5 * time.Millisecond) } // Wait for event manager to finish loading @@ -172,7 +173,6 @@ func (te *testEnv) pushRecordEvents(did string, count int, live bool) []uint { } } - // Capture IDs by checking nextID before and after startID := uint(te.events.nextID.Load()) err := te.events.AddRecordEvents(te.ctx, evts, live, func(tx *gorm.DB) error { @@ -184,7 +184,7 @@ func (te *testEnv) pushRecordEvents(did string, count int, live bool) []uint { ids := make([]uint, count) for i := 0; i < count; i++ { - ids[i] = startID + uint(i) + 1 // nextID is pre-incremented via Add(1) + ids[i] = startID + uint(i) + 1 } return ids } @@ -212,6 +212,83 @@ func (te *testEnv) pushIdentityEvent(did, handle string, status models.AccountSt return startID + 1 } +// testConsumer is a WebSocket client that connects to the /channel endpoint +// and collects received events for test assertions. +type testConsumer struct { + conn *websocket.Conn + messages []MarshallableEvt + mu sync.Mutex + done chan struct{} +} + +func newTestConsumer(url string) (*testConsumer, error) { + conn, _, err := websocket.DefaultDialer.Dial(url, nil) + if err != nil { + return nil, err + } + + tc := &testConsumer{ + conn: conn, + done: make(chan struct{}), + } + + go tc.readLoop() + + return tc, nil +} + +func (tc *testConsumer) readLoop() { + defer close(tc.done) + for { + _, message, err := tc.conn.ReadMessage() + if err != nil { + return + } + + var evt MarshallableEvt + if err := json.Unmarshal(message, &evt); err != nil { + continue + } + + tc.mu.Lock() + tc.messages = append(tc.messages, evt) + tc.mu.Unlock() + } +} + +func (tc *testConsumer) waitForMessages(count int, timeout time.Duration) []MarshallableEvt { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + tc.mu.Lock() + n := len(tc.messages) + if n >= count { + result := make([]MarshallableEvt, n) + copy(result, tc.messages) + tc.mu.Unlock() + return result + } + tc.mu.Unlock() + time.Sleep(5 * time.Millisecond) + } + tc.mu.Lock() + defer tc.mu.Unlock() + result := make([]MarshallableEvt, len(tc.messages)) + copy(result, tc.messages) + return result +} + +func (tc *testConsumer) sendAck(id uint) error { + return tc.conn.WriteJSON(WsResponse{ + Type: WsResponseAck, + ID: id, + }) +} + +func (tc *testConsumer) close() { + tc.conn.Close() + <-tc.done +} + // testWebhookReceiver is an HTTP server that collects webhook POST bodies. type testWebhookReceiver struct { server *httptest.Server @@ -250,7 +327,7 @@ func (r *testWebhookReceiver) waitForMessages(count int, timeout time.Duration) return result } r.mu.Unlock() - time.Sleep(10 * time.Millisecond) + time.Sleep(5 * time.Millisecond) } r.mu.Lock() defer r.mu.Unlock() diff --git a/cmd/tap/ws_consumer_test.go b/cmd/tap/ws_consumer_test.go deleted file mode 100644 index 7118a4c0..00000000 --- a/cmd/tap/ws_consumer_test.go +++ /dev/null @@ -1,89 +0,0 @@ -package main - -import ( - "encoding/json" - "sync" - "time" - - "github.com/gorilla/websocket" -) - -// testConsumer is a WebSocket client that connects to the /channel endpoint -// and collects received events for test assertions. -type testConsumer struct { - conn *websocket.Conn - messages []MarshallableEvt - mu sync.Mutex - done chan struct{} -} - -func newTestConsumer(url string) (*testConsumer, error) { - conn, _, err := websocket.DefaultDialer.Dial(url, nil) - if err != nil { - return nil, err - } - - tc := &testConsumer{ - conn: conn, - done: make(chan struct{}), - } - - go tc.readLoop() - - return tc, nil -} - -func (tc *testConsumer) readLoop() { - defer close(tc.done) - for { - _, message, err := tc.conn.ReadMessage() - if err != nil { - return - } - - var evt MarshallableEvt - if err := json.Unmarshal(message, &evt); err != nil { - continue - } - - tc.mu.Lock() - tc.messages = append(tc.messages, evt) - tc.mu.Unlock() - } -} - -// waitForMessages polls until the consumer has received at least count messages -// or the timeout expires. Returns all received messages. -func (tc *testConsumer) waitForMessages(count int, timeout time.Duration) []MarshallableEvt { - deadline := time.Now().Add(timeout) - for time.Now().Before(deadline) { - tc.mu.Lock() - n := len(tc.messages) - if n >= count { - result := make([]MarshallableEvt, n) - copy(result, tc.messages) - tc.mu.Unlock() - return result - } - tc.mu.Unlock() - time.Sleep(10 * time.Millisecond) - } - tc.mu.Lock() - defer tc.mu.Unlock() - result := make([]MarshallableEvt, len(tc.messages)) - copy(result, tc.messages) - return result -} - -// sendAck writes a WsResponse ack back to the server. -func (tc *testConsumer) sendAck(id uint) error { - return tc.conn.WriteJSON(WsResponse{ - Type: WsResponseAck, - ID: id, - }) -} - -func (tc *testConsumer) close() { - tc.conn.Close() - <-tc.done -} -- 2.51.2