Monorepo for Tangled tangled.org

jestream: allow caller to optionally waitForDid

This is handy because the knotserver only subscribes to specific dids,
and we don't want jestream to run without (which defaults to all dids).
The oppsite is true for the appview which wants to see all dids.

Also adds some nil checks to ensure it doesn't break when event is not a
Commit.

anirudh.fi 0aa1229c d527fd0c

verified
Changed files
+25 -9
appview
state
cmd
knotserver
jetstream
knotserver
+7 -2
appview/state/state.go
··· 59 60 resolver := appview.NewResolver() 61 62 - jc, err := jetstream.NewJetstreamClient("appview", []string{tangled.GraphFollowNSID}, nil, db) 63 if err != nil { 64 return nil, fmt.Errorf("failed to create jetstream client: %w", err) 65 } 66 err = jc.StartJetstream(context.Background(), func(ctx context.Context, e *models.Event) error { 67 did := e.Did 68 - raw := e.Commit.Record 69 70 switch e.Commit.Collection { 71 case tangled.GraphFollowNSID:
··· 59 60 resolver := appview.NewResolver() 61 62 + jc, err := jetstream.NewJetstreamClient("appview", []string{tangled.GraphFollowNSID}, nil, db, false) 63 if err != nil { 64 return nil, fmt.Errorf("failed to create jetstream client: %w", err) 65 } 66 err = jc.StartJetstream(context.Background(), func(ctx context.Context, e *models.Event) error { 67 + if e.Kind != models.EventKindCommit { 68 + return nil 69 + } 70 + 71 did := e.Did 72 + fmt.Println("got event", e.Commit.Collection, e.Commit.RKey, e.Commit.Record) 73 + raw := json.RawMessage(e.Commit.Record) 74 75 switch e.Commit.Collection { 76 case tangled.GraphFollowNSID:
+1 -1
cmd/knotserver/main.go
··· 45 jc, err := jetstream.NewJetstreamClient("knotserver", []string{ 46 tangled.PublicKeyNSID, 47 tangled.KnotMemberNSID, 48 - }, nil, db) 49 if err != nil { 50 l.Error("failed to setup jetstream", "error", err) 51 }
··· 45 jc, err := jetstream.NewJetstreamClient("knotserver", []string{ 46 tangled.PublicKeyNSID, 47 tangled.KnotMemberNSID, 48 + }, nil, db, true) 49 if err != nil { 50 l.Error("failed to setup jetstream", "error", err) 51 }
+14 -6
jetstream/jetstream.go
··· 24 25 db DB 26 reconnectCh chan struct{} 27 mu sync.RWMutex 28 } 29 ··· 41 j.reconnectCh <- struct{}{} 42 } 43 44 - func NewJetstreamClient(ident string, collections []string, cfg *client.ClientConfig, db DB) (*JetstreamClient, error) { 45 if cfg == nil { 46 cfg = client.DefaultClientConfig() 47 cfg.WebsocketURL = "wss://jetstream1.us-west.bsky.network/subscribe" ··· 49 } 50 51 return &JetstreamClient{ 52 - cfg: cfg, 53 - ident: ident, 54 - db: db, 55 reconnectCh: make(chan struct{}, 1), 56 }, nil 57 } ··· 82 83 go func() { 84 lastTimeUs := j.getLastTimeUs(ctx) 85 - for len(j.cfg.WantedDids) == 0 { 86 - time.Sleep(time.Second) 87 } 88 j.connectAndRead(ctx, &lastTimeUs) 89 }() 90
··· 24 25 db DB 26 reconnectCh chan struct{} 27 + waitForDid bool 28 mu sync.RWMutex 29 } 30 ··· 42 j.reconnectCh <- struct{}{} 43 } 44 45 + func NewJetstreamClient(ident string, collections []string, cfg *client.ClientConfig, db DB, waitForDid bool) (*JetstreamClient, error) { 46 if cfg == nil { 47 cfg = client.DefaultClientConfig() 48 cfg.WebsocketURL = "wss://jetstream1.us-west.bsky.network/subscribe" ··· 50 } 51 52 return &JetstreamClient{ 53 + cfg: cfg, 54 + ident: ident, 55 + db: db, 56 + 57 + // This will make the goroutine in StartJetstream wait until 58 + // cfg.WantedDids has been populated, typically using UpdateDids. 59 + waitForDid: waitForDid, 60 reconnectCh: make(chan struct{}, 1), 61 }, nil 62 } ··· 87 88 go func() { 89 lastTimeUs := j.getLastTimeUs(ctx) 90 + if j.waitForDid { 91 + for len(j.cfg.WantedDids) == 0 { 92 + time.Sleep(time.Second) 93 + } 94 } 95 + logger.Info("done waiting for did") 96 j.connectAndRead(ctx, &lastTimeUs) 97 }() 98
+3
knotserver/jetstream.go
··· 108 109 func (h *Handle) processMessages(ctx context.Context, event *models.Event) error { 110 did := event.Did 111 112 raw := json.RawMessage(event.Commit.Record) 113
··· 108 109 func (h *Handle) processMessages(ctx context.Context, event *models.Event) error { 110 did := event.Did 111 + if event.Kind != models.EventKindCommit { 112 + return nil 113 + } 114 115 raw := json.RawMessage(event.Commit.Record) 116