diff --git a/recording_blockstore/recording_blockstore.go b/recording_blockstore/recording_blockstore.go new file mode 100644 index 0000000..312fccc --- /dev/null +++ b/recording_blockstore/recording_blockstore.go @@ -0,0 +1,77 @@ +package recording_blockstore + +import ( + "context" + + blockformat "github.com/ipfs/go-block-format" + "github.com/ipfs/go-cid" + blockstore "github.com/ipfs/go-ipfs-blockstore" +) + +type RecordingBlockstore struct { + base blockstore.Blockstore + + inserts map[cid.Cid]blockformat.Block +} + +func New(base blockstore.Blockstore) *RecordingBlockstore { + return &RecordingBlockstore{ + base: base, + inserts: make(map[cid.Cid]blockformat.Block), + } +} + +func (bs *RecordingBlockstore) Has(ctx context.Context, c cid.Cid) (bool, error) { + return bs.base.Has(ctx, c) +} + +func (bs *RecordingBlockstore) Get(ctx context.Context, c cid.Cid) (blockformat.Block, error) { + return bs.base.Get(ctx, c) +} + +func (bs *RecordingBlockstore) GetSize(ctx context.Context, c cid.Cid) (int, error) { + return bs.base.GetSize(ctx, c) +} + +func (bs *RecordingBlockstore) DeleteBlock(ctx context.Context, c cid.Cid) error { + return bs.base.DeleteBlock(ctx, c) +} + +func (bs *RecordingBlockstore) Put(ctx context.Context, block blockformat.Block) error { + if err := bs.base.Put(ctx, block); err != nil { + return err + } + bs.inserts[block.Cid()] = block + return nil +} + +func (bs *RecordingBlockstore) PutMany(ctx context.Context, blocks []blockformat.Block) error { + if err := bs.base.PutMany(ctx, blocks); err != nil { + return err + } + + for _, b := range blocks { + bs.inserts[b.Cid()] = b + } + + return nil +} + +func (bs *RecordingBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { + return bs.AllKeysChan(ctx) +} + +func (bs *RecordingBlockstore) HashOnRead(enabled bool) { +} + +func (bs *RecordingBlockstore) GetLogMap() map[cid.Cid]blockformat.Block { + return bs.inserts +} + +func (bs *RecordingBlockstore) GetLogArray() []blockformat.Block { + var blocks []blockformat.Block + for _, b := range bs.inserts { + blocks = append(blocks, b) + } + return blocks +} diff --git a/server/handle_import_repo.go b/server/handle_import_repo.go index 2104c88..42c0503 100644 --- a/server/handle_import_repo.go +++ b/server/handle_import_repo.go @@ -9,7 +9,6 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/repo" - "github.com/haileyok/cocoon/blockstore" "github.com/haileyok/cocoon/internal/helpers" "github.com/haileyok/cocoon/models" blocks "github.com/ipfs/go-block-format" @@ -27,7 +26,7 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { return helpers.ServerError(e, nil) } - bs := blockstore.New(urepo.Repo.Did, s.db) + bs := s.createBlockstore(urepo.Repo.Did) cs, err := car.NewCarReader(bytes.NewReader(b)) if err != nil { @@ -107,7 +106,7 @@ func (s *Server) handleRepoImportRepo(e echo.Context) error { return helpers.ServerError(e, nil) } - if err := bs.UpdateRepo(context.TODO(), root, rev); err != nil { + if err := s.UpdateRepo(context.TODO(), urepo.Repo.Did, root, rev); err != nil { s.logger.Error("error updating repo after commit", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_server_create_account.go b/server/handle_server_create_account.go index 47bcaee..87da0f3 100644 --- a/server/handle_server_create_account.go +++ b/server/handle_server_create_account.go @@ -14,7 +14,6 @@ import ( "github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/repo" "github.com/bluesky-social/indigo/util" - "github.com/haileyok/cocoon/blockstore" "github.com/haileyok/cocoon/internal/helpers" "github.com/haileyok/cocoon/models" "github.com/labstack/echo/v4" @@ -177,7 +176,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { } if customDidHeader == "" { - bs := blockstore.New(signupDid, s.db) + bs := s.createBlockstore(signupDid) r := repo.NewRepo(context.TODO(), signupDid, bs) root, rev, err := r.Commit(context.TODO(), urepo.SignFor) @@ -186,7 +185,7 @@ func (s *Server) handleCreateAccount(e echo.Context) error { return helpers.ServerError(e, nil) } - if err := bs.UpdateRepo(context.TODO(), root, rev); err != nil { + if err := s.UpdateRepo(context.TODO(), urepo.Did, root, rev); err != nil { s.logger.Error("error updating repo after commit", "error", err) return helpers.ServerError(e, nil) } diff --git a/server/handle_sync_get_blocks.go b/server/handle_sync_get_blocks.go index 2f6c90b..83abfaa 100644 --- a/server/handle_sync_get_blocks.go +++ b/server/handle_sync_get_blocks.go @@ -6,7 +6,6 @@ import ( "strings" "github.com/bluesky-social/indigo/carstore" - "github.com/haileyok/cocoon/blockstore" "github.com/haileyok/cocoon/internal/helpers" "github.com/ipfs/go-cid" cbor "github.com/ipfs/go-ipld-cbor" @@ -54,7 +53,7 @@ func (s *Server) handleGetBlocks(e echo.Context) error { return helpers.ServerError(e, nil) } - bs := blockstore.New(urepo.Repo.Did, s.db) + bs := s.createBlockstore(urepo.Repo.Did) for _, c := range cids { b, err := bs.Get(context.TODO(), c) diff --git a/server/repo.go b/server/repo.go index 7e061ee..8f38b55 100644 --- a/server/repo.go +++ b/server/repo.go @@ -16,10 +16,9 @@ import ( "github.com/bluesky-social/indigo/events" lexutil "github.com/bluesky-social/indigo/lex/util" "github.com/bluesky-social/indigo/repo" - "github.com/bluesky-social/indigo/util" - "github.com/haileyok/cocoon/blockstore" "github.com/haileyok/cocoon/internal/db" "github.com/haileyok/cocoon/models" + "github.com/haileyok/cocoon/recording_blockstore" blocks "github.com/ipfs/go-block-format" "github.com/ipfs/go-cid" cbor "github.com/ipfs/go-ipld-cbor" @@ -103,7 +102,8 @@ func (rm *RepoMan) applyWrites(urepo models.Repo, writes []Op, swapCommit *strin return nil, err } - dbs := blockstore.New(urepo.Did, rm.db) + dbs := rm.s.createBlockstore(urepo.Did) + bs := recording_blockstore.New(dbs) r, err := repo.OpenRepo(context.TODO(), dbs, rootcid) entries := []models.Record{} @@ -274,7 +274,7 @@ func (rm *RepoMan) applyWrites(urepo models.Repo, writes []Op, swapCommit *strin } } - for _, op := range dbs.GetLog() { + for _, op := range bs.GetLogMap() { if _, err := carstore.LdWrite(buf, op.Cid().Bytes(), op.RawData()); err != nil { return nil, err } @@ -324,7 +324,7 @@ func (rm *RepoMan) applyWrites(urepo models.Repo, writes []Op, swapCommit *strin }, }) - if err := dbs.UpdateRepo(context.TODO(), newroot, rev); err != nil { + if err := rm.s.UpdateRepo(context.TODO(), urepo.Did, newroot, rev); err != nil { return nil, err } @@ -345,8 +345,8 @@ func (rm *RepoMan) getRecordProof(urepo models.Repo, collection, rkey string) (c return cid.Undef, nil, err } - dbs := blockstore.New(urepo.Did, rm.db) - bs := util.NewLoggingBstore(dbs) + dbs := rm.s.createBlockstore(urepo.Did) + bs := recording_blockstore.New(dbs) r, err := repo.OpenRepo(context.TODO(), bs, c) if err != nil { @@ -358,7 +358,7 @@ func (rm *RepoMan) getRecordProof(urepo models.Repo, collection, rkey string) (c return cid.Undef, nil, err } - return c, bs.GetLoggedBlocks(), nil + return c, bs.GetLogArray(), nil } func (rm *RepoMan) incrementBlobRefs(urepo models.Repo, cbor []byte) ([]cid.Cid, error) { @@ -414,10 +414,10 @@ func getBlobCidsFromCbor(cbor []byte) ([]cid.Cid, error) { return nil, fmt.Errorf("error unmarshaling cbor: %w", err) } - var deepiter func(interface{}) error - deepiter = func(item interface{}) error { + var deepiter func(any) error + deepiter = func(item any) error { switch val := item.(type) { - case map[string]interface{}: + case map[string]any: if val["$type"] == "blob" { if ref, ok := val["ref"].(string); ok { c, err := cid.Parse(ref) @@ -430,7 +430,7 @@ func getBlobCidsFromCbor(cbor []byte) ([]cid.Cid, error) { return deepiter(v) } } - case []interface{}: + case []any: for _, v := range val { deepiter(v) } diff --git a/server/server.go b/server/server.go index be1f739..b04148b 100644 --- a/server/server.go +++ b/server/server.go @@ -38,6 +38,9 @@ import ( "github.com/haileyok/cocoon/oauth/dpop" "github.com/haileyok/cocoon/oauth/provider" "github.com/haileyok/cocoon/plc" + "github.com/haileyok/cocoon/sqlite_blockstore" + "github.com/ipfs/go-cid" + blockstore "github.com/ipfs/go-ipfs-blockstore" echo_session "github.com/labstack/echo-contrib/session" "github.com/labstack/echo/v4" "github.com/labstack/echo/v4/middleware" @@ -641,3 +644,16 @@ func (s *Server) backupRoutine() { go s.doBackup() } } + +func (s *Server) createBlockstore(did string) blockstore.Blockstore { + // TODO: eventually configurable blockstore types here + return sqlite_blockstore.New(did, s.db) +} + +func (s *Server) UpdateRepo(ctx context.Context, did string, root cid.Cid, rev string) error { + if err := s.db.Exec("UPDATE repos SET root = ?, rev = ? WHERE did = ?", nil, root.Bytes(), rev, did).Error; err != nil { + return err + } + + return nil +} diff --git a/blockstore/blockstore.go b/sqlite_blockstore/sqlite_blockstore.go similarity index 92% rename from blockstore/blockstore.go rename to sqlite_blockstore/sqlite_blockstore.go index 2dee9ff..982836d 100644 --- a/blockstore/blockstore.go +++ b/sqlite_blockstore/sqlite_blockstore.go @@ -1,4 +1,4 @@ -package blockstore +package sqlite_blockstore import ( "context" @@ -136,14 +136,6 @@ func (bs *SqliteBlockstore) HashOnRead(enabled bool) { panic("not implemented") } -func (bs *SqliteBlockstore) UpdateRepo(ctx context.Context, root cid.Cid, rev string) error { - if err := bs.db.Exec("UPDATE repos SET root = ?, rev = ? WHERE did = ?", nil, root.Bytes(), rev, bs.did).Error; err != nil { - return err - } - - return nil -} - func (bs *SqliteBlockstore) Execute(ctx context.Context) error { if !bs.readonly { return fmt.Errorf("blockstore was not readonly")