this repo has no description
database backed event stream persister
+138
-42
+20
-3
bgs/bgs.go
+20
-3
bgs/bgs.go
···
5
5
"errors"
6
6
"fmt"
7
7
"net/http"
8
+
_ "net/http/pprof"
8
9
"net/url"
9
10
"strconv"
10
11
"strings"
···
26
27
"github.com/labstack/echo/v4"
27
28
"github.com/labstack/echo/v4/middleware"
28
29
promclient "github.com/prometheus/client_golang/prometheus"
30
+
"go.opentelemetry.io/otel"
29
31
"gorm.io/gorm"
30
32
)
31
33
···
63
65
ix.CreateExternalUser = bgs.createExternalUser
64
66
bgs.slurper = NewSlurper(db, bgs.handleFedEvent, ssl)
65
67
return bgs
68
+
}
69
+
70
+
func (bgs *BGS) StartDebug(listen string) error {
71
+
http.Handle("/prometheus", prometheusHandler())
72
+
73
+
return http.ListenAndServe(listen, nil)
66
74
}
67
75
68
76
func (bgs *BGS) Start(listen string) error {
···
152
160
case evt.Append != nil:
153
161
header.Op = events.EvtKindRepoAppend
154
162
obj = evt.Append
163
+
case evt.Error != nil:
164
+
header.Op = events.EvtKindErrorFrame
165
+
obj = evt.Error
166
+
case evt.Info != nil:
167
+
header.Op = events.EvtKindInfoFrame
168
+
obj = evt.Info
155
169
default:
156
170
return fmt.Errorf("unrecognized event kind")
157
171
}
···
227
241
}
228
242
229
243
var prevcid *cid.Cid
230
-
if evt.Prev != "" {
231
-
c, err := cid.Decode(evt.Prev)
244
+
if evt.Prev != nil {
245
+
c, err := cid.Decode(*evt.Prev)
232
246
if err != nil {
233
247
return fmt.Errorf("invalid value for prev cid in event: %w", err)
234
248
}
···
257
271
}
258
272
259
273
func (s *BGS) createExternalUser(ctx context.Context, did string) (*models.ActorInfo, error) {
274
+
ctx, span := otel.Tracer("bgs").Start(ctx, "createExternalUser")
275
+
defer span.End()
276
+
260
277
log.Infof("create external user: %s", did)
261
278
doc, err := s.didr.GetDocument(ctx, did)
262
279
if err != nil {
···
328
345
329
346
exu, err := s.Index.LookupUserByDid(ctx, did)
330
347
if err == nil {
331
-
log.Warnf("lost the race to create a new user: %d", did)
348
+
log.Warnf("lost the race to create a new user: %s", did)
332
349
return exu, nil
333
350
}
334
351
+8
-3
cmd/bigsky/main.go
+8
-3
cmd/bigsky/main.go
···
2
2
3
3
import (
4
4
"context"
5
+
"fmt"
5
6
"os"
6
7
"path/filepath"
7
8
"time"
···
16
17
"github.com/bluesky-social/indigo/plc"
17
18
"github.com/bluesky-social/indigo/repomgr"
18
19
19
-
"net/http"
20
20
_ "net/http/pprof"
21
21
22
22
logging "github.com/ipfs/go-log"
···
131
131
132
132
repoman := repomgr.NewRepoManager(db, cstore, kmgr)
133
133
134
-
evtman := events.NewEventManager()
134
+
dbp, err := events.NewDbPersistence(db, cstore)
135
+
if err != nil {
136
+
return fmt.Errorf("setting up db event persistence: %w", err)
137
+
}
138
+
139
+
evtman := events.NewEventManager(dbp)
135
140
136
141
go evtman.Run()
137
142
···
152
157
153
158
// set up pprof endpoint
154
159
go func() {
155
-
if err := http.ListenAndServe("localhost:2471", nil); err != nil {
160
+
if err := bgs.StartDebug("localhost:2471"); err != nil {
156
161
panic(err)
157
162
}
158
163
}()
+22
-2
cmd/gosky/main.go
+22
-2
cmd/gosky/main.go
···
9
9
"fmt"
10
10
"net/http"
11
11
"os"
12
+
"os/signal"
12
13
"strings"
14
+
"syscall"
13
15
"time"
14
16
15
17
comatproto "github.com/bluesky-social/indigo/api/atproto"
···
804
806
},
805
807
},
806
808
Action: func(cctx *cli.Context) error {
809
+
ctx, cancel := context.WithCancel(context.Background())
810
+
811
+
ch := make(chan os.Signal)
812
+
signal.Notify(ch, syscall.SIGINT)
813
+
814
+
go func() {
815
+
<-ch
816
+
cancel()
817
+
}()
818
+
807
819
d := websocket.DefaultDialer
808
820
con, _, err := d.Dial(cctx.Args().First(), http.Header{})
809
821
if err != nil {
···
817
829
fmt.Println("Stream Exited", time.Now().Format(time.RFC3339))
818
830
}()
819
831
820
-
return events.HandleRepoStream(context.TODO(), con, &events.RepoStreamCallbacks{
832
+
defer func() {
833
+
_ = con.Close()
834
+
}()
835
+
836
+
return events.HandleRepoStream(ctx, con, &events.RepoStreamCallbacks{
821
837
Append: func(evt *events.RepoAppend) error {
822
838
if jsonfmt {
823
839
b, err := json.Marshal(evt)
···
837
853
fmt.Println(string(b))
838
854
839
855
} else {
840
-
fmt.Printf("(%d) RepoAppend: %s (%s -> %s)\n", evt.Seq, evt.Repo, evt.Prev, evt.Commit)
856
+
pstr := "<nil>"
857
+
if evt.Prev != nil {
858
+
pstr = *evt.Prev
859
+
}
860
+
fmt.Printf("(%d) RepoAppend: %s (%s -> %s)\n", evt.Seq, evt.Repo, pstr, evt.Commit)
841
861
}
842
862
843
863
return nil
+26
-10
events/cbor_gen.go
+26
-10
events/cbor_gen.go
···
173
173
return err
174
174
}
175
175
176
-
if len(t.Prev) > cbg.MaxLength {
177
-
return xerrors.Errorf("Value in field t.Prev was too long")
178
-
}
176
+
if t.Prev == nil {
177
+
if _, err := cw.Write(cbg.CborNull); err != nil {
178
+
return err
179
+
}
180
+
} else {
181
+
if len(*t.Prev) > cbg.MaxLength {
182
+
return xerrors.Errorf("Value in field t.Prev was too long")
183
+
}
179
184
180
-
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(t.Prev))); err != nil {
181
-
return err
182
-
}
183
-
if _, err := io.WriteString(w, string(t.Prev)); err != nil {
184
-
return err
185
+
if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(*t.Prev))); err != nil {
186
+
return err
187
+
}
188
+
if _, err := io.WriteString(w, string(*t.Prev)); err != nil {
189
+
return err
190
+
}
185
191
}
186
192
187
193
// t.Repo (string) (string)
···
402
408
case "prev":
403
409
404
410
{
405
-
sval, err := cbg.ReadString(cr)
411
+
b, err := cr.ReadByte()
406
412
if err != nil {
407
413
return err
408
414
}
415
+
if b != cbg.CborNull[0] {
416
+
if err := cr.UnreadByte(); err != nil {
417
+
return err
418
+
}
409
419
410
-
t.Prev = string(sval)
420
+
sval, err := cbg.ReadString(cr)
421
+
if err != nil {
422
+
return err
423
+
}
424
+
425
+
t.Prev = (*string)(&sval)
426
+
}
411
427
}
412
428
// t.Repo (string) (string)
413
429
case "repo":
+9
-6
events/events.go
+9
-6
events/events.go
···
1
1
package events
2
2
3
3
import (
4
+
"context"
4
5
"fmt"
5
6
6
7
logging "github.com/ipfs/go-log"
···
18
19
persister EventPersistence
19
20
}
20
21
21
-
func NewEventManager() *EventManager {
22
+
func NewEventManager(persister EventPersistence) *EventManager {
22
23
return &EventManager{
23
24
ops: make(chan *Operation),
24
25
closed: make(chan struct{}),
25
26
bufferSize: 1024,
26
-
persister: NewMemPersister(),
27
+
persister: persister,
27
28
}
28
29
}
29
30
···
53
54
}
54
55
}
55
56
case opSend:
56
-
em.persister.Persist(op.evt)
57
+
if err := em.persister.Persist(context.TODO(), op.evt); err != nil {
58
+
log.Errorf("failed to persist outbound event: %s", err)
59
+
}
57
60
58
61
for _, s := range em.subs {
59
62
if s.filter(op.evt) {
···
107
110
// Repo is the DID of the repo this event is about
108
111
Repo string `cborgen:"repo"`
109
112
110
-
Commit string `cborgen:"commit"`
111
-
Prev string `cborgen:"prev"`
113
+
Commit string `cborgen:"commit"`
114
+
Prev *string `cborgen:"prev"`
112
115
//Commit cid.Cid `cborgen:"commit"`
113
116
//Prev *cid.Cid `cborgen:"prev"`
114
117
···
155
158
156
159
go func() {
157
160
if since != nil {
158
-
if err := em.persister.Playback(*since, func(e *RepoStreamEvent) error {
161
+
if err := em.persister.Playback(context.TODO(), *since, func(e *RepoStreamEvent) error {
159
162
select {
160
163
case <-done:
161
164
return fmt.Errorf("shutting down")
+5
-4
events/persist.go
+5
-4
events/persist.go
···
1
1
package events
2
2
3
3
import (
4
+
"context"
4
5
"sync"
5
6
)
6
7
7
8
type EventPersistence interface {
8
-
Persist(e *RepoStreamEvent) error
9
-
Playback(since int64, cb func(*RepoStreamEvent) error) error
9
+
Persist(ctx context.Context, e *RepoStreamEvent) error
10
+
Playback(ctx context.Context, since int64, cb func(*RepoStreamEvent) error) error
10
11
}
11
12
12
13
// MemPersister is the most naive implementation of event persistence
···
21
22
return &MemPersister{}
22
23
}
23
24
24
-
func (mp *MemPersister) Persist(e *RepoStreamEvent) error {
25
+
func (mp *MemPersister) Persist(ctx context.Context, e *RepoStreamEvent) error {
25
26
mp.lk.Lock()
26
27
defer mp.lk.Unlock()
27
28
mp.seq++
···
36
37
return nil
37
38
}
38
39
39
-
func (mp *MemPersister) Playback(since int64, cb func(*RepoStreamEvent) error) error {
40
+
func (mp *MemPersister) Playback(ctx context.Context, since int64, cb func(*RepoStreamEvent) error) error {
40
41
mp.lk.Lock()
41
42
l := len(mp.buf)
42
43
mp.lk.Unlock()
+4
indexer/crawler.go
+4
indexer/crawler.go
···
5
5
6
6
"github.com/bluesky-social/indigo/events"
7
7
"github.com/bluesky-social/indigo/models"
8
+
"go.opentelemetry.io/otel"
8
9
)
9
10
10
11
type CrawlDispatcher struct {
···
164
165
}
165
166
166
167
func (c *CrawlDispatcher) Crawl(ctx context.Context, ai *models.ActorInfo) error {
168
+
ctx, span := otel.Tracer("crawler").Start(ctx, "addToCrawler")
169
+
defer span.End()
170
+
167
171
select {
168
172
case c.ingest <- ai:
169
173
return nil
+6
-2
indexer/indexer.go
+6
-2
indexer/indexer.go
···
6
6
"errors"
7
7
"fmt"
8
8
"strings"
9
+
"time"
9
10
10
11
atproto "github.com/bluesky-social/indigo/api/atproto"
11
12
bsky "github.com/bluesky-social/indigo/api/bsky"
···
14
15
"github.com/bluesky-social/indigo/notifs"
15
16
"github.com/bluesky-social/indigo/plc"
16
17
"github.com/bluesky-social/indigo/repomgr"
18
+
"github.com/bluesky-social/indigo/util"
17
19
"github.com/bluesky-social/indigo/xrpc"
18
20
"github.com/ipfs/go-cid"
19
21
logging "github.com/ipfs/go-log"
···
116
118
}
117
119
118
120
// TODO: these should be cids on the wire
119
-
var prevstr string
121
+
var prevstr *string
120
122
if evt.OldRoot != nil {
121
-
prevstr = evt.OldRoot.String()
123
+
s := evt.OldRoot.String()
124
+
prevstr = &s
122
125
}
123
126
124
127
log.Infow("Sending event", "did", did)
···
128
131
Prev: prevstr,
129
132
Blocks: evt.RepoSlice,
130
133
Commit: evt.NewRoot.String(),
134
+
Time: time.Now().Format(util.ISO8601),
131
135
},
132
136
PrivUid: evt.User,
133
137
}); err != nil {
+1
-1
indexer/posts_test.go
+1
-1
indexer/posts_test.go
···
58
58
59
59
repoman := repomgr.NewRepoManager(maindb, cs, &util.FakeKeyManager{})
60
60
notifman := notifs.NewNotificationManager(maindb, repoman.GetRecord)
61
-
evtman := events.NewEventManager()
61
+
evtman := events.NewEventManager(events.NewMemPersister())
62
62
63
63
didr := testPLC(t)
64
64
+8
-5
pds/server.go
+8
-5
pds/server.go
···
22
22
"github.com/bluesky-social/indigo/notifs"
23
23
"github.com/bluesky-social/indigo/plc"
24
24
"github.com/bluesky-social/indigo/repomgr"
25
+
bsutil "github.com/bluesky-social/indigo/util"
25
26
"github.com/bluesky-social/indigo/xrpc"
26
27
gojwt "github.com/golang-jwt/jwt"
27
28
"github.com/gorilla/websocket"
···
69
70
return nil, err
70
71
}
71
72
72
-
evtman := events.NewEventManager()
73
+
evtman := events.NewEventManager(events.NewMemPersister())
73
74
74
75
kmgr := indexer.NewKeyManager(didr, serkey)
75
76
···
142
143
}
143
144
144
145
var pcid *cid.Cid
145
-
if evt.Prev != "" {
146
-
prev, err := cid.Decode(evt.Prev)
146
+
if evt.Prev != nil {
147
+
prev, err := cid.Decode(*evt.Prev)
147
148
if err != nil {
148
149
return err
149
150
}
···
251
252
return nil, err
252
253
}
253
254
254
-
var prevcid string
255
+
var prevcid *string
255
256
if evt.OldRoot != nil {
256
-
prevcid = evt.OldRoot.String()
257
+
s := evt.OldRoot.String()
258
+
prevcid = &s
257
259
}
258
260
259
261
out := &events.RepoAppend{
260
262
Prev: prevcid,
261
263
Blocks: evt.RepoSlice,
262
264
Repo: did,
265
+
Time: time.Now().Format(bsutil.ISO8601),
263
266
//PrivUid: evt.User,
264
267
}
265
268
+28
-5
repomgr/repomgr.go
+28
-5
repomgr/repomgr.go
···
215
215
return "", cid.Undef, fmt.Errorf("updating user head: %w", err)
216
216
}
217
217
218
+
var oldroot *cid.Cid
219
+
if head.Defined() {
220
+
oldroot = &head
221
+
}
222
+
218
223
if rm.events != nil {
219
224
rm.events(ctx, &RepoEvent{
220
225
User: user,
221
-
OldRoot: &head,
226
+
OldRoot: oldroot,
222
227
NewRoot: nroot,
223
228
Ops: []RepoOp{{
224
229
Kind: EvtKindCreateRecord,
···
277
282
return cid.Undef, fmt.Errorf("updating user head: %w", err)
278
283
}
279
284
285
+
var oldroot *cid.Cid
286
+
if head.Defined() {
287
+
oldroot = &head
288
+
}
289
+
280
290
if rm.events != nil {
281
291
rm.events(ctx, &RepoEvent{
282
292
User: user,
283
-
OldRoot: &head,
293
+
OldRoot: oldroot,
284
294
NewRoot: nroot,
285
295
Ops: []RepoOp{{
286
296
Kind: EvtKindUpdateRecord,
···
337
347
if err := rm.updateUserRepoHead(ctx, user, nroot); err != nil {
338
348
return fmt.Errorf("updating user head: %w", err)
339
349
}
350
+
var oldroot *cid.Cid
351
+
if head.Defined() {
352
+
oldroot = &head
353
+
}
340
354
341
355
if rm.events != nil {
342
356
rm.events(ctx, &RepoEvent{
343
357
User: user,
344
-
OldRoot: &head,
358
+
OldRoot: oldroot,
345
359
NewRoot: nroot,
346
360
Ops: []RepoOp{{
347
361
Kind: EvtKindDeleteRecord,
···
751
765
return fmt.Errorf("updating user head: %w", err)
752
766
}
753
767
768
+
var oldroot *cid.Cid
769
+
if head.Defined() {
770
+
oldroot = &head
771
+
}
772
+
754
773
if rm.events != nil {
755
774
rm.events(ctx, &RepoEvent{
756
775
User: user,
757
-
OldRoot: &head,
776
+
OldRoot: oldroot,
758
777
NewRoot: nroot,
759
778
RepoSlice: rslice,
760
779
Ops: ops,
···
865
884
// TODO: this will lead to things being in an inconsistent state
866
885
return fmt.Errorf("failed to update repo head: %w", err)
867
886
}
887
+
var oldroot *cid.Cid
888
+
if old.Defined() {
889
+
oldroot = &old
890
+
}
868
891
869
892
if rm.events != nil {
870
893
rm.events(ctx, &RepoEvent{
871
894
User: user,
872
-
OldRoot: &old,
895
+
OldRoot: oldroot,
873
896
NewRoot: nu,
874
897
RepoSlice: slice,
875
898
Ops: ops,