bas.sh
forked from
tangled.org/core
Monorepo for Tangled
1package jetstream
2
3import (
4 "context"
5 "errors"
6 "testing"
7 "time"
8
9 "github.com/bluesky-social/jetstream/pkg/models"
10)
11
12type fakeCursorDB struct {
13 saved int64
14 savedErr error
15}
16
17func (f *fakeCursorDB) GetLastTimeUs() (int64, error) { return f.saved, f.savedErr }
18func (f *fakeCursorDB) SaveLastTimeUs(int64) error { return nil }
19
20func cursorFor(db DB) int64 {
21 j := &JetstreamClient{db: db, wantedDids: make(Set[string])}
22 return *j.getLastTimeUs(context.Background())
23}
24
25const twoDaysUs = int64(2 * 24 * 60 * 60 * 1000 * 1000)
26
27func TestStaleCursorIsHonored(t *testing.T) {
28 old := time.Now().UnixMicro() - 5*twoDaysUs
29 if got := cursorFor(&fakeCursorDB{saved: old}); got != old {
30 t.Fatalf("stale cursor must be honored, not snapped forward: got %d, want %d", got, old)
31 }
32}
33
34func TestZeroCursorIsHonored(t *testing.T) {
35 if got := cursorFor(&fakeCursorDB{saved: 0}); got != 0 {
36 t.Fatalf("cursor 0 must replay from the start: got %d, want 0", got)
37 }
38}
39
40func TestMissingCursorStartsFromNow(t *testing.T) {
41 before := time.Now().UnixMicro()
42 got := cursorFor(&fakeCursorDB{savedErr: errors.New("no row")})
43 after := time.Now().UnixMicro()
44 if got < before || got > after {
45 t.Fatalf("missing cursor must start from now: got %d, want within [%d,%d]", got, before, after)
46 }
47}
48
49func TestLastSeenTracksEveryEventPreFilter(t *testing.T) {
50 j := &JetstreamClient{wantedDids: Set[string]{"did:plc:boltless": {}}}
51 wrapped := j.withDidFilter(func(context.Context, *models.Event) error { return nil })
52
53 _ = wrapped(context.Background(), &models.Event{Did: "did:plc:akshay", TimeUS: 100})
54 if got := j.lastSeenUs.Load(); got != 101 {
55 t.Fatalf("filtered-out event must still advance last-seen: got %d, want 101", got)
56 }
57
58 _ = wrapped(context.Background(), &models.Event{Did: "did:plc:boltless", TimeUS: 200})
59 if got := j.lastSeenUs.Load(); got != 201 {
60 t.Fatalf("matching event must advance last-seen: got %d, want 201", got)
61 }
62}
63
64func TestLastSeenAdvancesAfterProcessing(t *testing.T) {
65 j := &JetstreamClient{wantedDids: make(Set[string])}
66
67 var seenDuringProcess int64
68 wrapped := j.withDidFilter(func(context.Context, *models.Event) error {
69 seenDuringProcess = j.lastSeenUs.Load()
70 return nil
71 })
72
73 _ = wrapped(context.Background(), &models.Event{Did: "did:plc:boltless", TimeUS: 500})
74
75 if seenDuringProcess == 501 {
76 t.Fatal("cursor advanced before the event finished processing; a crash mid-process would skip it")
77 }
78 if got := j.lastSeenUs.Load(); got != 501 {
79 t.Fatalf("cursor must advance once processing returns: got %d, want 501", got)
80 }
81}