···

       
63
       
 
       
	jc, err := jetstream.NewJetstreamClient(

     

       
64
       
 
       
		config.JetstreamEndpoint,

     

       
65
       
 
       
		"appview",

     

       
66
       
-
       
		[]string{tangled.GraphFollowNSID},

     

       
67
       
 
       
		nil,

     

       
68
       
 
       
		slog.Default(),

     

       
69
       
 
       
		wrapper,

     

···

       
63
       
 
       
	jc, err := jetstream.NewJetstreamClient(

     

       
64
       
 
       
		config.JetstreamEndpoint,

     

       
65
       
 
       
		"appview",

     

       
66
       
+
       
		[]string{tangled.GraphFollowNSID, tangled.FeedStarNSID},

     

       
67
       
 
       
		nil,

     

       
68
       
 
       
		slog.Default(),

     

       
69
       
 
       
		wrapper,

     

···

       
43
       
 
       
		return

     

       
44
       
 
       
	}

     

       
45
       
 
       


     

       
0
       
 
       

     

       
46
       
 
       
	j.mu.Lock()

     

       
47
       
 
       
	j.wantedDids[did] = struct{}{}

     

       
48
       
 
       
	j.mu.Unlock()

     
···

       
51
       
 
       
type processor func(context.Context, *models.Event) error

     

       
52
       
 
       


     

       
53
       
 
       
func (j *JetstreamClient) withDidFilter(processFunc processor) processor {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
54
       
 
       
	// since this closure references j.WantedDids; it should auto-update

     

       
55
       
 
       
	// existing instances of the closure when j.WantedDids is mutated

     

       
56
       
 
       
	return func(ctx context.Context, evt *models.Event) error {

     

···

       
43
       
 
       
		return

     

       
44
       
 
       
	}

     

       
45
       
 
       


     

       
46
       
+
       
	j.l.Info("adding did to in-memory filter", "did", did)

     

       
47
       
 
       
	j.mu.Lock()

     

       
48
       
 
       
	j.wantedDids[did] = struct{}{}

     

       
49
       
 
       
	j.mu.Unlock()

     
···

       
52
       
 
       
type processor func(context.Context, *models.Event) error

     

       
53
       
 
       


     

       
54
       
 
       
func (j *JetstreamClient) withDidFilter(processFunc processor) processor {

     

       
55
       
+
       
	// empty filter => all dids allowed

     

       
56
       
+
       
	if len(j.wantedDids) == 0 {

     

       
57
       
+
       
		return processFunc

     

       
58
       
+
       
	}

     

       
59
       
 
       
	// since this closure references j.WantedDids; it should auto-update

     

       
60
       
 
       
	// existing instances of the closure when j.WantedDids is mutated

     

       
61
       
 
       
	return func(ctx context.Context, evt *models.Event) error {

     

···

       
60
       
 
       
	if err != nil {

     

       
61
       
 
       
		return nil, fmt.Errorf("failed to get all Dids: %w", err)

     

       
62
       
 
       
	}

     

       
0
       
 
       

     

       
63
       
 
       
	if len(dids) > 0 {

     

       
64
       
 
       
		h.knotInitialized = true

     

       
65
       
 
       
		close(h.init)

     

···

       
60
       
 
       
	if err != nil {

     

       
61
       
 
       
		return nil, fmt.Errorf("failed to get all Dids: %w", err)

     

       
62
       
 
       
	}

     

       
63
       
+
       


     

       
64
       
 
       
	if len(dids) > 0 {

     

       
65
       
 
       
		h.knotInitialized = true

     

       
66
       
 
       
		close(h.init)