+16 -3

jetstream/jetstream.go

View file Reviewed

··· 30 30 ident string 31 31 l *slog.Logger 32 32 33 33 - logDids bool 34 34 - wantedDids Set[string] 33 33 + logDids bool 34 34 + wantedDids Set[string] 35 35 + unfilteredNsids Set[string] 35 36 db DB 36 37 waitForDid bool 37 38 mu sync.RWMutex ··· 55 56 j.mu.Unlock() 56 57 } 57 58 59 59 + func (j *JetstreamClient) ExemptCollection(nsid string) { 60 60 + j.mu.Lock() 61 61 + j.unfilteredNsids[nsid] = struct{}{} 62 62 + j.mu.Unlock() 63 63 + } 64 64 + 58 65 func (j *JetstreamClient) RemoveDid(did string) { 59 66 if did == "" { 60 67 return ··· 82 89 matches = true 83 90 } 84 91 } 92 92 + if !matches && evt.Commit != nil { 93 93 + if _, ok := j.unfilteredNsids[evt.Commit.Collection]; ok { 94 94 + matches = true 95 95 + } 96 96 + } 85 97 j.mu.RUnlock() 86 98 87 99 var err error ··· 106 118 ident: ident, 107 119 db: db, 108 120 l: logger, 109 109 - wantedDids: make(map[string]struct{}), 121 121 + wantedDids: make(map[string]struct{}), 122 122 + unfilteredNsids: make(map[string]struct{}), 110 123 111 124 logDids: logDids, 112 125

+3 -1

spindle/ingester.go

View file Reviewed

··· 27 27 switch e.Commit.Collection { 28 28 case tangled.SpindleMemberNSID: 29 29 err = s.ingestMember(ctx, e) 30 30 - case tangled.RepoNSID, tangled.RepoCollaboratorNSID: 30 30 + case tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID: 31 31 if evt, ok := jetstreamToTapEvent(e); ok { 32 32 err = s.tap.processEvent(ctx, evt) 33 33 } ··· 68 68 Collection: syntax.NSID(e.Commit.Collection), 69 69 Action: action, 70 70 Record: e.Commit.Record, 71 71 + // jetstream is only used for live 72 72 + Live: true, 71 73 }, 72 74 }, true 73 75 }

+3

spindle/server.go

View file Reviewed

··· 123 123 tangled.SpindleMemberNSID, 124 124 tangled.RepoNSID, 125 125 tangled.RepoCollaboratorNSID, 126 126 + tangled.RepoPullNSID, 126 127 } 127 128 jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, log.SubLogger(logger, "jetstream"), d, true, true) 128 129 if err != nil { 129 130 return nil, fmt.Errorf("failed to setup jetstream client: %w", err) 130 131 } 131 132 jc.AddDid(cfg.Server.Owner) 133 133 + // pull records are created by arbitrary users too, same hack as in tap 134 134 + jc.ExemptCollection(tangled.RepoPullNSID) 132 135 133 136 // Check if the spindle knows about any Dids; 134 137 dids, err := d.GetAllDids()