fix events tests add pebble test

events/dbpersist_test.go

···
-package events_test
+package events
 
 import (
 	"context"
···
 	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) {
···
 	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,
···
 
 	b.StopTimer()
 
-	dbp.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
+	dbp.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
 		outEvtCount++
 		return nil
 	})
···
 	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,
···
 
 	b.ResetTimer()
 
-	dbp.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
+	dbp.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
 		outEvtCount++
 		return nil
 	})

events/diskpersist_test.go

···
-package events_test
+package events
 
 import (
 	"context"
···
 	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"
 	"gorm.io/gorm"
 )
 
+func testPersister(t *testing.T, perisistenceFactory func(path string, db *gorm.DB) (EventPersistence, error)) {
+	ctx := context.Background()
+
+	db, _, cs, tempPath, err := setupDBs(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	db.AutoMigrate(&pds.User{})
+	db.AutoMigrate(&pds.Peering{})
+	db.AutoMigrate(&models.ActorInfo{})
+
+	db.Create(&models.ActorInfo{
+		Uid: 1,
+		Did: "did:example:123",
+	})
+
+	mgr := repomgr.NewRepoManager(cs, &util.FakeKeyManager{})
+
+	err = mgr.InitNewActor(ctx, 1, "alice", "did:example:123", "Alice", "", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, cid, err := mgr.CreateRecord(ctx, 1, "app.bsky.feed.post", &bsky.FeedPost{
+		Text:      "hello world",
+		CreatedAt: time.Now().Format(util.ISO8601),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer os.RemoveAll(tempPath)
+
+	// Initialize a persister
+	dp, err := perisistenceFactory(tempPath, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create a bunch of events
+	evtman := NewEventManager(dp)
+
+	userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	n := 100
+	inEvts := make([]*XRPCStreamEvent, n)
+	for i := 0; i < n; i++ {
+		cidLink := lexutil.LexLink(cid)
+		headLink := lexutil.LexLink(userRepoHead)
+		inEvts[i] = &XRPCStreamEvent{
+			RepoCommit: &atproto.SyncSubscribeRepos_Commit{
+				Repo:   "did:example:123",
+				Commit: headLink,
+				Ops: []*atproto.SyncSubscribeRepos_RepoOp{
+					{
+						Action: "add",
+						Cid:    &cidLink,
+						Path:   "path1",
+					},
+				},
+				Time: time.Now().Format(util.ISO8601),
+				Seq:  int64(i),
+			},
+		}
+	}
+
+	// Add events in parallel
+	for i := 0; i < n; i++ {
+		err = evtman.AddEvent(ctx, inEvts[i])
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if err := dp.Flush(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	outEvtCount := 0
+	expectedEvtCount := n
+
+	dp.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
+		outEvtCount++
+		return nil
+	})
+
+	if outEvtCount != expectedEvtCount {
+		t.Fatalf("expected %d events, got %d", expectedEvtCount, outEvtCount)
+	}
+
+	dp.Shutdown(ctx)
+
+	time.Sleep(time.Millisecond * 100)
+
+	dp2, err := NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &DiskPersistOptions{
+		EventsPerFile: 10,
+		UIDCacheSize:  100000,
+		DIDCacheSize:  100000,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	evtman2 := NewEventManager(dp2)
+
+	inEvts = make([]*XRPCStreamEvent, n)
+	for i := 0; i < n; i++ {
+		cidLink := lexutil.LexLink(cid)
+		headLink := lexutil.LexLink(userRepoHead)
+		inEvts[i] = &XRPCStreamEvent{
+			RepoCommit: &atproto.SyncSubscribeRepos_Commit{
+				Repo:   "did:example:123",
+				Commit: headLink,
+				Ops: []*atproto.SyncSubscribeRepos_RepoOp{
+					{
+						Action: "add",
+						Cid:    &cidLink,
+						Path:   "path1",
+					},
+				},
+				Time: time.Now().Format(util.ISO8601),
+			},
+		}
+	}
+
+	for i := 0; i < n; i++ {
+		err = evtman2.AddEvent(ctx, inEvts[i])
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
 func TestDiskPersist(t *testing.T) {
+	factory := func(tempPath string, db *gorm.DB) (EventPersistence, error) {
+		return NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &DiskPersistOptions{
+			EventsPerFile: 10,
+			UIDCacheSize:  100000,
+			DIDCacheSize:  100000,
+		})
+	}
+	testPersister(t, factory)
+}
+
+func XTestDiskPersist(t *testing.T) {
 	ctx := context.Background()
 
 	db, _, cs, tempPath, err := setupDBs(t)
···
 
 	// Initialize a DBPersister
 
-	dp, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
+	dp, err := NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &DiskPersistOptions{
 		EventsPerFile: 10,
 		UIDCacheSize:  100000,
 		DIDCacheSize:  100000,
···
 	}
 
 	// Create a bunch of events
-	evtman := events.NewEventManager(dp)
+	evtman := NewEventManager(dp)
 
 	userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
 	if err != nil {
···
 	}
 
 	n := 100
-	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,
···
 	outEvtCount := 0
 	expectedEvtCount := n
 
-	dp.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
+	dp.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
 		outEvtCount++
 		return nil
 	})
···
 
 	time.Sleep(time.Millisecond * 100)
 
-	dp2, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
+	dp2, err := NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &DiskPersistOptions{
 		EventsPerFile: 10,
 		UIDCacheSize:  100000,
 		DIDCacheSize:  100000,
···
 		t.Fatal(err)
 	}
 
-	evtman2 := events.NewEventManager(dp2)
+	evtman2 := NewEventManager(dp2)
 
-	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,
···
 
 	// Initialize a DBPersister
 
-	dp, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
+	dp, err := NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &DiskPersistOptions{
 		EventsPerFile: 5000,
 		UIDCacheSize:  100000,
 		DIDCacheSize:  100000,
···
 
 }
 
-func runPersisterBenchmark(b *testing.B, cs carstore.CarStore, db *gorm.DB, p events.EventPersistence) {
+func runPersisterBenchmark(b *testing.B, cs carstore.CarStore, db *gorm.DB, p EventPersistence) {
 	ctx := context.Background()
 
 	db.AutoMigrate(&pds.User{})
···
 	}
 
 	// Create a bunch of events
-	evtman := events.NewEventManager(p)
+	evtman := NewEventManager(p)
 
 	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,
···
 
 	// Initialize a DBPersister
 
-	dp, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
+	dp, err := NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &DiskPersistOptions{
 		EventsPerFile: 20,
 		UIDCacheSize:  100000,
 		DIDCacheSize:  100000,
···
 	runEventManagerTest(t, cs, db, dp)
 }
 
-func runEventManagerTest(t *testing.T, cs carstore.CarStore, db *gorm.DB, p events.EventPersistence) {
+func runEventManagerTest(t *testing.T, cs carstore.CarStore, db *gorm.DB, p EventPersistence) {
 	ctx := context.Background()
 
 	db.AutoMigrate(&pds.User{})
···
 		t.Fatal(err)
 	}
 
-	evtman := events.NewEventManager(p)
+	evtman := NewEventManager(p)
 
 	userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
 	if err != nil {
···
 	}
 
 	testSize := 100 // you can adjust this number as needed
-	inEvts := make([]*events.XRPCStreamEvent, testSize)
+	inEvts := make([]*XRPCStreamEvent, testSize)
 	for i := 0; i < testSize; 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,
···
 	}
 
 	outEvtCount := 0
-	p.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
+	p.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
 		// Check that the contents of the output events match the input events
 		// Clear cache, don't care if one has it and not the other
 		inEvts[outEvtCount].Preserialized = nil
···
 
 	// Initialize a DBPersister
 
-	dp, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
+	dp, err := NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &DiskPersistOptions{
 		EventsPerFile: 10,
 		UIDCacheSize:  100000,
 		DIDCacheSize:  100000,
···
 	runTakedownTest(t, cs, db, dp)
 }
 
-func runTakedownTest(t *testing.T, cs carstore.CarStore, db *gorm.DB, p events.EventPersistence) {
+func runTakedownTest(t *testing.T, cs carstore.CarStore, db *gorm.DB, p EventPersistence) {
 	ctx := context.TODO()
 
 	db.AutoMigrate(&pds.User{})
···
 		}
 	}
 
-	evtman := events.NewEventManager(p)
+	evtman := NewEventManager(p)
 
 	testSize := 100 // you can adjust this number as needed
-	inEvts := make([]*events.XRPCStreamEvent, testSize*userCount)
+	inEvts := make([]*XRPCStreamEvent, testSize*userCount)
 	for i := 0; i < testSize*userCount; i++ {
 		user := users[i%userCount]
 		_, cid, err := mgr.CreateRecord(ctx, user.Uid, "app.bsky.feed.post", &bsky.FeedPost{
···
 
 		cidLink := lexutil.LexLink(cid)
 		headLink := lexutil.LexLink(userRepoHead)
-		inEvts[i] = &events.XRPCStreamEvent{
+		inEvts[i] = &XRPCStreamEvent{
 			RepoCommit: &atproto.SyncSubscribeRepos_Commit{
 				Repo:   user.Did,
 				Commit: headLink,
···
 
 	// Verify that the events of the user have been removed from the event stream
 	var evtsCount int
-	if err := p.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
+	if err := p.Playback(ctx, 0, func(evt *XRPCStreamEvent) error {
 		evtsCount++
 		if evt.RepoCommit.Repo == takeDownUser.Did {
 			t.Fatalf("found event for user %d after takedown", takeDownUser.Uid)

events/pebblepersist.go

···
 type PebblePersist struct {
 	broadcast func(*XRPCStreamEvent)
 	db        *pebble.DB
+
+	prevSeq      int64
+	prevSeqExtra uint32
 }
 
 func NewPebblePersistance(path string) (*PebblePersist, error) {
···
 	blob := e.Preserialized
 
 	seq := e.Sequence()
+	log.Infof("persist %d", seq)
+
 	if seq < 0 {
-		// drop event
-		// TODO: persist with longer key? {prev 8 byte key}{int32 extra counter}
-		return nil
-	}
+		// persist with longer key {prev 8 byte key}{int32 extra counter}
+		pp.prevSeqExtra++
+		var key [12]byte
+		binary.BigEndian.PutUint64(key[:8], uint64(pp.prevSeq))
+		binary.BigEndian.PutUint32(key[8:], pp.prevSeqExtra)
 
-	var key [8]byte
-	binary.BigEndian.PutUint64(key[:], uint64(seq))
+		err = pp.db.Set(key[:], blob, pebble.Sync)
+		return nil
+	} else {
+		pp.prevSeq = seq
+		pp.prevSeqExtra = 0
+		var key [8]byte
+		binary.BigEndian.PutUint64(key[:], uint64(seq))
 
-	err = pp.db.Set(key[:], blob, pebble.Sync)
+		err = pp.db.Set(key[:], blob, pebble.Sync)
+	}
 
 	return err
 }

events/pebblepersist_test.go

···
+package events
+
+import (
+	"gorm.io/gorm"
+	"path/filepath"
+	"testing"
+)
+
+func TestPebblePersist(t *testing.T) {
+	factory := func(tempPath string, db *gorm.DB) (EventPersistence, error) {
+		return NewPebblePersistance(filepath.Join(tempPath, "pebble.db"))
+	}
+	testPersister(t, factory)
+}

pds/data/types.go

···
+package data
+
+import (
+	"github.com/bluesky-social/indigo/models"
+	"gorm.io/gorm"
+	"time"
+)
+
+type User struct {
+	ID          models.Uid `gorm:"primarykey"`
+	CreatedAt   time.Time
+	UpdatedAt   time.Time
+	DeletedAt   gorm.DeletedAt `gorm:"index"`
+	Handle      string         `gorm:"uniqueIndex"`
+	Password    string
+	RecoveryKey string
+	Email       string
+	Did         string `gorm:"uniqueIndex"`
+	PDS         uint
+}
+
+type Peering struct {
+	gorm.Model
+	Host     string
+	Did      string
+	Approved bool
+}

pds/server.go

···
 	lexutil "github.com/bluesky-social/indigo/lex/util"
 	"github.com/bluesky-social/indigo/models"
 	"github.com/bluesky-social/indigo/notifs"
+	pdsdata "github.com/bluesky-social/indigo/pds/data"
 	"github.com/bluesky-social/indigo/plc"
 	"github.com/bluesky-social/indigo/repomgr"
 	"github.com/bluesky-social/indigo/util"
···
 	return c.String(200, u.Did)
 }
 
-type User struct {
-	ID          models.Uid `gorm:"primarykey"`
-	CreatedAt   time.Time
-	UpdatedAt   time.Time
-	DeletedAt   gorm.DeletedAt `gorm:"index"`
-	Handle      string         `gorm:"uniqueIndex"`
-	Password    string
-	RecoveryKey string
-	Email       string
-	Did         string `gorm:"uniqueIndex"`
-	PDS         uint
-}
+type User = pdsdata.User
 
 type RefreshToken struct {
 	gorm.Model
···
 	panic("nyi")
 }
 
-type Peering struct {
-	gorm.Model
-	Host     string
-	Did      string
-	Approved bool
-}
+type Peering = pdsdata.Peering
 
 func (s *Server) EventsHandler(c echo.Context) error {
 	conn, err := websocket.Upgrade(c.Response().Writer, c.Request(), c.Response().Header(), 1<<10, 1<<10)