relay splitter daemon + pebble persistence

cmd/rainbow/main.go

@@ -52,6 +52,11 @@ func run(args []string) {
 			Name:  "splitter-host",
 			Value: "bsky.network",
 		},
+		&cli.StringFlag{
+			Name:  "persist-db",
+			Value: "",
+			Usage: "path to persistence db",
+		},
 		&cli.StringFlag{
 			Name:  "api-listen",
 			Value: ":2480",
@@ -110,7 +115,21 @@ func Splitter(cctx *cli.Context) error {
 		otel.SetTracerProvider(tp)
 	}
 
-	spl := splitter.NewSplitter(cctx.String("splitter-host"))
+	persistPath := cctx.String("persist-db")
+	upstreamHost := cctx.String("splitter-host")
+	var spl *splitter.Splitter
+	var err error
+	if persistPath != "" {
+		log.Infof("building splitter with storage at: %s", persistPath)
+		spl, err = splitter.NewDiskSplitter(upstreamHost, persistPath)
+		if err != nil {
+			log.Fatalw("failed to create splitter", "path", persistPath, "error", err)
+			return err
+		}
+	} else {
+		log.Info("building in-memory splitter")
+		spl = splitter.NewMemSplitter(upstreamHost)
+	}
 
 	// set up metrics endpoint
 	go func() {

events/dbpersist_test.go

@@ -1,4 +1,4 @@
-package events_test
+package events
 
 import (
 	"context"
@@ -11,19 +11,18 @@ import (
 	atproto "github.com/bluesky-social/indigo/api/atproto"
 	"github.com/bluesky-social/indigo/api/bsky"
 	"github.com/bluesky-social/indigo/carstore"
-	"github.com/bluesky-social/indigo/events"
 	lexutil "github.com/bluesky-social/indigo/lex/util"
 	"github.com/bluesky-social/indigo/models"
-	"github.com/bluesky-social/indigo/pds"
+	pds "github.com/bluesky-social/indigo/pds/data"
 	"github.com/bluesky-social/indigo/repomgr"
 	"github.com/bluesky-social/indigo/util"
-	"github.com/ipfs/go-log/v2"
+	logging "github.com/ipfs/go-log/v2"
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
 )
 
 func init() {
-	log.SetAllLoggers(log.LevelDebug)
+	logging.SetAllLoggers(logging.LevelDebug)
 }
 
 func BenchmarkDBPersist(b *testing.B) {
@@ -61,24 +60,24 @@ func BenchmarkDBPersist(b *testing.B) {
 	defer os.RemoveAll(tempPath)
 
 	// Initialize a DBPersister
-	dbp, err := events.NewDbPersistence(db, cs, nil)
+	dbp, err := NewDbPersistence(db, cs, nil)
 	if err != nil {
 		b.Fatal(err)
 	}
 
 	// Create a bunch of events
-	evtman := events.NewEventManager(dbp)
+	evtman := NewEventManager(dbp)
 
 	userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
 	if err != nil {
 		b.Fatal(err)
 	}
 
-	inEvts := make([]*events.XRPCStreamEvent, b.N)
+	inEvts := make([]*XRPCStreamEvent, b.N)
 	for i := 0; i < b.N; i++ {
 		cidLink := lexutil.LexLink(cid)
 		headLink := lexutil.LexLink(userRepoHead)
-		inEvts[i] = &events.XRPCStreamEvent{
+		inEvts[i] = &XRPCStreamEvent{
 			RepoCommit: &atproto.SyncSubscribeRepos_Commit{
 				Repo:   "did:example:123",
 				Commit: headLink,
@@ -136,7 +135,7 @@ func BenchmarkDBPersist(b *testing.B) {
 
 	b.StopTimer()
 
-	dbp.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
+	dbp.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
 		outEvtCount++
 		return nil
 	})
@@ -183,24 +182,24 @@ func BenchmarkPlayback(b *testing.B) {
 	defer os.RemoveAll(tempPath)
 
 	// Initialize a DBPersister
-	dbp, err := events.NewDbPersistence(db, cs, nil)
+	dbp, err := NewDbPersistence(db, cs, nil)
 	if err != nil {
 		b.Fatal(err)
 	}
 
 	// Create a bunch of events
-	evtman := events.NewEventManager(dbp)
+	evtman := NewEventManager(dbp)
 
 	userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
 	if err != nil {
 		b.Fatal(err)
 	}
 
-	inEvts := make([]*events.XRPCStreamEvent, n)
+	inEvts := make([]*XRPCStreamEvent, n)
 	for i := 0; i < n; i++ {
 		cidLink := lexutil.LexLink(cid)
 		headLink := lexutil.LexLink(userRepoHead)
-		inEvts[i] = &events.XRPCStreamEvent{
+		inEvts[i] = &XRPCStreamEvent{
 			RepoCommit: &atproto.SyncSubscribeRepos_Commit{
 				Repo:   "did:example:123",
 				Commit: headLink,
@@ -256,7 +255,7 @@ func BenchmarkPlayback(b *testing.B) {
 
 	b.ResetTimer()
 
-	dbp.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
+	dbp.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
 		outEvtCount++
 		return nil
 	})