diff --git a/cmd/cocoon/main.go b/cmd/cocoon/main.go index 4944f45..ede07c0 100644 --- a/cmd/cocoon/main.go +++ b/cmd/cocoon/main.go @@ -136,6 +136,11 @@ func main() { EnvVars: []string{"COCOON_DEFAULT_ATPROTO_PROXY"}, Value: "did:web:api.bsky.app#bsky_appview", }, + &cli.StringFlag{ + Name: "blockstore-variant", + EnvVars: []string{"COCOON_BLOCKSTORE_VARIANT"}, + Value: "sqlite", + }, }, Commands: []*cli.Command{ runServe, @@ -158,6 +163,7 @@ var runServe = &cli.Command{ Usage: "Start the cocoon PDS", Flags: []cli.Flag{}, Action: func(cmd *cli.Context) error { + s, err := server.New(&server.Args{ Addr: cmd.String("addr"), DbName: cmd.String("db-name"), @@ -185,6 +191,7 @@ var runServe = &cli.Command{ }, SessionSecret: cmd.String("session-secret"), DefaultAtprotoProxy: cmd.String("default-atproto-proxy"), + BlockstoreVariant: server.MustReturnBlockstoreVariant(cmd.String("blockstore-variant")), }) if err != nil { fmt.Printf("error creating cocoon: %v", err) 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/blockstore_variant.go b/server/blockstore_variant.go new file mode 100644 index 0000000..874f512 --- /dev/null +++ b/server/blockstore_variant.go @@ -0,0 +1,30 @@ +package server + +import ( + "github.com/haileyok/cocoon/sqlite_blockstore" + blockstore "github.com/ipfs/go-ipfs-blockstore" +) + +type BlockstoreVariant int + +const ( + BlockstoreVariantSqlite = iota +) + +func MustReturnBlockstoreVariant(maybeBsv string) BlockstoreVariant { + switch maybeBsv { + case "sqlite": + return BlockstoreVariantSqlite + default: + panic("invalid blockstore variant provided") + } +} + +func (s *Server) getBlockstore(did string) blockstore.Blockstore { + switch s.config.BlockstoreVariant { + case BlockstoreVariantSqlite: + return sqlite_blockstore.New(did, s.db) + default: + return sqlite_blockstore.New(did, s.db) + } +} diff --git a/server/handle_import_repo.go b/server/handle_import_repo.go index 2104c88..b54aa68 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.getBlockstore(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..27c7b9a 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.getBlockstore(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..2506c19 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.getBlockstore(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..c04469f 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.getBlockstore(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.getBlockstore(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..05cf19d 100644 --- a/server/server.go +++ b/server/server.go @@ -38,6 +38,7 @@ import ( "github.com/haileyok/cocoon/oauth/dpop" "github.com/haileyok/cocoon/oauth/provider" "github.com/haileyok/cocoon/plc" + "github.com/ipfs/go-cid" echo_session "github.com/labstack/echo-contrib/session" "github.com/labstack/echo/v4" "github.com/labstack/echo/v4/middleware" @@ -104,6 +105,8 @@ type Args struct { SessionSecret string DefaultAtprotoProxy string + + BlockstoreVariant BlockstoreVariant } type config struct { @@ -117,6 +120,7 @@ type config struct { SmtpEmail string SmtpName string DefaultAtprotoProxy string + BlockstoreVariant BlockstoreVariant } type CustomValidator struct { @@ -349,6 +353,7 @@ func New(args *Args) (*Server, error) { SmtpName: args.SmtpName, SmtpEmail: args.SmtpEmail, DefaultAtprotoProxy: args.DefaultAtprotoProxy, + BlockstoreVariant: args.BlockstoreVariant, }, evtman: events.NewEventManager(events.NewMemPersister()), passport: identity.NewPassport(h, identity.NewMemCache(10_000)), @@ -641,3 +646,11 @@ func (s *Server) backupRoutine() { go s.doBackup() } } + +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")