An atproto PDS written in Go
switch to use the new repo package (#71)
* Switch to the new repo lib
* Fix nil pointer
* caching
* clean
+411
-181
+4
-5
server/handle_import_repo.go
+4
-5
server/handle_import_repo.go
···
8
8
"strings"
9
9
10
10
"github.com/bluesky-social/indigo/atproto/syntax"
11
-
"github.com/bluesky-social/indigo/repo"
12
11
"github.com/haileyok/cocoon/internal/helpers"
13
12
"github.com/haileyok/cocoon/models"
14
13
blocks "github.com/ipfs/go-block-format"
···
60
59
return helpers.ServerError(e, nil)
61
60
}
62
61
63
-
r, err := repo.OpenRepo(context.TODO(), bs, cs.Header.Roots[0])
62
+
r, err := openRepo(context.TODO(), bs, cs.Header.Roots[0], urepo.Repo.Did)
64
63
if err != nil {
65
64
logger.Error("could not open repo", "error", err)
66
65
return helpers.ServerError(e, nil)
···
70
69
71
70
clock := syntax.NewTIDClock(0)
72
71
73
-
if err := r.ForEach(context.TODO(), "", func(key string, cid cid.Cid) error {
74
-
pts := strings.Split(key, "/")
72
+
if err := r.MST.Walk(func(key []byte, cid cid.Cid) error {
73
+
pts := strings.Split(string(key), "/")
75
74
nsid := pts[0]
76
75
rkey := pts[1]
77
76
cidStr := cid.String()
···
103
102
104
103
tx.Commit()
105
104
106
-
root, rev, err := r.Commit(context.TODO(), urepo.SignFor)
105
+
root, rev, err := commitRepo(context.TODO(), bs, r, urepo.Repo.SigningKey)
107
106
if err != nil {
108
107
logger.Error("error committing", "error", err)
109
108
return helpers.ServerError(e, nil)
+12
-3
server/handle_server_create_account.go
+12
-3
server/handle_server_create_account.go
···
10
10
"github.com/Azure/go-autorest/autorest/to"
11
11
"github.com/bluesky-social/indigo/api/atproto"
12
12
"github.com/bluesky-social/indigo/atproto/atcrypto"
13
+
atp "github.com/bluesky-social/indigo/atproto/repo"
14
+
"github.com/bluesky-social/indigo/atproto/repo/mst"
15
+
"github.com/bluesky-social/indigo/atproto/syntax"
13
16
"github.com/bluesky-social/indigo/events"
14
-
"github.com/bluesky-social/indigo/repo"
15
17
"github.com/bluesky-social/indigo/util"
16
18
"github.com/haileyok/cocoon/internal/helpers"
17
19
"github.com/haileyok/cocoon/models"
···
220
222
221
223
if request.Did == nil || *request.Did == "" {
222
224
bs := s.getBlockstore(signupDid)
223
-
r := repo.NewRepo(context.TODO(), signupDid, bs)
225
+
226
+
clk := syntax.NewTIDClock(0)
227
+
r := &atp.Repo{
228
+
DID: syntax.DID(signupDid),
229
+
Clock: clk,
230
+
MST: mst.NewEmptyTree(),
231
+
RecordStore: bs,
232
+
}
224
233
225
-
root, rev, err := r.Commit(context.TODO(), urepo.SignFor)
234
+
root, rev, err := commitRepo(context.TODO(), bs, r, urepo.SigningKey)
226
235
if err != nil {
227
236
logger.Error("error committing", "error", err)
228
237
return helpers.ServerError(e, nil)
+391
-167
server/repo.go
+391
-167
server/repo.go
···
6
6
"encoding/json"
7
7
"fmt"
8
8
"io"
9
+
"sync"
9
10
"time"
10
11
11
12
"github.com/Azure/go-autorest/autorest/to"
12
13
"github.com/bluesky-social/indigo/api/atproto"
14
+
"github.com/bluesky-social/indigo/atproto/atcrypto"
13
15
"github.com/bluesky-social/indigo/atproto/atdata"
16
+
atp "github.com/bluesky-social/indigo/atproto/repo"
17
+
"github.com/bluesky-social/indigo/atproto/repo/mst"
14
18
"github.com/bluesky-social/indigo/atproto/syntax"
15
19
"github.com/bluesky-social/indigo/carstore"
16
20
"github.com/bluesky-social/indigo/events"
17
21
lexutil "github.com/bluesky-social/indigo/lex/util"
18
-
"github.com/bluesky-social/indigo/repo"
19
22
"github.com/haileyok/cocoon/internal/db"
20
23
"github.com/haileyok/cocoon/metrics"
21
24
"github.com/haileyok/cocoon/models"
22
25
"github.com/haileyok/cocoon/recording_blockstore"
23
26
blocks "github.com/ipfs/go-block-format"
24
27
"github.com/ipfs/go-cid"
28
+
blockstore "github.com/ipfs/go-ipfs-blockstore"
25
29
cbor "github.com/ipfs/go-ipld-cbor"
26
30
"github.com/ipld/go-car"
31
+
"github.com/multiformats/go-multihash"
27
32
"gorm.io/gorm/clause"
28
33
)
29
34
35
+
type cachedRepo struct {
36
+
mu sync.Mutex
37
+
repo *atp.Repo
38
+
root cid.Cid
39
+
}
40
+
30
41
type RepoMan struct {
31
42
db *db.DB
32
43
s *Server
33
44
clock *syntax.TIDClock
45
+
46
+
cacheMu sync.Mutex
47
+
cache map[string]*cachedRepo
34
48
}
35
49
36
50
func NewRepoMan(s *Server) *RepoMan {
···
40
54
s: s,
41
55
db: s.db,
42
56
clock: clock,
57
+
cache: make(map[string]*cachedRepo),
43
58
}
44
59
}
45
60
61
+
func (rm *RepoMan) withRepo(ctx context.Context, did string, rootCid cid.Cid, fn func(r *atp.Repo) (newRoot cid.Cid, err error)) error {
62
+
rm.cacheMu.Lock()
63
+
cr, ok := rm.cache[did]
64
+
if !ok {
65
+
cr = &cachedRepo{}
66
+
rm.cache[did] = cr
67
+
}
68
+
rm.cacheMu.Unlock()
69
+
70
+
cr.mu.Lock()
71
+
defer cr.mu.Unlock()
72
+
73
+
if cr.repo == nil || cr.root != rootCid {
74
+
bs := rm.s.getBlockstore(did)
75
+
r, err := openRepo(ctx, bs, rootCid, did)
76
+
if err != nil {
77
+
return err
78
+
}
79
+
cr.repo = r
80
+
cr.root = rootCid
81
+
}
82
+
83
+
newRoot, err := fn(cr.repo)
84
+
if err != nil {
85
+
// invalidate on error since the tree may be partially mutated
86
+
cr.repo = nil
87
+
cr.root = cid.Undef
88
+
return err
89
+
}
90
+
91
+
cr.root = newRoot
92
+
return nil
93
+
}
94
+
46
95
type OpType string
47
96
48
97
var (
···
96
145
Rev string `json:"rev"`
97
146
}
98
147
148
+
func openRepo(ctx context.Context, bs blockstore.Blockstore, rootCid cid.Cid, did string) (*atp.Repo, error) {
149
+
commitBlock, err := bs.Get(ctx, rootCid)
150
+
if err != nil {
151
+
return nil, fmt.Errorf("reading commit block: %w", err)
152
+
}
153
+
154
+
var commit atp.Commit
155
+
if err := commit.UnmarshalCBOR(bytes.NewReader(commitBlock.RawData())); err != nil {
156
+
return nil, fmt.Errorf("parsing commit block: %w", err)
157
+
}
158
+
159
+
tree, err := mst.LoadTreeFromStore(ctx, bs, commit.Data)
160
+
if err != nil {
161
+
return nil, fmt.Errorf("loading MST: %w", err)
162
+
}
163
+
164
+
clk := syntax.ClockFromTID(syntax.TID(commit.Rev))
165
+
return &atp.Repo{
166
+
DID: syntax.DID(did),
167
+
Clock: &clk,
168
+
MST: *tree,
169
+
RecordStore: bs,
170
+
}, nil
171
+
}
172
+
173
+
func commitRepo(ctx context.Context, bs blockstore.Blockstore, r *atp.Repo, signingKey []byte) (cid.Cid, string, error) {
174
+
if _, err := r.MST.WriteDiffBlocks(ctx, bs); err != nil {
175
+
return cid.Undef, "", fmt.Errorf("writing MST blocks: %w", err)
176
+
}
177
+
178
+
commit, err := r.Commit()
179
+
if err != nil {
180
+
return cid.Undef, "", fmt.Errorf("creating commit: %w", err)
181
+
}
182
+
183
+
privkey, err := atcrypto.ParsePrivateBytesK256(signingKey)
184
+
if err != nil {
185
+
return cid.Undef, "", fmt.Errorf("parsing signing key: %w", err)
186
+
}
187
+
if err := commit.Sign(privkey); err != nil {
188
+
return cid.Undef, "", fmt.Errorf("signing commit: %w", err)
189
+
}
190
+
191
+
buf := new(bytes.Buffer)
192
+
if err := commit.MarshalCBOR(buf); err != nil {
193
+
return cid.Undef, "", fmt.Errorf("marshaling commit: %w", err)
194
+
}
195
+
196
+
pref := cid.NewPrefixV1(cid.DagCBOR, multihash.SHA2_256)
197
+
commitCid, err := pref.Sum(buf.Bytes())
198
+
if err != nil {
199
+
return cid.Undef, "", fmt.Errorf("computing commit CID: %w", err)
200
+
}
201
+
202
+
blk, err := blocks.NewBlockWithCid(buf.Bytes(), commitCid)
203
+
if err != nil {
204
+
return cid.Undef, "", fmt.Errorf("creating commit block: %w", err)
205
+
}
206
+
if err := bs.Put(ctx, blk); err != nil {
207
+
return cid.Undef, "", fmt.Errorf("writing commit block: %w", err)
208
+
}
209
+
210
+
return commitCid, commit.Rev, nil
211
+
}
212
+
213
+
func putRecordBlock(ctx context.Context, bs blockstore.Blockstore, rec *MarshalableMap) (cid.Cid, error) {
214
+
buf := new(bytes.Buffer)
215
+
if err := rec.MarshalCBOR(buf); err != nil {
216
+
return cid.Undef, err
217
+
}
218
+
219
+
pref := cid.NewPrefixV1(cid.DagCBOR, multihash.SHA2_256)
220
+
c, err := pref.Sum(buf.Bytes())
221
+
if err != nil {
222
+
return cid.Undef, err
223
+
}
224
+
225
+
blk, err := blocks.NewBlockWithCid(buf.Bytes(), c)
226
+
if err != nil {
227
+
return cid.Undef, err
228
+
}
229
+
if err := bs.Put(ctx, blk); err != nil {
230
+
return cid.Undef, err
231
+
}
232
+
233
+
return c, nil
234
+
}
235
+
99
236
// TODO make use of swap commit
100
237
func (rm *RepoMan) applyWrites(ctx context.Context, urepo models.Repo, writes []Op, swapCommit *string) ([]ApplyWriteResult, error) {
101
238
rootcid, err := cid.Cast(urepo.Root)
···
105
242
106
243
dbs := rm.s.getBlockstore(urepo.Did)
107
244
bs := recording_blockstore.New(dbs)
108
-
r, err := repo.OpenRepo(ctx, bs, rootcid)
109
245
110
246
var results []ApplyWriteResult
247
+
var ops []*atp.Operation
248
+
var entries []models.Record
249
+
var newroot cid.Cid
250
+
var rev string
111
251
112
-
entries := make([]models.Record, 0, len(writes))
113
-
for i, op := range writes {
114
-
// updates or deletes must supply an rkey
115
-
if op.Type != OpTypeCreate && op.Rkey == nil {
116
-
return nil, fmt.Errorf("invalid rkey")
117
-
} else if op.Type == OpTypeCreate && op.Rkey != nil {
118
-
// we should conver this op to an update if the rkey already exists
119
-
_, _, err := r.GetRecord(ctx, fmt.Sprintf("%s/%s", op.Collection, *op.Rkey))
120
-
if err == nil {
121
-
op.Type = OpTypeUpdate
252
+
if err := rm.withRepo(ctx, urepo.Did, rootcid, func(r *atp.Repo) (cid.Cid, error) {
253
+
entries = make([]models.Record, 0, len(writes))
254
+
for i, op := range writes {
255
+
// updates or deletes must supply an rkey
256
+
if op.Type != OpTypeCreate && op.Rkey == nil {
257
+
return cid.Undef, fmt.Errorf("invalid rkey")
258
+
} else if op.Type == OpTypeCreate && op.Rkey != nil {
259
+
// we should convert this op to an update if the rkey already exists
260
+
path := fmt.Sprintf("%s/%s", op.Collection, *op.Rkey)
261
+
existing, _ := r.MST.Get([]byte(path))
262
+
if existing != nil {
263
+
op.Type = OpTypeUpdate
264
+
}
265
+
} else if op.Rkey == nil {
266
+
// creates that don't supply an rkey will have one generated for them
267
+
op.Rkey = to.StringPtr(rm.clock.Next().String())
268
+
writes[i].Rkey = op.Rkey
122
269
}
123
-
} else if op.Rkey == nil {
124
-
// creates that don't supply an rkey will have one generated for them
125
-
op.Rkey = to.StringPtr(rm.clock.Next().String())
126
-
writes[i].Rkey = op.Rkey
127
-
}
128
270
129
-
// validate the record key is actually valid
130
-
_, err := syntax.ParseRecordKey(*op.Rkey)
131
-
if err != nil {
132
-
return nil, err
133
-
}
271
+
path := fmt.Sprintf("%s/%s", op.Collection, *op.Rkey)
134
272
135
-
switch op.Type {
136
-
case OpTypeCreate:
137
-
// HACK: this fixes some type conversions, mainly around integers
138
-
// first we convert to json bytes
139
-
b, err := json.Marshal(*op.Record)
273
+
// validate the record key is actually valid
274
+
_, err := syntax.ParseRecordKey(*op.Rkey)
140
275
if err != nil {
141
-
return nil, err
276
+
return cid.Undef, err
142
277
}
143
-
// then we use atdata.UnmarshalJSON to convert it back to a map
144
-
out, err := atdata.UnmarshalJSON(b)
145
-
if err != nil {
146
-
return nil, err
147
-
}
148
-
// finally we can cast to a MarshalableMap
149
-
mm := MarshalableMap(out)
150
278
151
-
// HACK: if a record doesn't contain a $type, we can manually set it here based on the op's collection
152
-
// i forget why this is actually necessary?
153
-
if mm["$type"] == "" {
154
-
mm["$type"] = op.Collection
155
-
}
279
+
switch op.Type {
280
+
case OpTypeCreate:
281
+
// HACK: this fixes some type conversions, mainly around integers
282
+
b, err := json.Marshal(*op.Record)
283
+
if err != nil {
284
+
return cid.Undef, err
285
+
}
286
+
out, err := atdata.UnmarshalJSON(b)
287
+
if err != nil {
288
+
return cid.Undef, err
289
+
}
290
+
mm := MarshalableMap(out)
156
291
157
-
nc, err := r.PutRecord(ctx, fmt.Sprintf("%s/%s", op.Collection, *op.Rkey), &mm)
158
-
if err != nil {
159
-
return nil, err
160
-
}
292
+
// HACK: if a record doesn't contain a $type, we can manually set it here based on the op's collection
293
+
if mm["$type"] == "" {
294
+
mm["$type"] = op.Collection
295
+
}
161
296
162
-
d, err := atdata.MarshalCBOR(mm)
163
-
if err != nil {
164
-
return nil, err
165
-
}
297
+
nc, err := putRecordBlock(ctx, bs, &mm)
298
+
if err != nil {
299
+
return cid.Undef, err
300
+
}
166
301
167
-
entries = append(entries, models.Record{
168
-
Did: urepo.Did,
169
-
CreatedAt: rm.clock.Next().String(),
170
-
Nsid: op.Collection,
171
-
Rkey: *op.Rkey,
172
-
Cid: nc.String(),
173
-
Value: d,
174
-
})
302
+
atpOp, err := atp.ApplyOp(&r.MST, path, &nc)
303
+
if err != nil {
304
+
return cid.Undef, err
305
+
}
306
+
ops = append(ops, atpOp)
175
307
176
-
results = append(results, ApplyWriteResult{
177
-
Type: to.StringPtr(OpTypeCreate.String()),
178
-
Uri: to.StringPtr("at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey),
179
-
Cid: to.StringPtr(nc.String()),
180
-
ValidationStatus: to.StringPtr("valid"), // TODO: obviously this might not be true atm lol
181
-
})
182
-
case OpTypeDelete:
183
-
// try to find the old record in the database
184
-
var old models.Record
185
-
if err := rm.db.Raw(ctx, "SELECT value FROM records WHERE did = ? AND nsid = ? AND rkey = ?", nil, urepo.Did, op.Collection, op.Rkey).Scan(&old).Error; err != nil {
186
-
return nil, err
187
-
}
308
+
d, err := atdata.MarshalCBOR(mm)
309
+
if err != nil {
310
+
return cid.Undef, err
311
+
}
188
312
189
-
// TODO: this is really confusing, and looking at it i have no idea why i did this. below when we are doing deletes, we
190
-
// check if `cid` here is nil to indicate if we should delete. that really doesn't make much sense and its super illogical
191
-
// when reading this code. i dont feel like fixing right now though so
192
-
entries = append(entries, models.Record{
193
-
Did: urepo.Did,
194
-
Nsid: op.Collection,
195
-
Rkey: *op.Rkey,
196
-
Value: old.Value,
197
-
})
313
+
entries = append(entries, models.Record{
314
+
Did: urepo.Did,
315
+
CreatedAt: rm.clock.Next().String(),
316
+
Nsid: op.Collection,
317
+
Rkey: *op.Rkey,
318
+
Cid: nc.String(),
319
+
Value: d,
320
+
})
198
321
199
-
// delete the record from the repo
200
-
err := r.DeleteRecord(ctx, fmt.Sprintf("%s/%s", op.Collection, *op.Rkey))
201
-
if err != nil {
202
-
return nil, err
203
-
}
322
+
results = append(results, ApplyWriteResult{
323
+
Type: to.StringPtr(OpTypeCreate.String()),
324
+
Uri: to.StringPtr("at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey),
325
+
Cid: to.StringPtr(nc.String()),
326
+
ValidationStatus: to.StringPtr("valid"), // TODO: obviously this might not be true atm lol
327
+
})
328
+
case OpTypeDelete:
329
+
// try to find the old record in the database
330
+
var old models.Record
331
+
if err := rm.db.Raw(ctx, "SELECT value FROM records WHERE did = ? AND nsid = ? AND rkey = ?", nil, urepo.Did, op.Collection, op.Rkey).Scan(&old).Error; err != nil {
332
+
return cid.Undef, err
333
+
}
204
334
205
-
// add a result for the delete
206
-
results = append(results, ApplyWriteResult{
207
-
Type: to.StringPtr(OpTypeDelete.String()),
208
-
})
209
-
case OpTypeUpdate:
210
-
// HACK: same hack as above for type fixes
211
-
b, err := json.Marshal(*op.Record)
212
-
if err != nil {
213
-
return nil, err
214
-
}
215
-
out, err := atdata.UnmarshalJSON(b)
216
-
if err != nil {
217
-
return nil, err
218
-
}
219
-
mm := MarshalableMap(out)
335
+
// TODO: this is really confusing, and looking at it i have no idea why i did this. below when we are doing deletes, we
336
+
// check if `cid` here is nil to indicate if we should delete. that really doesn't make much sense and its super illogical
337
+
// when reading this code. i dont feel like fixing right now though so
338
+
entries = append(entries, models.Record{
339
+
Did: urepo.Did,
340
+
Nsid: op.Collection,
341
+
Rkey: *op.Rkey,
342
+
Value: old.Value,
343
+
})
220
344
221
-
nc, err := r.UpdateRecord(ctx, fmt.Sprintf("%s/%s", op.Collection, *op.Rkey), &mm)
222
-
if err != nil {
223
-
return nil, err
224
-
}
345
+
atpOp, err := atp.ApplyOp(&r.MST, path, nil)
346
+
if err != nil {
347
+
return cid.Undef, err
348
+
}
349
+
ops = append(ops, atpOp)
225
350
226
-
d, err := atdata.MarshalCBOR(mm)
227
-
if err != nil {
228
-
return nil, err
229
-
}
351
+
results = append(results, ApplyWriteResult{
352
+
Type: to.StringPtr(OpTypeDelete.String()),
353
+
})
354
+
case OpTypeUpdate:
355
+
// HACK: same hack as above for type fixes
356
+
b, err := json.Marshal(*op.Record)
357
+
if err != nil {
358
+
return cid.Undef, err
359
+
}
360
+
out, err := atdata.UnmarshalJSON(b)
361
+
if err != nil {
362
+
return cid.Undef, err
363
+
}
364
+
mm := MarshalableMap(out)
230
365
231
-
entries = append(entries, models.Record{
232
-
Did: urepo.Did,
233
-
CreatedAt: rm.clock.Next().String(),
234
-
Nsid: op.Collection,
235
-
Rkey: *op.Rkey,
236
-
Cid: nc.String(),
237
-
Value: d,
238
-
})
366
+
nc, err := putRecordBlock(ctx, bs, &mm)
367
+
if err != nil {
368
+
return cid.Undef, err
369
+
}
370
+
371
+
atpOp, err := atp.ApplyOp(&r.MST, path, &nc)
372
+
if err != nil {
373
+
return cid.Undef, err
374
+
}
375
+
ops = append(ops, atpOp)
376
+
377
+
d, err := atdata.MarshalCBOR(mm)
378
+
if err != nil {
379
+
return cid.Undef, err
380
+
}
381
+
382
+
entries = append(entries, models.Record{
383
+
Did: urepo.Did,
384
+
CreatedAt: rm.clock.Next().String(),
385
+
Nsid: op.Collection,
386
+
Rkey: *op.Rkey,
387
+
Cid: nc.String(),
388
+
Value: d,
389
+
})
239
390
240
-
results = append(results, ApplyWriteResult{
241
-
Type: to.StringPtr(OpTypeUpdate.String()),
242
-
Uri: to.StringPtr("at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey),
243
-
Cid: to.StringPtr(nc.String()),
244
-
ValidationStatus: to.StringPtr("valid"), // TODO: obviously this might not be true atm lol
245
-
})
391
+
results = append(results, ApplyWriteResult{
392
+
Type: to.StringPtr(OpTypeUpdate.String()),
393
+
Uri: to.StringPtr("at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey),
394
+
Cid: to.StringPtr(nc.String()),
395
+
ValidationStatus: to.StringPtr("valid"), // TODO: obviously this might not be true atm lol
396
+
})
397
+
}
246
398
}
247
-
}
399
+
400
+
// commit and get the new root
401
+
var commitErr error
402
+
newroot, rev, commitErr = commitRepo(ctx, bs, r, urepo.SigningKey)
403
+
if commitErr != nil {
404
+
return cid.Undef, commitErr
405
+
}
248
406
249
-
// commit and get the new root
250
-
newroot, rev, err := r.Commit(ctx, urepo.SignFor)
251
-
if err != nil {
407
+
return newroot, nil
408
+
}); err != nil {
252
409
return nil, err
253
410
}
254
411
···
270
427
return nil, err
271
428
}
272
429
273
-
// get a diff of the changes to the repo
274
-
diffops, err := r.DiffSince(ctx, rootcid)
275
-
if err != nil {
276
-
return nil, err
277
-
}
278
-
279
-
// create the repo ops for the given diff
280
-
ops := make([]*atproto.SyncSubscribeRepos_RepoOp, 0, len(diffops))
281
-
for _, op := range diffops {
282
-
var c cid.Cid
283
-
switch op.Op {
284
-
case "add", "mut":
430
+
// create the repo ops for the firehose from the tracked operations
431
+
repoOps := make([]*atproto.SyncSubscribeRepos_RepoOp, 0, len(ops))
432
+
for _, op := range ops {
433
+
if op.IsCreate() || op.IsUpdate() {
285
434
kind := "create"
286
-
if op.Op == "mut" {
435
+
if op.IsUpdate() {
287
436
kind = "update"
288
437
}
289
438
290
-
c = op.NewCid
291
-
ll := lexutil.LexLink(op.NewCid)
292
-
ops = append(ops, &atproto.SyncSubscribeRepos_RepoOp{
439
+
ll := lexutil.LexLink(*op.Value)
440
+
repoOps = append(repoOps, &atproto.SyncSubscribeRepos_RepoOp{
293
441
Action: kind,
294
-
Path: op.Rpath,
442
+
Path: op.Path,
295
443
Cid: &ll,
296
444
})
297
445
298
-
case "del":
299
-
c = op.OldCid
300
-
ll := lexutil.LexLink(op.OldCid)
301
-
ops = append(ops, &atproto.SyncSubscribeRepos_RepoOp{
446
+
blk, err := dbs.Get(ctx, *op.Value)
447
+
if err != nil {
448
+
return nil, err
449
+
}
450
+
if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil {
451
+
return nil, err
452
+
}
453
+
} else if op.IsDelete() {
454
+
ll := lexutil.LexLink(*op.Prev)
455
+
repoOps = append(repoOps, &atproto.SyncSubscribeRepos_RepoOp{
302
456
Action: "delete",
303
-
Path: op.Rpath,
457
+
Path: op.Path,
304
458
Cid: nil,
305
459
Prev: &ll,
306
460
})
307
-
}
308
461
309
-
blk, err := dbs.Get(ctx, c)
310
-
if err != nil {
311
-
return nil, err
312
-
}
313
-
314
-
// write the block to the buffer
315
-
if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil {
316
-
return nil, err
462
+
blk, err := dbs.Get(ctx, *op.Prev)
463
+
if err != nil {
464
+
return nil, err
465
+
}
466
+
if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil {
467
+
return nil, err
468
+
}
317
469
}
318
470
}
319
471
320
472
// write the writelog to the buffer
321
-
for _, op := range bs.GetWriteLog() {
322
-
if _, err := carstore.LdWrite(buf, op.Cid().Bytes(), op.RawData()); err != nil {
473
+
for _, blk := range bs.GetWriteLog() {
474
+
if _, err := carstore.LdWrite(buf, blk.Cid().Bytes(), blk.RawData()); err != nil {
323
475
return nil, err
324
476
}
325
477
}
···
374
526
Since: &urepo.Rev,
375
527
Commit: lexutil.LexLink(newroot),
376
528
Time: time.Now().Format(time.RFC3339Nano),
377
-
Ops: ops,
529
+
Ops: repoOps,
378
530
TooBig: false,
379
531
},
380
532
})
···
394
546
return results, nil
395
547
}
396
548
397
-
// this is a fun little guy. to get a proof, we need to read the record out of the blockstore and record how we actually
398
-
// got to the guy. we'll wrap a new blockstore in a recording blockstore, then return the log for proof
399
549
func (rm *RepoMan) getRecordProof(ctx context.Context, urepo models.Repo, collection, rkey string) (cid.Cid, []blocks.Block, error) {
400
-
c, err := cid.Cast(urepo.Root)
550
+
commitCid, err := cid.Cast(urepo.Root)
401
551
if err != nil {
402
552
return cid.Undef, nil, err
403
553
}
404
554
405
555
dbs := rm.s.getBlockstore(urepo.Did)
406
-
bs := recording_blockstore.New(dbs)
556
+
557
+
var proofBlocks []blocks.Block
558
+
var recordCid *cid.Cid
559
+
560
+
if err := rm.withRepo(ctx, urepo.Did, commitCid, func(r *atp.Repo) (cid.Cid, error) {
561
+
path := collection + "/" + rkey
562
+
563
+
// walk the cached in-memory tree to find the record and collect MST node CIDs on the path
564
+
nodeCIDs := collectPathNodeCIDs(r.MST.Root, []byte(path))
565
+
566
+
rc, getErr := r.MST.Get([]byte(path))
567
+
if getErr != nil {
568
+
return cid.Undef, getErr
569
+
}
570
+
if rc == nil {
571
+
return cid.Undef, fmt.Errorf("record not found: %s", path)
572
+
}
573
+
recordCid = rc
574
+
575
+
// read the commit block
576
+
commitBlk, err := dbs.Get(ctx, commitCid)
577
+
if err != nil {
578
+
return cid.Undef, fmt.Errorf("reading commit block for proof: %w", err)
579
+
}
580
+
proofBlocks = append(proofBlocks, commitBlk)
581
+
582
+
// read the MST nodes on the path
583
+
for _, nc := range nodeCIDs {
584
+
blk, err := dbs.Get(ctx, nc)
585
+
if err != nil {
586
+
return cid.Undef, fmt.Errorf("reading MST node for proof: %w", err)
587
+
}
588
+
proofBlocks = append(proofBlocks, blk)
589
+
}
590
+
591
+
// read the record block
592
+
recordBlk, err := dbs.Get(ctx, *recordCid)
593
+
if err != nil {
594
+
return cid.Undef, fmt.Errorf("reading record block for proof: %w", err)
595
+
}
596
+
proofBlocks = append(proofBlocks, recordBlk)
407
597
408
-
r, err := repo.OpenRepo(ctx, bs, c)
409
-
if err != nil {
598
+
// read-only, return same root
599
+
return commitCid, nil
600
+
}); err != nil {
410
601
return cid.Undef, nil, err
411
602
}
412
603
413
-
_, _, err = r.GetRecordBytes(ctx, fmt.Sprintf("%s/%s", collection, rkey))
414
-
if err != nil {
415
-
return cid.Undef, nil, err
604
+
return commitCid, proofBlocks, nil
605
+
}
606
+
607
+
func collectPathNodeCIDs(n *mst.Node, key []byte) []cid.Cid {
608
+
if n == nil {
609
+
return nil
610
+
}
611
+
612
+
var cids []cid.Cid
613
+
if n.CID != nil {
614
+
cids = append(cids, *n.CID)
615
+
}
616
+
617
+
height := mst.HeightForKey(key)
618
+
if height >= n.Height {
619
+
// key is at or above this level, no need to descend
620
+
return cids
416
621
}
417
622
418
-
return c, bs.GetReadLog(), nil
623
+
// find the child node that covers this key
624
+
childIdx := -1
625
+
for i, e := range n.Entries {
626
+
if e.IsChild() {
627
+
childIdx = i
628
+
continue
629
+
}
630
+
if e.IsValue() {
631
+
if bytes.Compare(key, e.Key) <= 0 {
632
+
break
633
+
}
634
+
childIdx = -1
635
+
}
636
+
}
637
+
638
+
if childIdx >= 0 && n.Entries[childIdx].Child != nil {
639
+
cids = append(cids, collectPathNodeCIDs(n.Entries[childIdx].Child, key)...)
640
+
}
641
+
642
+
return cids
419
643
}
420
644
421
645
func (rm *RepoMan) incrementBlobRefs(ctx context.Context, urepo models.Repo, cbor []byte) ([]cid.Cid, error) {
+4
-6
test.go
+4
-6
test.go
···
13
13
"github.com/bluesky-social/indigo/atproto/syntax"
14
14
"github.com/bluesky-social/indigo/events"
15
15
"github.com/bluesky-social/indigo/events/schedulers/parallel"
16
+
atp "github.com/bluesky-social/indigo/atproto/repo"
16
17
lexutil "github.com/bluesky-social/indigo/lex/util"
17
-
"github.com/bluesky-social/indigo/repo"
18
18
"github.com/bluesky-social/indigo/repomgr"
19
19
"github.com/gorilla/websocket"
20
20
)
···
82
82
panic(err)
83
83
}
84
84
85
-
rr, err := repo.ReadRepoFromCar(context.TODO(), bytes.NewReader(evt.Blocks))
85
+
_, rr, err := atp.LoadRepoFromCAR(context.TODO(), bytes.NewReader(evt.Blocks))
86
86
if err != nil {
87
87
panic(err)
88
88
}
···
98
98
go func() {
99
99
switch ek {
100
100
case repomgr.EvtKindCreateRecord, repomgr.EvtKindUpdateRecord:
101
-
rc, recordCBOR, err := rr.GetRecordBytes(context.TODO(), op.Path)
101
+
recordCBOR, rc, err := rr.GetRecordBytes(context.TODO(), collection, rkey)
102
102
if err != nil {
103
103
panic(err)
104
104
}
105
105
106
-
if op.Cid == nil || lexutil.LexLink(rc) != *op.Cid {
106
+
if op.Cid == nil || rc == nil || lexutil.LexLink(*rc) != *op.Cid {
107
107
panic("nocid")
108
108
}
109
109
110
-
_ = collection
111
-
_ = rkey
112
110
_ = recordCBOR
113
111
_ = did
114
112