···

       
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 {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
67
       
 
       
		did := e.Did

     

       
68
       
-
       
		raw := e.Commit.Record

     

       
0
       
 
       

     

       
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:

     

···

       
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
       
 
       
	}

     

···

       
24
       
 
       


     

       
25
       
 
       
	db          DB

     

       
26
       
 
       
	reconnectCh chan struct{}

     

       
0
       
 
       

     

       
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,

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
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)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
87
       
 
       
		}

     

       
0
       
 
       

     

       
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
       
 
       


     

···

       
108
       
 
       


     

       
109
       
 
       
func (h *Handle) processMessages(ctx context.Context, event *models.Event) error {

     

       
110
       
 
       
	did := event.Did

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
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