-163

blockstore/blockstore.go

··· 1 1 - package blockstore 2 2 - 3 3 - import ( 4 4 - "context" 5 5 - "fmt" 6 6 - 7 7 - "github.com/bluesky-social/indigo/atproto/syntax" 8 8 - "github.com/haileyok/cocoon/internal/db" 9 9 - "github.com/haileyok/cocoon/models" 10 10 - blocks "github.com/ipfs/go-block-format" 11 11 - "github.com/ipfs/go-cid" 12 12 - "gorm.io/gorm/clause" 13 13 - ) 14 14 - 15 15 - type SqliteBlockstore struct { 16 16 - db *db.DB 17 17 - did string 18 18 - readonly bool 19 19 - inserts map[cid.Cid]blocks.Block 20 20 - } 21 21 - 22 22 - func New(did string, db *db.DB) *SqliteBlockstore { 23 23 - return &SqliteBlockstore{ 24 24 - did: did, 25 25 - db: db, 26 26 - readonly: false, 27 27 - inserts: map[cid.Cid]blocks.Block{}, 28 28 - } 29 29 - } 30 30 - 31 31 - func NewReadOnly(did string, db *db.DB) *SqliteBlockstore { 32 32 - return &SqliteBlockstore{ 33 33 - did: did, 34 34 - db: db, 35 35 - readonly: true, 36 36 - inserts: map[cid.Cid]blocks.Block{}, 37 37 - } 38 38 - } 39 39 - 40 40 - func (bs *SqliteBlockstore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, error) { 41 41 - var block models.Block 42 42 - 43 43 - maybeBlock, ok := bs.inserts[cid] 44 44 - if ok { 45 45 - return maybeBlock, nil 46 46 - } 47 47 - 48 48 - if err := bs.db.Raw("SELECT * FROM blocks WHERE did = ? AND cid = ?", nil, bs.did, cid.Bytes()).Scan(&block).Error; err != nil { 49 49 - return nil, err 50 50 - } 51 51 - 52 52 - b, err := blocks.NewBlockWithCid(block.Value, cid) 53 53 - if err != nil { 54 54 - return nil, err 55 55 - } 56 56 - 57 57 - return b, nil 58 58 - } 59 59 - 60 60 - func (bs *SqliteBlockstore) Put(ctx context.Context, block blocks.Block) error { 61 61 - bs.inserts[block.Cid()] = block 62 62 - 63 63 - if bs.readonly { 64 64 - return nil 65 65 - } 66 66 - 67 67 - b := models.Block{ 68 68 - Did: bs.did, 69 69 - Cid: block.Cid().Bytes(), 70 70 - Rev: syntax.NewTIDNow(0).String(), // TODO: WARN, this is bad. don't do this 71 71 - Value: block.RawData(), 72 72 - } 73 73 - 74 74 - if err := bs.db.Create(&b, []clause.Expression{clause.OnConflict{ 75 75 - Columns: []clause.Column{{Name: "did"}, {Name: "cid"}}, 76 76 - UpdateAll: true, 77 77 - }}).Error; err != nil { 78 78 - return err 79 79 - } 80 80 - 81 81 - return nil 82 82 - } 83 83 - 84 84 - func (bs *SqliteBlockstore) DeleteBlock(context.Context, cid.Cid) error { 85 85 - panic("not implemented") 86 86 - } 87 87 - 88 88 - func (bs *SqliteBlockstore) Has(context.Context, cid.Cid) (bool, error) { 89 89 - panic("not implemented") 90 90 - } 91 91 - 92 92 - func (bs *SqliteBlockstore) GetSize(context.Context, cid.Cid) (int, error) { 93 93 - panic("not implemented") 94 94 - } 95 95 - 96 96 - func (bs *SqliteBlockstore) PutMany(ctx context.Context, blocks []blocks.Block) error { 97 97 - tx := bs.db.BeginDangerously() 98 98 - 99 99 - for _, block := range blocks { 100 100 - bs.inserts[block.Cid()] = block 101 101 - 102 102 - if bs.readonly { 103 103 - continue 104 104 - } 105 105 - 106 106 - b := models.Block{ 107 107 - Did: bs.did, 108 108 - Cid: block.Cid().Bytes(), 109 109 - Rev: syntax.NewTIDNow(0).String(), // TODO: WARN, this is bad. don't do this 110 110 - Value: block.RawData(), 111 111 - } 112 112 - 113 113 - if err := tx.Clauses(clause.OnConflict{ 114 114 - Columns: []clause.Column{{Name: "did"}, {Name: "cid"}}, 115 115 - UpdateAll: true, 116 116 - }).Create(&b).Error; err != nil { 117 117 - tx.Rollback() 118 118 - return err 119 119 - } 120 120 - } 121 121 - 122 122 - if bs.readonly { 123 123 - return nil 124 124 - } 125 125 - 126 126 - tx.Commit() 127 127 - 128 128 - return nil 129 129 - } 130 130 - 131 131 - func (bs *SqliteBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { 132 132 - panic("not implemented") 133 133 - } 134 134 - 135 135 - func (bs *SqliteBlockstore) HashOnRead(enabled bool) { 136 136 - panic("not implemented") 137 137 - } 138 138 - 139 139 - func (bs *SqliteBlockstore) UpdateRepo(ctx context.Context, root cid.Cid, rev string) error { 140 140 - if err := bs.db.Exec("UPDATE repos SET root = ?, rev = ? WHERE did = ?", nil, root.Bytes(), rev, bs.did).Error; err != nil { 141 141 - return err 142 142 - } 143 143 - 144 144 - return nil 145 145 - } 146 146 - 147 147 - func (bs *SqliteBlockstore) Execute(ctx context.Context) error { 148 148 - if !bs.readonly { 149 149 - return fmt.Errorf("blockstore was not readonly") 150 150 - } 151 151 - 152 152 - bs.readonly = false 153 153 - for _, b := range bs.inserts { 154 154 - bs.Put(ctx, b) 155 155 - } 156 156 - bs.readonly = true 157 157 - 158 158 - return nil 159 159 - } 160 160 - 161 161 - func (bs *SqliteBlockstore) GetLog() map[cid.Cid]blocks.Block { 162 162 - return bs.inserts 163 163 - }

+7

cmd/cocoon/main.go

··· 136 136 EnvVars: []string{"COCOON_DEFAULT_ATPROTO_PROXY"}, 137 137 Value: "did:web:api.bsky.app#bsky_appview", 138 138 }, 139 139 + &cli.StringFlag{ 140 140 + Name: "blockstore-variant", 141 141 + EnvVars: []string{"COCOON_BLOCKSTORE_VARIANT"}, 142 142 + Value: "sqlite", 143 143 + }, 139 144 }, 140 145 Commands: []*cli.Command{ 141 146 runServe, ··· 158 163 Usage: "Start the cocoon PDS", 159 164 Flags: []cli.Flag{}, 160 165 Action: func(cmd *cli.Context) error { 166 166 + 161 167 s, err := server.New(&server.Args{ 162 168 Addr: cmd.String("addr"), 163 169 DbName: cmd.String("db-name"), ··· 185 191 }, 186 192 SessionSecret: cmd.String("session-secret"), 187 193 DefaultAtprotoProxy: cmd.String("default-atproto-proxy"), 194 194 + BlockstoreVariant: server.MustReturnBlockstoreVariant(cmd.String("blockstore-variant")), 188 195 }) 189 196 if err != nil { 190 197 fmt.Printf("error creating cocoon: %v", err)

+73 -54

identity/identity.go

··· 13 13 "github.com/bluesky-social/indigo/util" 14 14 ) 15 15 16 16 - func ResolveHandle(ctx context.Context, cli *http.Client, handle string) (string, error) { 17 17 - if cli == nil { 18 18 - cli = util.RobustHTTPClient() 19 19 - } 20 20 - 21 21 - var did string 22 22 - 23 23 - _, err := syntax.ParseHandle(handle) 16 16 + func ResolveHandleFromTXT(ctx context.Context, handle string) (string, error) { 17 17 + name := fmt.Sprintf("_atproto.%s", handle) 18 18 + recs, err := net.LookupTXT(name) 24 19 if err != nil { 25 25 - return "", err 20 20 + return "", fmt.Errorf("handle could not be resolved via txt: %w", err) 26 21 } 27 22 28 28 - recs, err := net.LookupTXT(fmt.Sprintf("_atproto.%s", handle)) 29 29 - if err == nil { 30 30 - for _, rec := range recs { 31 31 - if strings.HasPrefix(rec, "did=") { 32 32 - did = strings.Split(rec, "did=")[1] 33 33 - break 23 23 + for _, rec := range recs { 24 24 + if strings.HasPrefix(rec, "did=") { 25 25 + maybeDid := strings.Split(rec, "did=")[1] 26 26 + if _, err := syntax.ParseDID(maybeDid); err == nil { 27 27 + return maybeDid, nil 34 28 } 35 29 } 36 36 - } else { 37 37 - fmt.Printf("erorr getting txt records: %v\n", err) 38 30 } 39 31 40 40 - if did == "" { 41 41 - req, err := http.NewRequestWithContext( 42 42 - ctx, 43 43 - "GET", 44 44 - fmt.Sprintf("https://%s/.well-known/atproto-did", handle), 45 45 - nil, 46 46 - ) 47 47 - if err != nil { 48 48 - return "", nil 49 49 - } 32 32 + return "", fmt.Errorf("handle could not be resolved via txt: no record found") 33 33 + } 50 34 51 51 - resp, err := http.DefaultClient.Do(req) 52 52 - if err != nil { 53 53 - return "", nil 54 54 - } 55 55 - defer resp.Body.Close() 35 35 + func ResolveHandleFromWellKnown(ctx context.Context, cli *http.Client, handle string) (string, error) { 36 36 + ustr := fmt.Sprintf("https://%s/.well=known/atproto-did", handle) 37 37 + req, err := http.NewRequestWithContext( 38 38 + ctx, 39 39 + "GET", 40 40 + ustr, 41 41 + nil, 42 42 + ) 43 43 + if err != nil { 44 44 + return "", fmt.Errorf("handle could not be resolved via web: %w", err) 45 45 + } 56 46 57 57 - if resp.StatusCode != http.StatusOK { 58 58 - io.Copy(io.Discard, resp.Body) 59 59 - return "", fmt.Errorf("unable to resolve handle") 60 60 - } 47 47 + resp, err := cli.Do(req) 48 48 + if err != nil { 49 49 + return "", fmt.Errorf("handle could not be resolved via web: %w", err) 50 50 + } 51 51 + defer resp.Body.Close() 61 52 62 62 - b, err := io.ReadAll(resp.Body) 63 63 - if err != nil { 64 64 - return "", err 65 65 - } 53 53 + b, err := io.ReadAll(resp.Body) 54 54 + if err != nil { 55 55 + return "", fmt.Errorf("handle could not be resolved via web: %w", err) 56 56 + } 66 57 67 67 - maybeDid := string(b) 58 58 + if resp.StatusCode != http.StatusOK { 59 59 + return "", fmt.Errorf("handle could not be resolved via web: invalid status code %d", resp.StatusCode) 60 60 + } 68 61 69 69 - if _, err := syntax.ParseDID(maybeDid); err != nil { 70 70 - return "", fmt.Errorf("unable to resolve handle") 71 71 - } 62 62 + maybeDid := string(b) 72 63 73 73 - did = maybeDid 64 64 + if _, err := syntax.ParseDID(maybeDid); err != nil { 65 65 + return "", fmt.Errorf("handle could not be resolved via web: invalid did in document") 74 66 } 75 67 76 76 - return did, nil 68 68 + return maybeDid, nil 77 69 } 78 70 79 79 - func FetchDidDoc(ctx context.Context, cli *http.Client, did string) (*DidDoc, error) { 71 71 + func ResolveHandle(ctx context.Context, cli *http.Client, handle string) (string, error) { 80 72 if cli == nil { 81 73 cli = util.RobustHTTPClient() 82 74 } 83 75 84 84 - var ustr string 76 76 + _, err := syntax.ParseHandle(handle) 77 77 + if err != nil { 78 78 + return "", err 79 79 + } 80 80 + 81 81 + if maybeDidFromTxt, err := ResolveHandleFromTXT(ctx, handle); err == nil { 82 82 + return maybeDidFromTxt, nil 83 83 + } 84 84 + 85 85 + if maybeDidFromWeb, err := ResolveHandleFromWellKnown(ctx, cli, handle); err == nil { 86 86 + return maybeDidFromWeb, nil 87 87 + } 88 88 + 89 89 + return "", fmt.Errorf("handle could not be resolved") 90 90 + } 91 91 + 92 92 + func DidToDocUrl(did string) (string, error) { 85 93 if strings.HasPrefix(did, "did:plc:") { 86 86 - ustr = fmt.Sprintf("https://plc.directory/%s", did) 94 94 + return fmt.Sprintf("https://plc.directory/%s", did), nil 87 95 } else if strings.HasPrefix(did, "did:web:") { 88 88 - ustr = fmt.Sprintf("https://%s/.well-known/did.json", strings.TrimPrefix(did, "did:web:")) 96 96 + return fmt.Sprintf("https://%s/.well-known/did.json", strings.TrimPrefix(did, "did:web:")), nil 89 97 } else { 90 90 - return nil, fmt.Errorf("did was not a supported did type") 98 98 + return "", fmt.Errorf("did was not a supported did type") 99 99 + } 100 100 + } 101 101 + 102 102 + func FetchDidDoc(ctx context.Context, cli *http.Client, did string) (*DidDoc, error) { 103 103 + if cli == nil { 104 104 + cli = util.RobustHTTPClient() 105 105 + } 106 106 + 107 107 + ustr, err := DidToDocUrl(did) 108 108 + if err != nil { 109 109 + return nil, err 91 110 } 92 111 93 112 req, err := http.NewRequestWithContext(ctx, "GET", ustr, nil) ··· 95 114 return nil, err 96 115 } 97 116 98 98 - resp, err := http.DefaultClient.Do(req) 117 117 + resp, err := cli.Do(req) 99 118 if err != nil { 100 119 return nil, err 101 120 } ··· 103 122 104 123 if resp.StatusCode != 200 { 105 124 io.Copy(io.Discard, resp.Body) 106 106 - return nil, fmt.Errorf("could not find identity in plc registry") 125 125 + return nil, fmt.Errorf("unable to find did doc at url. did: %s. url: %s", did, ustr) 107 126 } 108 127 109 128 var diddoc DidDoc ··· 127 146 return nil, err 128 147 } 129 148 130 130 - resp, err := http.DefaultClient.Do(req) 149 149 + resp, err := cli.Do(req) 131 150 if err != nil { 132 151 return nil, err 133 152 }

+16 -5

identity/passport.go

··· 19 19 type Passport struct { 20 20 h *http.Client 21 21 bc BackingCache 22 22 - lk sync.Mutex 22 22 + mu sync.RWMutex 23 23 } 24 24 25 25 func NewPassport(h *http.Client, bc BackingCache) *Passport { ··· 30 30 return &Passport{ 31 31 h: h, 32 32 bc: bc, 33 33 - lk: sync.Mutex{}, 34 33 } 35 34 } 36 35 ··· 38 37 skipCache, _ := ctx.Value("skip-cache").(bool) 39 38 40 39 if !skipCache { 40 40 + p.mu.RLock() 41 41 cached, ok := p.bc.GetDoc(did) 42 42 + p.mu.RUnlock() 43 43 + 42 44 if ok { 43 45 return cached, nil 44 46 } 45 47 } 46 48 47 47 - p.lk.Lock() // this is pretty pathetic, and i should rethink this. but for now, fuck it 48 48 - defer p.lk.Unlock() 49 49 - 49 49 + // TODO: should coalesce requests here 50 50 doc, err := FetchDidDoc(ctx, p.h, did) 51 51 if err != nil { 52 52 return nil, err 53 53 } 54 54 55 55 + p.mu.Lock() 55 56 p.bc.PutDoc(did, doc) 57 57 + p.mu.Unlock() 56 58 57 59 return doc, nil 58 60 } ··· 61 63 skipCache, _ := ctx.Value("skip-cache").(bool) 62 64 63 65 if !skipCache { 66 66 + p.mu.RLock() 64 67 cached, ok := p.bc.GetDid(handle) 68 68 + p.mu.RUnlock() 69 69 + 65 70 if ok { 66 71 return cached, nil 67 72 } ··· 72 77 return "", err 73 78 } 74 79 80 80 + p.mu.Lock() 75 81 p.bc.PutDid(handle, did) 82 82 + p.mu.Unlock() 76 83 77 84 return did, nil 78 85 } 79 86 80 87 func (p *Passport) BustDoc(ctx context.Context, did string) error { 88 88 + p.mu.Lock() 89 89 + defer p.mu.Unlock() 81 90 return p.bc.BustDoc(did) 82 91 } 83 92 84 93 func (p *Passport) BustDid(ctx context.Context, handle string) error { 94 94 + p.mu.Lock() 95 95 + defer p.mu.Unlock() 85 96 return p.bc.BustDid(handle) 86 97 }

+77

recording_blockstore/recording_blockstore.go

··· 1 1 + package recording_blockstore 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + 6 6 + blockformat "github.com/ipfs/go-block-format" 7 7 + "github.com/ipfs/go-cid" 8 8 + blockstore "github.com/ipfs/go-ipfs-blockstore" 9 9 + ) 10 10 + 11 11 + type RecordingBlockstore struct { 12 12 + base blockstore.Blockstore 13 13 + 14 14 + inserts map[cid.Cid]blockformat.Block 15 15 + } 16 16 + 17 17 + func New(base blockstore.Blockstore) *RecordingBlockstore { 18 18 + return &RecordingBlockstore{ 19 19 + base: base, 20 20 + inserts: make(map[cid.Cid]blockformat.Block), 21 21 + } 22 22 + } 23 23 + 24 24 + func (bs *RecordingBlockstore) Has(ctx context.Context, c cid.Cid) (bool, error) { 25 25 + return bs.base.Has(ctx, c) 26 26 + } 27 27 + 28 28 + func (bs *RecordingBlockstore) Get(ctx context.Context, c cid.Cid) (blockformat.Block, error) { 29 29 + return bs.base.Get(ctx, c) 30 30 + } 31 31 + 32 32 + func (bs *RecordingBlockstore) GetSize(ctx context.Context, c cid.Cid) (int, error) { 33 33 + return bs.base.GetSize(ctx, c) 34 34 + } 35 35 + 36 36 + func (bs *RecordingBlockstore) DeleteBlock(ctx context.Context, c cid.Cid) error { 37 37 + return bs.base.DeleteBlock(ctx, c) 38 38 + } 39 39 + 40 40 + func (bs *RecordingBlockstore) Put(ctx context.Context, block blockformat.Block) error { 41 41 + if err := bs.base.Put(ctx, block); err != nil { 42 42 + return err 43 43 + } 44 44 + bs.inserts[block.Cid()] = block 45 45 + return nil 46 46 + } 47 47 + 48 48 + func (bs *RecordingBlockstore) PutMany(ctx context.Context, blocks []blockformat.Block) error { 49 49 + if err := bs.base.PutMany(ctx, blocks); err != nil { 50 50 + return err 51 51 + } 52 52 + 53 53 + for _, b := range blocks { 54 54 + bs.inserts[b.Cid()] = b 55 55 + } 56 56 + 57 57 + return nil 58 58 + } 59 59 + 60 60 + func (bs *RecordingBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { 61 61 + return bs.AllKeysChan(ctx) 62 62 + } 63 63 + 64 64 + func (bs *RecordingBlockstore) HashOnRead(enabled bool) { 65 65 + } 66 66 + 67 67 + func (bs *RecordingBlockstore) GetLogMap() map[cid.Cid]blockformat.Block { 68 68 + return bs.inserts 69 69 + } 70 70 + 71 71 + func (bs *RecordingBlockstore) GetLogArray() []blockformat.Block { 72 72 + var blocks []blockformat.Block 73 73 + for _, b := range bs.inserts { 74 74 + blocks = append(blocks, b) 75 75 + } 76 76 + return blocks 77 77 + }

+30

server/blockstore_variant.go

··· 1 1 + package server 2 2 + 3 3 + import ( 4 4 + "github.com/haileyok/cocoon/sqlite_blockstore" 5 5 + blockstore "github.com/ipfs/go-ipfs-blockstore" 6 6 + ) 7 7 + 8 8 + type BlockstoreVariant int 9 9 + 10 10 + const ( 11 11 + BlockstoreVariantSqlite = iota 12 12 + ) 13 13 + 14 14 + func MustReturnBlockstoreVariant(maybeBsv string) BlockstoreVariant { 15 15 + switch maybeBsv { 16 16 + case "sqlite": 17 17 + return BlockstoreVariantSqlite 18 18 + default: 19 19 + panic("invalid blockstore variant provided") 20 20 + } 21 21 + } 22 22 + 23 23 + func (s *Server) getBlockstore(did string) blockstore.Blockstore { 24 24 + switch s.config.BlockstoreVariant { 25 25 + case BlockstoreVariantSqlite: 26 26 + return sqlite_blockstore.New(did, s.db) 27 27 + default: 28 28 + return sqlite_blockstore.New(did, s.db) 29 29 + } 30 30 + }

+2 -3

server/handle_import_repo.go

··· 9 9 10 10 "github.com/bluesky-social/indigo/atproto/syntax" 11 11 "github.com/bluesky-social/indigo/repo" 12 12 - "github.com/haileyok/cocoon/blockstore" 13 12 "github.com/haileyok/cocoon/internal/helpers" 14 13 "github.com/haileyok/cocoon/models" 15 14 blocks "github.com/ipfs/go-block-format" ··· 27 26 return helpers.ServerError(e, nil) 28 27 } 29 28 30 30 - bs := blockstore.New(urepo.Repo.Did, s.db) 29 29 + bs := s.getBlockstore(urepo.Repo.Did) 31 30 32 31 cs, err := car.NewCarReader(bytes.NewReader(b)) 33 32 if err != nil { ··· 107 106 return helpers.ServerError(e, nil) 108 107 } 109 108 110 110 - if err := bs.UpdateRepo(context.TODO(), root, rev); err != nil { 109 109 + if err := s.UpdateRepo(context.TODO(), urepo.Repo.Did, root, rev); err != nil { 111 110 s.logger.Error("error updating repo after commit", "error", err) 112 111 return helpers.ServerError(e, nil) 113 112 }

+2 -3

server/handle_server_create_account.go

··· 14 14 "github.com/bluesky-social/indigo/events" 15 15 "github.com/bluesky-social/indigo/repo" 16 16 "github.com/bluesky-social/indigo/util" 17 17 - "github.com/haileyok/cocoon/blockstore" 18 17 "github.com/haileyok/cocoon/internal/helpers" 19 18 "github.com/haileyok/cocoon/models" 20 19 "github.com/labstack/echo/v4" ··· 177 176 } 178 177 179 178 if customDidHeader == "" { 180 180 - bs := blockstore.New(signupDid, s.db) 179 179 + bs := s.getBlockstore(signupDid) 181 180 r := repo.NewRepo(context.TODO(), signupDid, bs) 182 181 183 182 root, rev, err := r.Commit(context.TODO(), urepo.SignFor) ··· 186 185 return helpers.ServerError(e, nil) 187 186 } 188 187 189 189 - if err := bs.UpdateRepo(context.TODO(), root, rev); err != nil { 188 188 + if err := s.UpdateRepo(context.TODO(), urepo.Did, root, rev); err != nil { 190 189 s.logger.Error("error updating repo after commit", "error", err) 191 190 return helpers.ServerError(e, nil) 192 191 }

+1 -2

server/handle_sync_get_blocks.go

··· 6 6 "strings" 7 7 8 8 "github.com/bluesky-social/indigo/carstore" 9 9 - "github.com/haileyok/cocoon/blockstore" 10 9 "github.com/haileyok/cocoon/internal/helpers" 11 10 "github.com/ipfs/go-cid" 12 11 cbor "github.com/ipfs/go-ipld-cbor" ··· 54 53 return helpers.ServerError(e, nil) 55 54 } 56 55 57 57 - bs := blockstore.New(urepo.Repo.Did, s.db) 56 56 + bs := s.getBlockstore(urepo.Repo.Did) 58 57 59 58 for _, c := range cids { 60 59 b, err := bs.Get(context.TODO(), c)

+12 -12

server/repo.go

··· 16 16 "github.com/bluesky-social/indigo/events" 17 17 lexutil "github.com/bluesky-social/indigo/lex/util" 18 18 "github.com/bluesky-social/indigo/repo" 19 19 - "github.com/bluesky-social/indigo/util" 20 20 - "github.com/haileyok/cocoon/blockstore" 21 19 "github.com/haileyok/cocoon/internal/db" 22 20 "github.com/haileyok/cocoon/models" 21 21 + "github.com/haileyok/cocoon/recording_blockstore" 23 22 blocks "github.com/ipfs/go-block-format" 24 23 "github.com/ipfs/go-cid" 25 24 cbor "github.com/ipfs/go-ipld-cbor" ··· 103 102 return nil, err 104 103 } 105 104 106 106 - dbs := blockstore.New(urepo.Did, rm.db) 105 105 + dbs := rm.s.getBlockstore(urepo.Did) 106 106 + bs := recording_blockstore.New(dbs) 107 107 r, err := repo.OpenRepo(context.TODO(), dbs, rootcid) 108 108 109 109 entries := []models.Record{} ··· 274 274 } 275 275 } 276 276 277 277 - for _, op := range dbs.GetLog() { 277 277 + for _, op := range bs.GetLogMap() { 278 278 if _, err := carstore.LdWrite(buf, op.Cid().Bytes(), op.RawData()); err != nil { 279 279 return nil, err 280 280 } ··· 324 324 }, 325 325 }) 326 326 327 327 - if err := dbs.UpdateRepo(context.TODO(), newroot, rev); err != nil { 327 327 + if err := rm.s.UpdateRepo(context.TODO(), urepo.Did, newroot, rev); err != nil { 328 328 return nil, err 329 329 } 330 330 ··· 345 345 return cid.Undef, nil, err 346 346 } 347 347 348 348 - dbs := blockstore.New(urepo.Did, rm.db) 349 349 - bs := util.NewLoggingBstore(dbs) 348 348 + dbs := rm.s.getBlockstore(urepo.Did) 349 349 + bs := recording_blockstore.New(dbs) 350 350 351 351 r, err := repo.OpenRepo(context.TODO(), bs, c) 352 352 if err != nil { ··· 358 358 return cid.Undef, nil, err 359 359 } 360 360 361 361 - return c, bs.GetLoggedBlocks(), nil 361 361 + return c, bs.GetLogArray(), nil 362 362 } 363 363 364 364 func (rm *RepoMan) incrementBlobRefs(urepo models.Repo, cbor []byte) ([]cid.Cid, error) { ··· 414 414 return nil, fmt.Errorf("error unmarshaling cbor: %w", err) 415 415 } 416 416 417 417 - var deepiter func(interface{}) error 418 418 - deepiter = func(item interface{}) error { 417 417 + var deepiter func(any) error 418 418 + deepiter = func(item any) error { 419 419 switch val := item.(type) { 420 420 - case map[string]interface{}: 420 420 + case map[string]any: 421 421 if val["$type"] == "blob" { 422 422 if ref, ok := val["ref"].(string); ok { 423 423 c, err := cid.Parse(ref) ··· 430 430 return deepiter(v) 431 431 } 432 432 } 433 433 - case []interface{}: 433 433 + case []any: 434 434 for _, v := range val { 435 435 deepiter(v) 436 436 }

+13

server/server.go

··· 38 38 "github.com/haileyok/cocoon/oauth/dpop" 39 39 "github.com/haileyok/cocoon/oauth/provider" 40 40 "github.com/haileyok/cocoon/plc" 41 41 + "github.com/ipfs/go-cid" 41 42 echo_session "github.com/labstack/echo-contrib/session" 42 43 "github.com/labstack/echo/v4" 43 44 "github.com/labstack/echo/v4/middleware" ··· 104 105 SessionSecret string 105 106 106 107 DefaultAtprotoProxy string 108 108 + 109 109 + BlockstoreVariant BlockstoreVariant 107 110 } 108 111 109 112 type config struct { ··· 117 120 SmtpEmail string 118 121 SmtpName string 119 122 DefaultAtprotoProxy string 123 123 + BlockstoreVariant BlockstoreVariant 120 124 } 121 125 122 126 type CustomValidator struct { ··· 349 353 SmtpName: args.SmtpName, 350 354 SmtpEmail: args.SmtpEmail, 351 355 DefaultAtprotoProxy: args.DefaultAtprotoProxy, 356 356 + BlockstoreVariant: args.BlockstoreVariant, 352 357 }, 353 358 evtman: events.NewEventManager(events.NewMemPersister()), 354 359 passport: identity.NewPassport(h, identity.NewMemCache(10_000)), ··· 641 646 go s.doBackup() 642 647 } 643 648 } 649 649 + 650 650 + func (s *Server) UpdateRepo(ctx context.Context, did string, root cid.Cid, rev string) error { 651 651 + if err := s.db.Exec("UPDATE repos SET root = ?, rev = ? WHERE did = ?", nil, root.Bytes(), rev, did).Error; err != nil { 652 652 + return err 653 653 + } 654 654 + 655 655 + return nil 656 656 + }

+155

sqlite_blockstore/sqlite_blockstore.go

··· 1 1 + package sqlite_blockstore 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + 7 7 + "github.com/bluesky-social/indigo/atproto/syntax" 8 8 + "github.com/haileyok/cocoon/internal/db" 9 9 + "github.com/haileyok/cocoon/models" 10 10 + blocks "github.com/ipfs/go-block-format" 11 11 + "github.com/ipfs/go-cid" 12 12 + "gorm.io/gorm/clause" 13 13 + ) 14 14 + 15 15 + type SqliteBlockstore struct { 16 16 + db *db.DB 17 17 + did string 18 18 + readonly bool 19 19 + inserts map[cid.Cid]blocks.Block 20 20 + } 21 21 + 22 22 + func New(did string, db *db.DB) *SqliteBlockstore { 23 23 + return &SqliteBlockstore{ 24 24 + did: did, 25 25 + db: db, 26 26 + readonly: false, 27 27 + inserts: map[cid.Cid]blocks.Block{}, 28 28 + } 29 29 + } 30 30 + 31 31 + func NewReadOnly(did string, db *db.DB) *SqliteBlockstore { 32 32 + return &SqliteBlockstore{ 33 33 + did: did, 34 34 + db: db, 35 35 + readonly: true, 36 36 + inserts: map[cid.Cid]blocks.Block{}, 37 37 + } 38 38 + } 39 39 + 40 40 + func (bs *SqliteBlockstore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, error) { 41 41 + var block models.Block 42 42 + 43 43 + maybeBlock, ok := bs.inserts[cid] 44 44 + if ok { 45 45 + return maybeBlock, nil 46 46 + } 47 47 + 48 48 + if err := bs.db.Raw("SELECT * FROM blocks WHERE did = ? AND cid = ?", nil, bs.did, cid.Bytes()).Scan(&block).Error; err != nil { 49 49 + return nil, err 50 50 + } 51 51 + 52 52 + b, err := blocks.NewBlockWithCid(block.Value, cid) 53 53 + if err != nil { 54 54 + return nil, err 55 55 + } 56 56 + 57 57 + return b, nil 58 58 + } 59 59 + 60 60 + func (bs *SqliteBlockstore) Put(ctx context.Context, block blocks.Block) error { 61 61 + bs.inserts[block.Cid()] = block 62 62 + 63 63 + if bs.readonly { 64 64 + return nil 65 65 + } 66 66 + 67 67 + b := models.Block{ 68 68 + Did: bs.did, 69 69 + Cid: block.Cid().Bytes(), 70 70 + Rev: syntax.NewTIDNow(0).String(), // TODO: WARN, this is bad. don't do this 71 71 + Value: block.RawData(), 72 72 + } 73 73 + 74 74 + if err := bs.db.Create(&b, []clause.Expression{clause.OnConflict{ 75 75 + Columns: []clause.Column{{Name: "did"}, {Name: "cid"}}, 76 76 + UpdateAll: true, 77 77 + }}).Error; err != nil { 78 78 + return err 79 79 + } 80 80 + 81 81 + return nil 82 82 + } 83 83 + 84 84 + func (bs *SqliteBlockstore) DeleteBlock(context.Context, cid.Cid) error { 85 85 + panic("not implemented") 86 86 + } 87 87 + 88 88 + func (bs *SqliteBlockstore) Has(context.Context, cid.Cid) (bool, error) { 89 89 + panic("not implemented") 90 90 + } 91 91 + 92 92 + func (bs *SqliteBlockstore) GetSize(context.Context, cid.Cid) (int, error) { 93 93 + panic("not implemented") 94 94 + } 95 95 + 96 96 + func (bs *SqliteBlockstore) PutMany(ctx context.Context, blocks []blocks.Block) error { 97 97 + tx := bs.db.BeginDangerously() 98 98 + 99 99 + for _, block := range blocks { 100 100 + bs.inserts[block.Cid()] = block 101 101 + 102 102 + if bs.readonly { 103 103 + continue 104 104 + } 105 105 + 106 106 + b := models.Block{ 107 107 + Did: bs.did, 108 108 + Cid: block.Cid().Bytes(), 109 109 + Rev: syntax.NewTIDNow(0).String(), // TODO: WARN, this is bad. don't do this 110 110 + Value: block.RawData(), 111 111 + } 112 112 + 113 113 + if err := tx.Clauses(clause.OnConflict{ 114 114 + Columns: []clause.Column{{Name: "did"}, {Name: "cid"}}, 115 115 + UpdateAll: true, 116 116 + }).Create(&b).Error; err != nil { 117 117 + tx.Rollback() 118 118 + return err 119 119 + } 120 120 + } 121 121 + 122 122 + if bs.readonly { 123 123 + return nil 124 124 + } 125 125 + 126 126 + tx.Commit() 127 127 + 128 128 + return nil 129 129 + } 130 130 + 131 131 + func (bs *SqliteBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { 132 132 + panic("not implemented") 133 133 + } 134 134 + 135 135 + func (bs *SqliteBlockstore) HashOnRead(enabled bool) { 136 136 + panic("not implemented") 137 137 + } 138 138 + 139 139 + func (bs *SqliteBlockstore) Execute(ctx context.Context) error { 140 140 + if !bs.readonly { 141 141 + return fmt.Errorf("blockstore was not readonly") 142 142 + } 143 143 + 144 144 + bs.readonly = false 145 145 + for _, b := range bs.inserts { 146 146 + bs.Put(ctx, b) 147 147 + } 148 148 + bs.readonly = true 149 149 + 150 150 + return nil 151 151 + } 152 152 + 153 153 + func (bs *SqliteBlockstore) GetLog() map[cid.Cid]blocks.Block { 154 154 + return bs.inserts 155 155 + }