this repo has no description
add missing file
+307
-6
+250
carstore/nonarchive.go
+250
carstore/nonarchive.go
···
1
+
package carstore
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"fmt"
7
+
"io"
8
+
"sync"
9
+
10
+
"github.com/bluesky-social/indigo/models"
11
+
blockformat "github.com/ipfs/go-block-format"
12
+
"github.com/ipfs/go-cid"
13
+
"github.com/ipfs/go-datastore"
14
+
blockstore "github.com/ipfs/go-ipfs-blockstore"
15
+
car "github.com/ipld/go-car"
16
+
"go.opentelemetry.io/otel"
17
+
"gorm.io/gorm"
18
+
"gorm.io/gorm/clause"
19
+
)
20
+
21
+
type NonArchivalCarstore struct {
22
+
db *gorm.DB
23
+
24
+
lk sync.Mutex
25
+
lastCommitCache map[models.Uid]*commitRefInfo
26
+
}
27
+
28
+
func NewNonArchivalCarstore(db *gorm.DB) (*NonArchivalCarstore, error) {
29
+
if err := db.AutoMigrate(&commitRefInfo{}); err != nil {
30
+
return nil, err
31
+
}
32
+
33
+
return &NonArchivalCarstore{
34
+
db: db,
35
+
lastCommitCache: make(map[models.Uid]*commitRefInfo),
36
+
}, nil
37
+
}
38
+
39
+
type commitRefInfo struct {
40
+
ID uint `gorm:"primarykey"`
41
+
Uid models.Uid `gorm:"uniqueIndex"`
42
+
Rev string
43
+
Root models.DbCID
44
+
}
45
+
46
+
func (cs *NonArchivalCarstore) checkLastShardCache(user models.Uid) *commitRefInfo {
47
+
cs.lk.Lock()
48
+
defer cs.lk.Unlock()
49
+
50
+
ls, ok := cs.lastCommitCache[user]
51
+
if ok {
52
+
return ls
53
+
}
54
+
55
+
return nil
56
+
}
57
+
58
+
func (cs *NonArchivalCarstore) removeLastShardCache(user models.Uid) {
59
+
cs.lk.Lock()
60
+
defer cs.lk.Unlock()
61
+
62
+
delete(cs.lastCommitCache, user)
63
+
}
64
+
65
+
func (cs *NonArchivalCarstore) putLastShardCache(ls *commitRefInfo) {
66
+
cs.lk.Lock()
67
+
defer cs.lk.Unlock()
68
+
69
+
cs.lastCommitCache[ls.Uid] = ls
70
+
}
71
+
72
+
func (cs *NonArchivalCarstore) loadCommitRefInfo(ctx context.Context, user models.Uid) (*commitRefInfo, error) {
73
+
var out commitRefInfo
74
+
if err := cs.db.Find(&out, "uid = ?", user).Error; err != nil {
75
+
return nil, err
76
+
}
77
+
78
+
return &out, nil
79
+
}
80
+
81
+
func (cs *NonArchivalCarstore) getCommitRefInfo(ctx context.Context, user models.Uid) (*commitRefInfo, error) {
82
+
ctx, span := otel.Tracer("carstore").Start(ctx, "getCommitRefInfo")
83
+
defer span.End()
84
+
85
+
maybeLs := cs.checkLastShardCache(user)
86
+
if maybeLs != nil {
87
+
return maybeLs, nil
88
+
}
89
+
90
+
lastShard, err := cs.loadCommitRefInfo(ctx, user)
91
+
if err != nil {
92
+
return nil, err
93
+
}
94
+
95
+
cs.putLastShardCache(lastShard)
96
+
return lastShard, nil
97
+
}
98
+
99
+
func (cs *NonArchivalCarstore) updateLastCommit(ctx context.Context, uid models.Uid, rev string, cid cid.Cid) error {
100
+
cri := &commitRefInfo{
101
+
Uid: uid,
102
+
Rev: rev,
103
+
Root: models.DbCID{cid},
104
+
}
105
+
106
+
if err := cs.db.Clauses(clause.OnConflict{
107
+
Columns: []clause.Column{{Name: "uid"}},
108
+
UpdateAll: true,
109
+
}).Create(cri).Error; err != nil {
110
+
return fmt.Errorf("update or set last commit info: %w", err)
111
+
}
112
+
113
+
cs.putLastShardCache(cri)
114
+
115
+
return nil
116
+
}
117
+
118
+
func (cs *NonArchivalCarstore) NewDeltaSession(ctx context.Context, user models.Uid, since *string) (*DeltaSession, error) {
119
+
ctx, span := otel.Tracer("carstore").Start(ctx, "NewSession")
120
+
defer span.End()
121
+
122
+
// TODO: ensure that we don't write updates on top of the wrong head
123
+
// this needs to be a compare and swap type operation
124
+
lastShard, err := cs.getCommitRefInfo(ctx, user)
125
+
if err != nil {
126
+
return nil, err
127
+
}
128
+
129
+
if since != nil && *since != lastShard.Rev {
130
+
return nil, fmt.Errorf("revision mismatch: %s != %s: %w", *since, lastShard.Rev, ErrRepoBaseMismatch)
131
+
}
132
+
133
+
return &DeltaSession{
134
+
fresh: blockstore.NewBlockstore(datastore.NewMapDatastore()),
135
+
blks: make(map[cid.Cid]blockformat.Block),
136
+
base: &userView{
137
+
user: user,
138
+
cs: cs,
139
+
prefetch: true,
140
+
cache: make(map[cid.Cid]blockformat.Block),
141
+
},
142
+
user: user,
143
+
baseCid: lastShard.Root.CID,
144
+
cs: cs,
145
+
seq: 0,
146
+
lastRev: lastShard.Rev,
147
+
}, nil
148
+
}
149
+
150
+
func (cs *NonArchivalCarstore) ReadOnlySession(user models.Uid) (*DeltaSession, error) {
151
+
return &DeltaSession{
152
+
base: &userView{
153
+
user: user,
154
+
cs: cs,
155
+
prefetch: false,
156
+
cache: make(map[cid.Cid]blockformat.Block),
157
+
},
158
+
readonly: true,
159
+
user: user,
160
+
cs: cs,
161
+
}, nil
162
+
}
163
+
164
+
// TODO: incremental is only ever called true, remove the param
165
+
func (cs *NonArchivalCarstore) ReadUserCar(ctx context.Context, user models.Uid, sinceRev string, incremental bool, w io.Writer) error {
166
+
return fmt.Errorf("not supported in non-archival mode")
167
+
}
168
+
169
+
func (cs *NonArchivalCarstore) ImportSlice(ctx context.Context, uid models.Uid, since *string, carslice []byte) (cid.Cid, *DeltaSession, error) {
170
+
ctx, span := otel.Tracer("carstore").Start(ctx, "ImportSlice")
171
+
defer span.End()
172
+
173
+
carr, err := car.NewCarReader(bytes.NewReader(carslice))
174
+
if err != nil {
175
+
return cid.Undef, nil, err
176
+
}
177
+
178
+
if len(carr.Header.Roots) != 1 {
179
+
return cid.Undef, nil, fmt.Errorf("invalid car file, header must have a single root (has %d)", len(carr.Header.Roots))
180
+
}
181
+
182
+
ds, err := cs.NewDeltaSession(ctx, uid, since)
183
+
if err != nil {
184
+
return cid.Undef, nil, fmt.Errorf("new delta session failed: %w", err)
185
+
}
186
+
187
+
var cids []cid.Cid
188
+
for {
189
+
blk, err := carr.Next()
190
+
if err != nil {
191
+
if err == io.EOF {
192
+
break
193
+
}
194
+
return cid.Undef, nil, err
195
+
}
196
+
197
+
cids = append(cids, blk.Cid())
198
+
199
+
if err := ds.Put(ctx, blk); err != nil {
200
+
return cid.Undef, nil, err
201
+
}
202
+
}
203
+
204
+
return carr.Header.Roots[0], ds, nil
205
+
}
206
+
207
+
func (cs *NonArchivalCarstore) GetUserRepoHead(ctx context.Context, user models.Uid) (cid.Cid, error) {
208
+
lastShard, err := cs.getCommitRefInfo(ctx, user)
209
+
if err != nil {
210
+
return cid.Undef, err
211
+
}
212
+
if lastShard.ID == 0 {
213
+
return cid.Undef, nil
214
+
}
215
+
216
+
return lastShard.Root.CID, nil
217
+
}
218
+
219
+
func (cs *NonArchivalCarstore) GetUserRepoRev(ctx context.Context, user models.Uid) (string, error) {
220
+
lastShard, err := cs.getCommitRefInfo(ctx, user)
221
+
if err != nil {
222
+
return "", err
223
+
}
224
+
if lastShard.ID == 0 {
225
+
return "", nil
226
+
}
227
+
228
+
return lastShard.Rev, nil
229
+
}
230
+
231
+
func (cs *NonArchivalCarstore) Stat(ctx context.Context, usr models.Uid) ([]UserStat, error) {
232
+
return nil, nil
233
+
}
234
+
235
+
func (cs *NonArchivalCarstore) WipeUserData(ctx context.Context, user models.Uid) error {
236
+
if err := cs.db.Raw("DELETE from commit_ref_infos WHERE uid = ?", user).Error; err != nil {
237
+
return err
238
+
}
239
+
240
+
cs.removeLastShardCache(user)
241
+
return nil
242
+
}
243
+
244
+
func (cs *NonArchivalCarstore) GetCompactionTargets(ctx context.Context, shardCount int) ([]CompactionTarget, error) {
245
+
return nil, fmt.Errorf("compaction not supported on non-archival")
246
+
}
247
+
248
+
func (cs *NonArchivalCarstore) CompactUserShards(ctx context.Context, user models.Uid, skipBigShards bool) (*CompactionStats, error) {
249
+
return nil, fmt.Errorf("compaction not supported in non-archival")
250
+
}
+1
indexer/indexer.go
+1
indexer/indexer.go
···
538
538
case *bsky.ActorProfile:
539
539
ix.log.Debug("TODO: got actor profile record creation, need to do something with this")
540
540
default:
541
+
log.Warnw("unrecognized record", "record", op.Record, "collection", op.Collection)
541
542
return nil, fmt.Errorf("unrecognized record type (creation): %s", op.Collection)
542
543
}
543
544
+51
-4
repomgr/repomgr.go
+51
-4
repomgr/repomgr.go
···
578
578
return fmt.Errorf("invalid rpath in mst diff, must have collection and rkey")
579
579
}
580
580
581
+
/*
582
+
switch EventKind(op.Action) {
583
+
case EvtKindCreateRecord:
584
+
evtops = append(evtops, RepoOp{
585
+
Kind: EvtKindCreateRecord,
586
+
Collection: parts[0],
587
+
Rkey: parts[1],
588
+
RecCid: (*cid.Cid)(op.Cid),
589
+
})
590
+
case EvtKindUpdateRecord:
591
+
evtops = append(evtops, RepoOp{
592
+
Kind: EvtKindUpdateRecord,
593
+
Collection: parts[0],
594
+
Rkey: parts[1],
595
+
RecCid: (*cid.Cid)(op.Cid),
596
+
})
597
+
case EvtKindDeleteRecord:
598
+
evtops = append(evtops, RepoOp{
599
+
Kind: EvtKindDeleteRecord,
600
+
Collection: parts[0],
601
+
Rkey: parts[1],
602
+
})
603
+
default:
604
+
return fmt.Errorf("unrecognized external user event kind: %q", op.Action)
605
+
}
606
+
*/
581
607
switch EventKind(op.Action) {
582
608
case EvtKindCreateRecord:
583
-
evtops = append(evtops, RepoOp{
609
+
rop := RepoOp{
584
610
Kind: EvtKindCreateRecord,
585
611
Collection: parts[0],
586
612
Rkey: parts[1],
587
613
RecCid: (*cid.Cid)(op.Cid),
588
-
})
614
+
}
615
+
616
+
if rm.hydrateRecords {
617
+
_, rec, err := r.GetRecord(ctx, op.Path)
618
+
if err != nil {
619
+
return fmt.Errorf("reading changed record from car slice: %w", err)
620
+
}
621
+
rop.Record = rec
622
+
}
623
+
624
+
evtops = append(evtops, rop)
589
625
case EvtKindUpdateRecord:
590
-
evtops = append(evtops, RepoOp{
626
+
rop := RepoOp{
591
627
Kind: EvtKindUpdateRecord,
592
628
Collection: parts[0],
593
629
Rkey: parts[1],
594
630
RecCid: (*cid.Cid)(op.Cid),
595
-
})
631
+
}
632
+
633
+
if rm.hydrateRecords {
634
+
_, rec, err := r.GetRecord(ctx, op.Path)
635
+
if err != nil {
636
+
return fmt.Errorf("reading changed record from car slice: %w", err)
637
+
}
638
+
639
+
rop.Record = rec
640
+
}
641
+
642
+
evtops = append(evtops, rop)
596
643
case EvtKindDeleteRecord:
597
644
evtops = append(evtops, RepoOp{
598
645
Kind: EvtKindDeleteRecord,
+2
-1
testing/integ_test.go
+2
-1
testing/integ_test.go
···
536
536
e1 := evts.Next()
537
537
assert.NotNil(e1.RepoCommit)
538
538
assert.Equal(e1.RepoCommit.Repo, bob.DID())
539
+
fmt.Println(e1.RepoCommit.Ops[0])
539
540
540
541
ctx := context.TODO()
541
542
rm := p1.server.Repoman()
···
544
545
}
545
546
546
547
e2 := evts.Next()
547
-
fmt.Println(e2.RepoCommit.Ops)
548
+
//fmt.Println(e2.RepoCommit.Ops[0])
548
549
assert.Equal(len(e2.RepoCommit.Ops), 0)
549
550
assert.Equal(e2.RepoCommit.Repo, bob.DID())
550
551
}
+3
-1
testing/utils.go
+3
-1
testing/utils.go
···
555
555
if err != nil {
556
556
return nil, err
557
557
}
558
-
*/
558
+
//*/
559
+
//*
559
560
cs, err := carstore.NewNonArchivalCarstore(cardb)
560
561
if err != nil {
561
562
return nil, err
562
563
}
564
+
//*/
563
565
564
566
//kmgr := indexer.NewKeyManager(didr, nil)
565
567
kmgr := &bsutil.FakeKeyManager{}