A fork of the Cocoon PDS but being made more distributed.
com.atproto.repo.deleteRecord (#1)
* deleteRecord
* tweak
* fixes
+126
-17
+55
server/handle_repo_delete_record.go
+55
server/handle_repo_delete_record.go
···
1
+
package server
2
+
3
+
import (
4
+
"github.com/haileyok/cocoon/internal/helpers"
5
+
"github.com/haileyok/cocoon/models"
6
+
"github.com/labstack/echo/v4"
7
+
)
8
+
9
+
type ComAtprotoRepoDeleteRecordRequest struct {
10
+
Repo string `json:"repo" validate:"required,atproto-did"`
11
+
Collection string `json:"collection" validate:"required,atproto-nsid"`
12
+
Rkey string `json:"rkey" validate:"required,atproto-rkey"`
13
+
SwapRecord *string `json:"swapRecord"`
14
+
SwapCommit *string `json:"swapCommit"`
15
+
}
16
+
17
+
func (s *Server) handleDeleteRecord(e echo.Context) error {
18
+
repo := e.Get("repo").(*models.RepoActor)
19
+
20
+
var req ComAtprotoRepoDeleteRecordRequest
21
+
if err := e.Bind(&req); err != nil {
22
+
s.logger.Error("error binding", "error", err)
23
+
return helpers.ServerError(e, nil)
24
+
}
25
+
26
+
if err := e.Validate(req); err != nil {
27
+
s.logger.Error("error validating", "error", err)
28
+
return helpers.InputError(e, nil)
29
+
}
30
+
31
+
if repo.Repo.Did != req.Repo {
32
+
s.logger.Warn("mismatched repo/auth")
33
+
return helpers.InputError(e, nil)
34
+
}
35
+
36
+
results, err := s.repoman.applyWrites(repo.Repo, []Op{
37
+
{
38
+
Type: OpTypeDelete,
39
+
Collection: req.Collection,
40
+
Rkey: &req.Rkey,
41
+
SwapRecord: req.SwapRecord,
42
+
},
43
+
}, req.SwapCommit)
44
+
if err != nil {
45
+
s.logger.Error("error applying writes", "error", err)
46
+
return helpers.ServerError(e, nil)
47
+
}
48
+
49
+
results[0].Type = nil
50
+
results[0].Uri = nil
51
+
results[0].Cid = nil
52
+
results[0].ValidationStatus = nil
53
+
54
+
return e.JSON(200, results[0])
55
+
}
+70
-17
server/repo.go
+70
-17
server/repo.go
···
84
84
85
85
type ApplyWriteResult struct {
86
86
Type *string `json:"$type,omitempty"`
87
-
Uri string `json:"uri"`
88
-
Cid string `json:"cid"`
87
+
Uri *string `json:"uri,omitempty"`
88
+
Cid *string `json:"cid,omitempty"`
89
89
Commit *RepoCommit `json:"commit,omitempty"`
90
-
ValidationStatus *string `json:"validationStatus"`
90
+
ValidationStatus *string `json:"validationStatus,omitempty"`
91
91
}
92
92
93
93
type RepoCommit struct {
···
139
139
})
140
140
results = append(results, ApplyWriteResult{
141
141
Type: to.StringPtr(OpTypeCreate.String()),
142
-
Uri: "at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey,
143
-
Cid: nc.String(),
142
+
Uri: to.StringPtr("at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey),
143
+
Cid: to.StringPtr(nc.String()),
144
144
ValidationStatus: to.StringPtr("valid"), // TODO: obviously this might not be true atm lol
145
145
})
146
146
case OpTypeDelete:
147
+
var old models.Record
148
+
if err := rm.db.Raw("SELECT value FROM records WHERE did = ? AND nsid = ? AND rkey = ?", urepo.Did, op.Collection, op.Rkey).Scan(&old).Error; err != nil {
149
+
return nil, err
150
+
}
151
+
entries = append(entries, models.Record{
152
+
Did: urepo.Did,
153
+
Nsid: op.Collection,
154
+
Rkey: *op.Rkey,
155
+
Value: old.Value,
156
+
})
147
157
err := r.DeleteRecord(context.TODO(), op.Collection+"/"+*op.Rkey)
148
158
if err != nil {
149
159
return nil, err
150
160
}
161
+
results = append(results, ApplyWriteResult{
162
+
Type: to.StringPtr(OpTypeDelete.String()),
163
+
})
151
164
case OpTypeUpdate:
152
165
nc, err := r.UpdateRecord(context.TODO(), op.Collection+"/"+*op.Rkey, op.Record)
153
166
if err != nil {
···
165
178
})
166
179
results = append(results, ApplyWriteResult{
167
180
Type: to.StringPtr(OpTypeUpdate.String()),
168
-
Uri: "at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey,
169
-
Cid: nc.String(),
181
+
Uri: to.StringPtr("at://" + urepo.Did + "/" + op.Collection + "/" + *op.Rkey),
182
+
Cid: to.StringPtr(nc.String()),
170
183
ValidationStatus: to.StringPtr("valid"), // TODO: obviously this might not be true atm lol
171
184
})
172
185
}
···
211
224
})
212
225
213
226
case "del":
227
+
ll := lexutil.LexLink(op.OldCid)
214
228
ops = append(ops, &atproto.SyncSubscribeRepos_RepoOp{
215
229
Action: "delete",
216
230
Path: op.Rpath,
217
231
Cid: nil,
232
+
Prev: &ll,
218
233
})
219
234
}
220
235
···
236
251
237
252
var blobs []lexutil.LexLink
238
253
for _, entry := range entries {
239
-
if err := rm.s.db.Clauses(clause.OnConflict{
240
-
Columns: []clause.Column{{Name: "did"}, {Name: "nsid"}, {Name: "rkey"}},
241
-
UpdateAll: true,
242
-
}).Create(&entry).Error; err != nil {
243
-
return nil, err
244
-
}
254
+
var cids []cid.Cid
255
+
if entry.Cid != "" {
256
+
if err := rm.s.db.Clauses(clause.OnConflict{
257
+
Columns: []clause.Column{{Name: "did"}, {Name: "nsid"}, {Name: "rkey"}},
258
+
UpdateAll: true,
259
+
}).Create(&entry).Error; err != nil {
260
+
return nil, err
261
+
}
245
262
246
-
// we should actually check the type (i.e. delete, create,., update) here but we'll do it later
247
-
cids, err := rm.incrementBlobRefs(urepo, entry.Value)
248
-
if err != nil {
249
-
return nil, err
263
+
cids, err = rm.incrementBlobRefs(urepo, entry.Value)
264
+
if err != nil {
265
+
return nil, err
266
+
}
267
+
} else {
268
+
if err := rm.s.db.Delete(&entry).Error; err != nil {
269
+
return nil, err
270
+
}
271
+
cids, err = rm.decrementBlobRefs(urepo, entry.Value)
272
+
if err != nil {
273
+
return nil, err
274
+
}
250
275
}
251
276
252
277
for _, c := range cids {
···
314
339
for _, c := range cids {
315
340
if err := rm.db.Exec("UPDATE blobs SET ref_count = ref_count + 1 WHERE did = ? AND cid = ?", urepo.Did, c.Bytes()).Error; err != nil {
316
341
return nil, err
342
+
}
343
+
}
344
+
345
+
return cids, nil
346
+
}
347
+
348
+
func (rm *RepoMan) decrementBlobRefs(urepo models.Repo, cbor []byte) ([]cid.Cid, error) {
349
+
cids, err := getBlobCidsFromCbor(cbor)
350
+
if err != nil {
351
+
return nil, err
352
+
}
353
+
354
+
for _, c := range cids {
355
+
var res struct {
356
+
ID uint
357
+
Count int
358
+
}
359
+
if err := rm.db.Raw("UPDATE blobs SET ref_count = ref_count - 1 WHERE did = ? AND cid = ? RETURNING id, ref_count", urepo.Did, c.Bytes()).Scan(&res).Error; err != nil {
360
+
return nil, err
361
+
}
362
+
363
+
if res.Count == 0 {
364
+
if err := rm.db.Exec("DELETE FROM blobs WHERE id = ?", res.ID).Error; err != nil {
365
+
return nil, err
366
+
}
367
+
if err := rm.db.Exec("DELETE FROM blob_parts WHERE blob_id = ?", res.ID).Error; err != nil {
368
+
return nil, err
369
+
}
317
370
}
318
371
}
319
372
+1
server/server.go
+1
server/server.go
···
391
391
// repo
392
392
s.echo.POST("/xrpc/com.atproto.repo.createRecord", s.handleCreateRecord, s.handleSessionMiddleware)
393
393
s.echo.POST("/xrpc/com.atproto.repo.putRecord", s.handlePutRecord, s.handleSessionMiddleware)
394
+
s.echo.POST("/xrpc/com.atproto.repo.deleteRecord", s.handleDeleteRecord, s.handleSessionMiddleware)
394
395
s.echo.POST("/xrpc/com.atproto.repo.applyWrites", s.handleApplyWrites, s.handleSessionMiddleware)
395
396
s.echo.POST("/xrpc/com.atproto.repo.uploadBlob", s.handleRepoUploadBlob, s.handleSessionMiddleware)
396
397