this repo has no description
Fewer database operations for event persistence (#176)
This is based on the rebase PR since they would conflict otherwise.
authored by
Whyrusleeping
and committed by
GitHub
cf3da8ac
4a875abc
+12
-56
+12
-56
events/dbpersist.go
+12
-56
events/dbpersist.go
···
34
34
Event string
35
35
Rebase bool
36
36
37
-
Ops []RepoOpRecord
38
-
}
39
-
40
-
type RepoOpRecord struct {
41
-
ID uint `gorm:"primarykey"`
42
-
RepoEventRecordID uint `gorm:"index"`
43
-
Path string
44
-
Action string
45
-
Rec *util.DbCID
37
+
Ops []byte
46
38
}
47
39
48
40
func NewDbPersistence(db *gorm.DB, cs *carstore.CarStore) (*DbPersistence, error) {
49
41
if err := db.AutoMigrate(&RepoEventRecord{}); err != nil {
50
-
return nil, err
51
-
}
52
-
53
-
if err := db.AutoMigrate(&RepoOpRecord{}); err != nil {
54
42
return nil, err
55
43
}
56
44
···
107
95
Rebase: evt.Rebase,
108
96
}
109
97
110
-
for _, op := range evt.Ops {
111
-
var rec *util.DbCID
112
-
if op.Cid != nil && op.Cid.Defined() {
113
-
rec = &util.DbCID{cid.Cid(*op.Cid)}
114
-
}
115
-
rer.Ops = append(rer.Ops, RepoOpRecord{
116
-
Path: op.Path,
117
-
Action: op.Action,
118
-
Rec: rec,
119
-
})
98
+
opsb, err := json.Marshal(evt.Ops)
99
+
if err != nil {
100
+
return err
120
101
}
102
+
rer.Ops = opsb
103
+
121
104
if err := p.db.Create(&rer).Error; err != nil {
122
105
return err
123
106
}
···
140
123
return err
141
124
}
142
125
143
-
var ops []RepoOpRecord
144
-
if err := p.db.Find(&ops, "repo_event_record_id = ?", evt.Seq).Error; err != nil {
145
-
return err
146
-
}
147
-
148
-
evt.Ops = ops
149
-
150
126
ra, err := p.hydrateRepoEvent(ctx, &evt)
151
127
if err != nil {
152
128
return fmt.Errorf("hydrating event: %w", err)
···
205
181
prevCID = &tmp
206
182
}
207
183
184
+
var ops []*comatproto.SyncSubscribeRepos_RepoOp
185
+
if err := json.Unmarshal(rer.Ops, &ops); err != nil {
186
+
return nil, err
187
+
}
188
+
208
189
out := &comatproto.SyncSubscribeRepos_Commit{
209
190
Seq: int64(rer.Seq),
210
191
Repo: did,
···
213
194
Time: rer.Time.Format(util.ISO8601),
214
195
Blobs: blobCIDs,
215
196
Rebase: rer.Rebase,
197
+
Ops: ops,
216
198
// TODO: there was previously an Event field here. are these all Commit, or are some other events?
217
199
}
218
200
219
-
for _, op := range rer.Ops {
220
-
var recCID *lexutil.LexLink
221
-
if op.Rec != nil {
222
-
tmp := lexutil.LexLink(op.Rec.CID)
223
-
recCID = &tmp
224
-
}
225
-
226
-
out.Ops = append(out.Ops, &comatproto.SyncSubscribeRepos_RepoOp{
227
-
Path: op.Path,
228
-
Action: op.Action,
229
-
Cid: recCID,
230
-
})
231
-
}
232
-
233
201
cs, err := p.readCarSlice(ctx, rer)
234
202
if err != nil {
235
203
return nil, fmt.Errorf("read car slice (%s): %w", rer.Commit.CID, err)
···
264
232
}
265
233
266
234
func (p *DbPersistence) deleteAllEventsForUser(ctx context.Context, usr util.Uid) error {
267
-
for {
268
-
q := p.db.Model(&RepoEventRecord{}).Where("repo = ?", usr).Limit(100).Select("seq")
269
-
res := p.db.Where("repo_event_record_id in (?)", q).Delete(&RepoOpRecord{})
270
-
if err := res.Error; err != nil {
271
-
return fmt.Errorf("failed to delete repo op records: %w", err)
272
-
}
273
-
274
-
if res.RowsAffected == 0 {
275
-
break
276
-
}
277
-
}
278
-
279
235
if err := p.db.Where("repo = ?", usr).Delete(&RepoEventRecord{}).Error; err != nil {
280
236
return err
281
237
}