+4 -4

events/cbor_gen.go

reviewed

··· 1351 1351 1352 1352 return nil 1353 1353 } 1354 1354 - func (t *LabelEvent) MarshalCBOR(w io.Writer) error { 1354 1354 + func (t *LabelBatch) MarshalCBOR(w io.Writer) error { 1355 1355 if t == nil { 1356 1356 _, err := w.Write(cbg.CborNull) 1357 1357 return err ··· 1412 1412 return nil 1413 1413 } 1414 1414 1415 1415 - func (t *LabelEvent) UnmarshalCBOR(r io.Reader) (err error) { 1416 1416 - *t = LabelEvent{} 1415 1415 + func (t *LabelBatch) UnmarshalCBOR(r io.Reader) (err error) { 1416 1416 + *t = LabelBatch{} 1417 1417 1418 1418 cr := cbg.NewCborReader(r) 1419 1419 ··· 1432 1432 } 1433 1433 1434 1434 if extra > cbg.MaxLength { 1435 1435 - return fmt.Errorf("LabelEvent: map struct too large (%d)", extra) 1435 1435 + return fmt.Errorf("LabelBatch: map struct too large (%d)", extra) 1436 1436 } 1437 1437 1438 1438 var name string

+43 -29

events/label_events.go

reviewed

··· 1 1 package events 2 2 3 3 import ( 4 4 + "context" 4 5 "fmt" 5 6 ) 6 7 ··· 14 15 persister LabelEventPersistence 15 16 } 16 17 17 17 - func NewLabelEventManager() *LabelEventManager { 18 18 + func NewLabelEventManager(persister LabelEventPersistence) *LabelEventManager { 18 19 return &LabelEventManager{ 19 20 ops: make(chan *LabelOperation), 20 21 closed: make(chan struct{}), 21 22 bufferSize: 1024, 22 22 - persister: NewMemLabelPersister(), 23 23 + persister: persister, 23 24 } 24 25 } 25 26 26 27 type LabelOperation struct { 27 28 op int 28 29 sub *LabelSubscriber 29 29 - evt *LabelEvent 30 30 + evt *LabelStreamEvent 30 31 } 31 32 32 33 func (em *LabelEventManager) Run() { ··· 34 35 switch op.op { 35 36 case opSubscribe: 36 37 em.subs = append(em.subs, op.sub) 37 37 - op.sub.outgoing <- &LabelEvent{} 38 38 case opUnsubscribe: 39 39 for i, s := range em.subs { 40 40 if s == op.sub { ··· 44 44 } 45 45 } 46 46 case opSend: 47 47 - em.persister.Persist(op.evt) 47 47 + if err := em.persister.Persist(context.TODO(), op.evt); err != nil { 48 48 + log.Errorf("failed to persist outbound event: %s", err) 49 49 + } 48 50 49 51 for _, s := range em.subs { 50 52 if s.filter(op.evt) { ··· 62 64 } 63 65 64 66 type LabelSubscriber struct { 65 65 - outgoing chan *LabelEvent 67 67 + outgoing chan *LabelStreamEvent 66 68 67 67 - filter func(*LabelEvent) bool 69 69 + filter func(*LabelStreamEvent) bool 68 70 69 71 done chan struct{} 70 72 } 71 73 72 72 - type LabelEvent struct { 74 74 + const ( 75 75 + LEvtKindErrorFrame = -1 76 76 + LEvtKindLabelBatch = 1 77 77 + LEvtKindInfoFrame = 2 78 78 + ) 79 79 + 80 80 + // EventHeader shared with repo events 81 81 + 82 82 + type LabelStreamEvent struct { 83 83 + Batch *LabelBatch 84 84 + Info *InfoFrame 85 85 + Error *ErrorFrame 86 86 + } 87 87 + 88 88 + type LabelBatch struct { 73 89 Seq int64 `cborgen:"seq"` 74 90 Labels []Label `cborgen:"labels"` 91 91 + // TODO: time? 75 92 } 76 93 77 94 // this is here, instead of under 'labeling' package, to avoid an import loop ··· 85 102 LabelUri *string `json:"labeluri,omitempty" cborgen:"labeluri"` 86 103 } 87 104 88 88 - func (em *LabelEventManager) AddEvent(ev *LabelEvent) error { 105 105 + func (em *LabelEventManager) AddEvent(ev *LabelStreamEvent) error { 89 106 select { 90 107 case em.ops <- &LabelOperation{ 91 108 op: opSend, ··· 97 114 } 98 115 } 99 116 100 100 - func (em *LabelEventManager) Subscribe(filter func(*LabelEvent) bool, since *int64) (<-chan *LabelEvent, func(), error) { 117 117 + func (em *LabelEventManager) Subscribe(ctx context.Context, filter func(*LabelStreamEvent) bool, since *int64) (<-chan *LabelStreamEvent, func(), error) { 101 118 if filter == nil { 102 102 - filter = func(*LabelEvent) bool { return true } 119 119 + filter = func(*LabelStreamEvent) bool { return true } 103 120 } 104 121 105 122 done := make(chan struct{}) 106 123 sub := &LabelSubscriber{ 107 107 - outgoing: make(chan *LabelEvent, em.bufferSize), 124 124 + outgoing: make(chan *LabelStreamEvent, em.bufferSize), 108 125 filter: filter, 109 126 done: done, 110 127 } 111 128 112 112 - select { 113 113 - case em.ops <- &LabelOperation{ 114 114 - op: opSubscribe, 115 115 - sub: sub, 116 116 - }: 117 117 - case <-em.closed: 118 118 - return nil, nil, fmt.Errorf("event manager shut down") 119 119 - } 120 120 - 121 121 - // receive the 'ack' that ensures our sub was received 122 122 - <-sub.outgoing 123 123 - 124 124 - if since != nil { 125 125 - go func() { 126 126 - if err := em.persister.Playback(*since, func(e *LabelEvent) error { 129 129 + go func() { 130 130 + if since != nil { 131 131 + if err := em.persister.Playback(ctx, *since, func(e *LabelStreamEvent) error { 127 132 select { 128 133 case <-done: 129 134 return fmt.Errorf("shutting down") ··· 133 138 }); err != nil { 134 139 log.Errorf("events playback: %s", err) 135 140 } 136 136 - }() 137 137 - } 141 141 + } 142 142 + 143 143 + select { 144 144 + case em.ops <- &LabelOperation{ 145 145 + op: opSubscribe, 146 146 + sub: sub, 147 147 + }: 148 148 + case <-em.closed: 149 149 + log.Errorf("failed to subscribe, event manager shut down") 150 150 + } 151 151 + }() 138 152 139 153 cleanup := func() { 140 154 close(done)

+17 -7

events/label_persist.go

reviewed

··· 1 1 package events 2 2 3 3 - import "sync" 3 3 + import ( 4 4 + "context" 5 5 + "sync" 6 6 + ) 4 7 5 8 type LabelEventPersistence interface { 6 6 - Persist(e *LabelEvent) 7 7 - Playback(since int64, cb func(*LabelEvent) error) error 9 9 + Persist(ctx context.Context, e *LabelStreamEvent) error 10 10 + Playback(ctx context.Context, since int64, cb func(*LabelStreamEvent) error) error 8 11 } 9 12 10 13 // MemLabelPersister is the most naive implementation of event persistence 11 14 // ill do better later 12 15 type MemLabelPersister struct { 13 13 - buf []*LabelEvent 16 16 + buf []*LabelStreamEvent 14 17 lk sync.Mutex 15 18 seq int64 16 19 } ··· 19 22 return &MemLabelPersister{} 20 23 } 21 24 22 22 - func (mp *MemLabelPersister) Persist(e *LabelEvent) { 25 25 + func (mp *MemLabelPersister) Persist(ctx context.Context, e *LabelStreamEvent) error { 23 26 mp.lk.Lock() 24 27 defer mp.lk.Unlock() 25 28 mp.seq++ 26 26 - e.Seq = mp.seq 29 29 + switch { 30 30 + case e.Batch != nil: 31 31 + e.Batch.Seq = mp.seq 32 32 + default: 33 33 + panic("no event in persist call") 34 34 + } 27 35 mp.buf = append(mp.buf, e) 36 36 + 37 37 + return nil 28 38 } 29 39 30 30 - func (mp *MemLabelPersister) Playback(since int64, cb func(*LabelEvent) error) error { 40 40 + func (mp *MemLabelPersister) Playback(ctx context.Context, since int64, cb func(*LabelStreamEvent) error) error { 31 41 mp.lk.Lock() 32 42 l := len(mp.buf) 33 43 mp.lk.Unlock()

+1 -1

gen/main.go

reviewed

··· 36 36 panic(err) 37 37 } 38 38 39 39 - if err := cbg.WriteMapEncodersToFile("events/cbor_gen.go", "events", events.EventHeader{}, events.RepoAppend{}, events.RepoOp{}, events.InfoFrame{}, events.ErrorFrame{}, events.Label{}, events.LabelEvent{}); err != nil { 39 39 + if err := cbg.WriteMapEncodersToFile("events/cbor_gen.go", "events", events.EventHeader{}, events.RepoAppend{}, events.RepoOp{}, events.InfoFrame{}, events.ErrorFrame{}, events.Label{}, events.LabelBatch{}); err != nil { 40 40 panic(err) 41 41 } 42 42 }

+14 -12

labeling/service.go

reviewed

··· 59 59 return nil, err 60 60 } 61 61 62 62 - db.AutoMigrate(&pds.User{}) 63 63 - db.AutoMigrate(&pds.Peering{}) 62 62 + db.AutoMigrate(models.PDS{}) 64 63 65 64 didr := &api.PLCServer{Host: plcUrl} 66 66 - kmgr := indexer.NewKeyManager(didr, nil) 67 67 - levtman := events.NewLabelEventManager() 65 65 + kmgr := indexer.NewKeyManager(didr, serkey) 66 66 + levtman := events.NewLabelEventManager(events.NewMemLabelPersister()) 68 67 repoman := repomgr.NewRepoManager(db, cs, kmgr) 69 68 70 69 user := &LabelmakerRepoConfig{ ··· 74 73 userId: 1, 75 74 } 76 75 77 77 - var kl = KeywordLabeler{value: "rude", keywords: []string{"🍆", "sex", "ab"}} 76 76 + var kl = KeywordLabeler{value: "rude", keywords: []string{"🍆", "sex", "ab", "before", "yours", "the"}} 78 77 79 78 s := &Server{ 80 79 db: db, ··· 104 103 105 104 // subscribe our RepoEvent slurper to the BGS, to receive incoming records for labeler 106 105 useWebsocketSSL := false 106 106 + log.Infof("subscribing to BGS: %s (SSL=%v)", bgsUrl, useWebsocketSSL) 107 107 s.bgsSlurper.SubscribeToPds(ctx, bgsUrl, useWebsocketSSL) 108 108 109 109 // NOTE: this is where outgoing RepoEvents could be generated ··· 127 127 128 128 // incoming repo events 129 129 func (s *Server) handleBgsRepoEvent(ctx context.Context, pds *models.PDS, evt *events.RepoStreamEvent) error { 130 130 - log.Info("got RepoEvent from BGS") 131 130 now := time.Now().Format(util.ISO8601) 132 131 switch { 133 132 case evt.Append != nil: ··· 146 145 // filter to creation/update of ony post/profile records 147 146 // TODO(bnewbold): how do I 'switch' on a tuple here in golang, instead of nested switch? 148 147 switch op.Action { 149 149 - case "createRecord", "updateRecord": 148 148 + case "create", "update": 149 149 + log.Infof("labeling record: %v", uri) 150 150 switch nsid { 151 151 case "app.bsky.feed.post": 152 152 cid, rec, err := sliceRepo.GetRecord(ctx, op.Path) ··· 212 212 log.Infof("persisted label: %s", labeluri) 213 213 } 214 214 215 215 - // ... then re-publish as LabelEvent 215 215 + // ... then re-publish as LabelStreamEvent 216 216 log.Infof("%s", labels) 217 217 if len(labels) > 0 { 218 218 - lev := events.LabelEvent{ 218 218 + lev := events.LabelStreamEvent{ 219 219 // XXX(bnewbold): what should sequence number be? do I need to maintain that? 220 220 - Seq: 0, 221 221 - Labels: labels, 220 220 + Batch: &events.LabelBatch{ 221 221 + Seq: 0, 222 222 + Labels: labels, 223 223 + }, 222 224 } 223 225 err = s.levents.AddEvent(&lev) 224 226 if err != nil { 225 225 - return fmt.Errorf("failed to publish LabelEvent: %w", err) 227 227 + return fmt.Errorf("failed to publish LabelStreamEvent: %w", err) 226 228 } 227 229 } 228 230 // TODO: update state that we successfully processed the repo event (aka, persist "last" seq in database, or something like that)

+52 -29

labeling/ws_endpoints.go

reviewed

··· 2 2 3 3 import ( 4 4 "fmt" 5 5 + "strconv" 5 6 6 7 "github.com/bluesky-social/indigo/events" 7 7 - "github.com/bluesky-social/indigo/pds" 8 8 + lexutil "github.com/bluesky-social/indigo/lex/util" 8 9 9 10 "github.com/gorilla/websocket" 10 11 "github.com/labstack/echo/v4" 11 12 ) 12 13 13 14 func (s *Server) EventsLabelsWebsocket(c echo.Context) error { 14 14 - did := c.Request().Header.Get("DID") 15 15 + var since *int64 16 16 + if sinceVal := c.QueryParam("cursor"); sinceVal != "" { 17 17 + sval, err := strconv.ParseInt(sinceVal, 10, 64) 18 18 + if err != nil { 19 19 + return err 20 20 + } 21 21 + since = &sval 22 22 + } 23 23 + 24 24 + ctx := c.Request().Context() 25 25 + 26 26 + // TODO: authhhh 15 27 conn, err := websocket.Upgrade(c.Response().Writer, c.Request(), c.Response().Header(), 1<<10, 1<<10) 16 28 if err != nil { 17 17 - return err 29 29 + return fmt.Errorf("upgrading websocket: %w", err) 18 30 } 19 31 20 20 - var peering pds.Peering 21 21 - if did != "" { 22 22 - if err := s.db.First(&peering, "did = ?", did).Error; err != nil { 23 23 - return err 24 24 - } 25 25 - } 26 26 - 27 27 - evts, cancel, err := s.levents.Subscribe(func(evt *events.LabelEvent) bool { 32 32 + evts, cancel, err := s.levents.Subscribe(ctx, func(evt *events.LabelStreamEvent) bool { 28 33 return true 29 29 - }, nil) 34 34 + }, since) 30 35 if err != nil { 31 36 return err 32 37 } 33 38 defer cancel() 34 39 35 35 - // TODO: Op: events.EvtKindLabel 36 36 - header := events.EventHeader{Op: events.EvtKindRepoAppend} 37 37 - for evt := range evts { 38 38 - wc, err := conn.NextWriter(websocket.BinaryMessage) 39 39 - if err != nil { 40 40 - return err 41 41 - } 40 40 + header := events.EventHeader{Op: events.LEvtKindLabelBatch} 41 41 + for { 42 42 + select { 43 43 + case evt := <-evts: 44 44 + wc, err := conn.NextWriter(websocket.BinaryMessage) 45 45 + if err != nil { 46 46 + return err 47 47 + } 48 48 + 49 49 + var obj lexutil.CBOR 50 50 + 51 51 + switch { 52 52 + case evt.Batch != nil: 53 53 + header.Op = events.LEvtKindLabelBatch 54 54 + obj = evt.Batch 55 55 + case evt.Error != nil: 56 56 + header.Op = events.LEvtKindErrorFrame 57 57 + obj = evt.Error 58 58 + case evt.Info != nil: 59 59 + header.Op = events.LEvtKindInfoFrame 60 60 + obj = evt.Info 61 61 + default: 62 62 + return fmt.Errorf("unrecognized event kind") 63 63 + } 42 64 43 43 - if err := header.MarshalCBOR(wc); err != nil { 44 44 - return fmt.Errorf("failed to write header: %w", err) 45 45 - } 65 65 + if err := header.MarshalCBOR(wc); err != nil { 66 66 + return fmt.Errorf("failed to write header: %w", err) 67 67 + } 46 68 47 47 - if err := evt.MarshalCBOR(wc); err != nil { 48 48 - return fmt.Errorf("failed to write event: %w", err) 49 49 - } 69 69 + if err := obj.MarshalCBOR(wc); err != nil { 70 70 + return fmt.Errorf("failed to write event: %w", err) 71 71 + } 50 72 51 51 - if err := wc.Close(); err != nil { 52 52 - return fmt.Errorf("failed to flush-close our event write: %w", err) 73 73 + if err := wc.Close(); err != nil { 74 74 + return fmt.Errorf("failed to flush-close our event write: %w", err) 75 75 + } 76 76 + case <-ctx.Done(): 77 77 + return nil 53 78 } 54 79 } 55 55 - 56 56 - return nil 57 80 }