this repo has no description
hack: non-archival relay work
+157
-18
+3
bgs/bgs.go
+3
bgs/bgs.go
···
1188
1188
}).Error; err != nil {
1189
1189
return err
1190
1190
}
1191
+
u.SetTombstoned(true)
1191
1192
1192
1193
if err := bgs.db.Model(&models.ActorInfo{}).Where("uid = ?", u.ID).UpdateColumns(map[string]any{
1193
1194
"handle": nil,
···
1416
1417
if err := s.db.Create(&u).Error; err != nil {
1417
1418
return nil, fmt.Errorf("failed to create user after handle conflict: %w", err)
1418
1419
}
1420
+
1421
+
s.userCache.Remove(did)
1419
1422
} else {
1420
1423
return nil, fmt.Errorf("failed to create other pds user: %w", err)
1421
1424
}
+28
-5
carstore/bs.go
+28
-5
carstore/bs.go
···
97
97
}
98
98
99
99
type userView struct {
100
-
cs *FileCarStore
100
+
cs CarStore
101
101
user models.Uid
102
102
103
103
cache map[cid.Cid]blockformat.Block
···
111
111
}
112
112
113
113
func (uv *userView) Has(ctx context.Context, k cid.Cid) (bool, error) {
114
-
return uv.cs.meta.HasUidCid(ctx, uv.user, k)
114
+
_, have := uv.cache[k]
115
+
if have {
116
+
return have, nil
117
+
}
118
+
119
+
fcd, ok := uv.cs.(*FileCarStore)
120
+
if !ok {
121
+
return false, nil
122
+
}
123
+
124
+
return fcd.meta.HasUidCid(ctx, uv.user, k)
115
125
}
116
126
117
127
var CacheHits int64
118
128
var CacheMiss int64
119
129
120
130
func (uv *userView) Get(ctx context.Context, k cid.Cid) (blockformat.Block, error) {
131
+
121
132
if !k.Defined() {
122
133
return nil, fmt.Errorf("attempted to 'get' undefined cid")
123
134
}
···
132
143
}
133
144
atomic.AddInt64(&CacheMiss, 1)
134
145
135
-
path, offset, user, err := uv.cs.meta.LookupBlockRef(ctx, k)
146
+
fcd, ok := uv.cs.(*FileCarStore)
147
+
if !ok {
148
+
return nil, ipld.ErrNotFound{Cid: k}
149
+
}
150
+
151
+
path, offset, user, err := fcd.meta.LookupBlockRef(ctx, k)
136
152
if err != nil {
137
153
return nil, err
138
154
}
···
272
288
baseCid cid.Cid
273
289
seq int
274
290
readonly bool
275
-
cs *FileCarStore
291
+
cs CarStore
276
292
lastRev string
277
293
}
278
294
···
587
603
return nil, fmt.Errorf("cannot write to readonly deltaSession")
588
604
}
589
605
590
-
return ds.cs.writeNewShard(ctx, root, rev, ds.user, ds.seq, ds.blks, ds.rmcids)
606
+
switch ocs := ds.cs.(type) {
607
+
case *FileCarStore:
608
+
return ocs.writeNewShard(ctx, root, rev, ds.user, ds.seq, ds.blks, ds.rmcids)
609
+
case *NonArchivalCarstore:
610
+
return nil, ocs.updateLastCommit(ctx, ds.user, rev, root)
611
+
default:
612
+
return nil, fmt.Errorf("unsupported carstore type")
613
+
}
591
614
}
592
615
593
616
func WriteCarHeader(w io.Writer, root cid.Cid) (int64, error) {
+10
-9
pds/server.go
+10
-9
pds/server.go
···
15
15
16
16
"github.com/bluesky-social/indigo/api/atproto"
17
17
comatproto "github.com/bluesky-social/indigo/api/atproto"
18
-
bsky "github.com/bluesky-social/indigo/api/bsky"
19
18
"github.com/bluesky-social/indigo/carstore"
20
19
"github.com/bluesky-social/indigo/events"
21
20
"github.com/bluesky-social/indigo/indexer"
···
202
201
handle = hurl.Host
203
202
}
204
203
205
-
profile, err := bsky.ActorGetProfile(ctx, c, did)
206
-
if err != nil {
207
-
return nil, err
208
-
}
204
+
/*
205
+
profile, err := bsky.ActorGetProfile(ctx, c, did)
206
+
if err != nil {
207
+
return nil, err
208
+
}
209
209
210
-
if handle != profile.Handle {
211
-
return nil, fmt.Errorf("mismatch in handle between did document and pds profile (%s != %s)", handle, profile.Handle)
212
-
}
210
+
if handle != profile.Handle {
211
+
return nil, fmt.Errorf("mismatch in handle between did document and pds profile (%s != %s)", handle, profile.Handle)
212
+
}
213
+
*/
213
214
214
215
// TODO: request this users info from their server to fill out our data...
215
216
u := User{
···
227
228
subj := &models.ActorInfo{
228
229
Uid: u.ID,
229
230
Handle: sql.NullString{String: handle, Valid: true},
230
-
DisplayName: *profile.DisplayName,
231
+
DisplayName: "missing display name",
231
232
Did: did,
232
233
Type: "",
233
234
PDS: peering.ID,
+8
-1
repomgr/bench_test.go
+8
-1
repomgr/bench_test.go
···
54
54
b.Fatal(err)
55
55
}
56
56
57
-
cs, err := carstore.NewCarStore(cardb, []string{cspath})
57
+
/*
58
+
cs, err := carstore.NewCarStore(cardb, []string{cspath})
59
+
if err != nil {
60
+
b.Fatal(err)
61
+
}
62
+
*/
63
+
cs, err := carstore.NewNonArchivalCarstore(cardb)
58
64
if err != nil {
59
65
b.Fatal(err)
60
66
}
61
67
62
68
repoman := NewRepoManager(cs, &util.FakeKeyManager{})
69
+
repoman.noArchive = true
63
70
64
71
ctx := context.TODO()
65
72
if err := repoman.InitNewActor(ctx, 1, "hello.world", "did:foo:bar", "catdog", "", ""); err != nil {
+8
-1
repomgr/ingest_test.go
+8
-1
repomgr/ingest_test.go
···
80
80
t.Fatal(err)
81
81
}
82
82
83
-
cs, err := carstore.NewCarStore(cardb, []string{cspath})
83
+
/*
84
+
cs, err := carstore.NewCarStore(cardb, []string{cspath})
85
+
if err != nil {
86
+
t.Fatal(err)
87
+
}
88
+
*/
89
+
90
+
cs, err := carstore.NewNonArchivalCarstore(cardb)
84
91
if err != nil {
85
92
t.Fatal(err)
86
93
}
+92
-1
repomgr/repomgr.go
+92
-1
repomgr/repomgr.go
···
34
34
35
35
func NewRepoManager(cs carstore.CarStore, kmgr KeyManager) *RepoManager {
36
36
37
+
var noArchive bool
38
+
if _, ok := cs.(*carstore.NonArchivalCarstore); ok {
39
+
noArchive = true
40
+
}
41
+
37
42
return &RepoManager{
38
43
cs: cs,
39
44
userLocks: make(map[models.Uid]*userLock),
40
45
kmgr: kmgr,
41
46
log: slog.Default().With("system", "repomgr"),
47
+
noArchive: noArchive,
42
48
}
43
49
}
44
50
···
62
68
events func(context.Context, *RepoEvent)
63
69
hydrateRecords bool
64
70
65
-
log *slog.Logger
71
+
log *slog.Logger
72
+
noArchive bool
66
73
}
67
74
68
75
type ActorInfo struct {
···
530
537
}
531
538
532
539
func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
540
+
if rm.noArchive {
541
+
return rm.handleExternalUserEventNoArchive(ctx, pdsid, uid, did, since, nrev, carslice, ops)
542
+
} else {
543
+
return rm.handleExternalUserEventArchive(ctx, pdsid, uid, did, since, nrev, carslice, ops)
544
+
}
545
+
}
546
+
547
+
func (rm *RepoManager) handleExternalUserEventNoArchive(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
548
+
ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
549
+
defer span.End()
550
+
551
+
span.SetAttributes(attribute.Int64("uid", int64(uid)))
552
+
553
+
log.Debugw("HandleExternalUserEvent", "pds", pdsid, "uid", uid, "since", since, "nrev", nrev)
554
+
555
+
unlock := rm.lockUser(ctx, uid)
556
+
defer unlock()
557
+
558
+
start := time.Now()
559
+
root, ds, err := rm.cs.ImportSlice(ctx, uid, since, carslice)
560
+
if err != nil {
561
+
return fmt.Errorf("importing external carslice: %w", err)
562
+
}
563
+
564
+
r, err := repo.OpenRepo(ctx, ds, root)
565
+
if err != nil {
566
+
return fmt.Errorf("opening external user repo (%d, root=%s): %w", uid, root, err)
567
+
}
568
+
569
+
if err := rm.CheckRepoSig(ctx, r, did); err != nil {
570
+
return fmt.Errorf("check repo sig: %w", err)
571
+
}
572
+
openAndSigCheckDuration.Observe(time.Since(start).Seconds())
573
+
574
+
evtops := make([]RepoOp, 0, len(ops))
575
+
for _, op := range ops {
576
+
parts := strings.SplitN(op.Path, "/", 2)
577
+
if len(parts) != 2 {
578
+
return fmt.Errorf("invalid rpath in mst diff, must have collection and rkey")
579
+
}
580
+
581
+
switch EventKind(op.Action) {
582
+
case EvtKindCreateRecord:
583
+
evtops = append(evtops, RepoOp{
584
+
Kind: EvtKindCreateRecord,
585
+
Collection: parts[0],
586
+
Rkey: parts[1],
587
+
RecCid: (*cid.Cid)(op.Cid),
588
+
})
589
+
case EvtKindUpdateRecord:
590
+
evtops = append(evtops, RepoOp{
591
+
Kind: EvtKindUpdateRecord,
592
+
Collection: parts[0],
593
+
Rkey: parts[1],
594
+
RecCid: (*cid.Cid)(op.Cid),
595
+
})
596
+
case EvtKindDeleteRecord:
597
+
evtops = append(evtops, RepoOp{
598
+
Kind: EvtKindDeleteRecord,
599
+
Collection: parts[0],
600
+
Rkey: parts[1],
601
+
})
602
+
default:
603
+
return fmt.Errorf("unrecognized external user event kind: %q", op.Action)
604
+
}
605
+
}
606
+
607
+
if rm.events != nil {
608
+
rm.events(ctx, &RepoEvent{
609
+
User: uid,
610
+
//OldRoot: prev,
611
+
NewRoot: root,
612
+
Rev: nrev,
613
+
Since: since,
614
+
Ops: evtops,
615
+
RepoSlice: carslice,
616
+
PDS: pdsid,
617
+
})
618
+
}
619
+
620
+
return nil
621
+
}
622
+
623
+
func (rm *RepoManager) handleExternalUserEventArchive(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
533
624
ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
534
625
defer span.End()
535
626
+1
testing/integ_test.go
+1
testing/integ_test.go
+7
-1
testing/utils.go
+7
-1
testing/utils.go
···
550
550
return nil, err
551
551
}
552
552
553
-
cs, err := carstore.NewCarStore(cardb, []string{cspath})
553
+
/*
554
+
cs, err := carstore.NewCarStore(cardb, []string{cspath})
555
+
if err != nil {
556
+
return nil, err
557
+
}
558
+
*/
559
+
cs, err := carstore.NewNonArchivalCarstore(cardb)
554
560
if err != nil {
555
561
return nil, err
556
562
}