WIP: allow event persister to broadcast events as it sees fit to allo…

…w future optimizations

events/dbpersist.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sync"
 	"time"
 
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
@@ -21,6 +22,10 @@ type DbPersistence struct {
 	db *gorm.DB
 
 	cs *carstore.CarStore
+
+	lk sync.Mutex
+
+	broadcast func(*XRPCStreamEvent)
 }
 
 type RepoEventRecord struct {
@@ -48,6 +53,10 @@ func NewDbPersistence(db *gorm.DB, cs *carstore.CarStore) (*DbPersistence, error
 	}, nil
 }
 
+func (p *DbPersistence) SetEventBroadcaster(brc func(*XRPCStreamEvent)) {
+	p.broadcast = brc
+}
+
 func (p *DbPersistence) Persist(ctx context.Context, e *XRPCStreamEvent) error {
 	if e.RepoCommit == nil {
 		return nil
@@ -101,12 +110,16 @@ func (p *DbPersistence) Persist(ctx context.Context, e *XRPCStreamEvent) error {
 	}
 	rer.Ops = opsb
 
+	p.lk.Lock()
+	defer p.lk.Unlock()
 	if err := p.db.Create(&rer).Error; err != nil {
 		return err
 	}
 
 	e.RepoCommit.Seq = int64(rer.Seq)
 
+	p.broadcast(e)
+
 	return nil
 }
 

events/events.go

@@ -26,10 +26,14 @@ type EventManager struct {
 }
 
 func NewEventManager(persister EventPersistence) *EventManager {
-	return &EventManager{
+	em := &EventManager{
 		bufferSize: 1024,
 		persister:  persister,
 	}
+
+	persister.SetEventBroadcaster(em.broadcastEvent)
+
+	return em
 }
 
 const (
@@ -45,7 +49,8 @@ type Operation struct {
 }
 
 func (em *EventManager) broadcastEvent(evt *XRPCStreamEvent) {
-	// NOTE: Assumes subsLk is held
+	em.subsLk.Lock()
+	defer em.subsLk.Unlock()
 
 	// TODO: for a larger fanout we should probably have dedicated goroutines
 	// for subsets of the subscriber set, and tiered channels to distribute
@@ -68,14 +73,9 @@ func (em *EventManager) broadcastEvent(evt *XRPCStreamEvent) {
 }
 
 func (em *EventManager) persistAndSendEvent(ctx context.Context, evt *XRPCStreamEvent) {
-	em.subsLk.Lock()
-	defer em.subsLk.Unlock()
-
 	if err := em.persister.Persist(context.TODO(), evt); err != nil {
 		log.Errorf("failed to persist outbound event: %s", err)
 	}
-
-	em.broadcastEvent(evt)
 }
 
 type Subscriber struct {

events/persist.go

@@ -14,6 +14,8 @@ type EventPersistence interface {
 	Playback(ctx context.Context, since int64, cb func(*XRPCStreamEvent) error) error
 	TakeDownRepo(ctx context.Context, usr util.Uid) error
 	RebaseRepoEvents(ctx context.Context, usr util.Uid) error
+
+	SetEventBroadcaster(func(*XRPCStreamEvent))
 }
 
 // MemPersister is the most naive implementation of event persistence
@@ -23,6 +25,8 @@ type MemPersister struct {
 	buf []*XRPCStreamEvent
 	lk  sync.Mutex
 	seq int64
+
+	broadcast func(*XRPCStreamEvent)
 }
 
 func NewMemPersister() *MemPersister {
@@ -49,6 +53,8 @@ func (mp *MemPersister) Persist(ctx context.Context, e *XRPCStreamEvent) error {
 	}
 	mp.buf = append(mp.buf, e)
 
+	mp.broadcast(e)
+
 	return nil
 }
 
@@ -78,3 +84,7 @@ func (mp *MemPersister) TakeDownRepo(ctx context.Context, uid util.Uid) error {
 func (mp *MemPersister) RebaseRepoEvents(ctx context.Context, usr util.Uid) error {
 	return fmt.Errorf("repo rebases not currently supported by memory persister, test usage only")
 }
+
+func (mp *MemPersister) SetEventBroadcaster(brc func(*XRPCStreamEvent)) {
+	mp.broadcast = brc
+}