this repo has no description
handle updates and pds migrations (#106)
basic handling for handle updates and pds migrations.
Theres likely more cases to be handled (migrateTo can be nil, what does
that imply?) but this should be an improvement for now.
authored by
Whyrusleeping
and committed by
GitHub
62ac17d8
69574f4a
+233
-7
+3
api/plc.go
+3
api/plc.go
···
121
121
}
122
122
123
123
return opdid, nil
124
+
}
124
125
126
+
func (s *PLCServer) UpdateUserHandle(ctx context.Context, did string, handle string) error {
127
+
return fmt.Errorf("handle updates not yet implemented")
125
128
}
126
129
127
130
func didForCreateOp(op *CreateOp) (string, error) {
+49
-2
bgs/bgs.go
+49
-2
bgs/bgs.go
···
15
15
16
16
"contrib.go.opencensus.io/exporter/prometheus"
17
17
atproto "github.com/bluesky-social/indigo/api/atproto"
18
+
comatproto "github.com/bluesky-social/indigo/api/atproto"
18
19
"github.com/bluesky-social/indigo/blobs"
19
20
"github.com/bluesky-social/indigo/carstore"
20
21
"github.com/bluesky-social/indigo/events"
···
23
24
"github.com/bluesky-social/indigo/models"
24
25
"github.com/bluesky-social/indigo/plc"
25
26
"github.com/bluesky-social/indigo/repomgr"
27
+
"github.com/bluesky-social/indigo/util"
26
28
bsutil "github.com/bluesky-social/indigo/util"
27
29
"github.com/bluesky-social/indigo/xrpc"
28
30
"github.com/gorilla/websocket"
···
373
375
}
374
376
375
377
return nil
378
+
case env.RepoHandle != nil:
379
+
380
+
// TODO: ignoring the data in the message and just going out to the DID doc
381
+
act, err := bgs.createExternalUser(ctx, env.RepoHandle.Did)
382
+
if err != nil {
383
+
return err
384
+
}
385
+
386
+
if act.Handle != env.RepoHandle.Handle {
387
+
log.Warnw("handle update did not update handle to asserted value", "did", env.RepoHandle.Did, "expected", env.RepoHandle.Handle, "actual", act.Handle)
388
+
}
389
+
390
+
return nil
391
+
case env.RepoMigrate != nil:
392
+
if _, err := bgs.createExternalUser(ctx, env.RepoMigrate.Did); err != nil {
393
+
return err
394
+
}
395
+
396
+
return nil
376
397
default:
377
398
return fmt.Errorf("invalid fed event")
378
399
}
···
404
425
return nil
405
426
}
406
427
428
+
// TODO: rename? This also updates users, and 'external' is an old phrasing
407
429
func (s *BGS) createExternalUser(ctx context.Context, did string) (*models.ActorInfo, error) {
408
430
ctx, span := otel.Tracer("bgs").Start(ctx, "createExternalUser")
409
431
defer span.End()
···
438
460
c := &xrpc.Client{Host: durl.String()}
439
461
440
462
if peering.ID == 0 {
441
-
463
+
// TODO: the case of handling a new user on a new PDS probably requires more thought
442
464
cfg, err := atproto.ServerDescribeServer(ctx, c)
443
465
if err != nil {
444
466
// TODO: failing this shouldnt halt our indexing
···
486
508
487
509
exu, err := s.Index.LookupUserByDid(ctx, did)
488
510
if err == nil {
489
-
log.Infof("lost the race to create a new user: %s", did)
511
+
log.Infow("lost the race to create a new user", "did", did, "handle", handle)
512
+
if exu.PDS != peering.ID {
513
+
// User is now on a different PDS, update
514
+
if err := s.db.Model(User{}).Where("id = ?", exu.ID).Update("pds", peering.ID).Error; err != nil {
515
+
return nil, fmt.Errorf("failed to update users pds: %w", err)
516
+
}
517
+
518
+
}
519
+
520
+
if exu.Handle != handle {
521
+
// Users handle has changed, update
522
+
if err := s.db.Model(User{}).Where("id = ?", exu.ID).Update("handle", peering.ID).Error; err != nil {
523
+
return nil, fmt.Errorf("failed to update users handle: %w", err)
524
+
}
525
+
526
+
if err := s.events.AddEvent(ctx, &events.XRPCStreamEvent{
527
+
RepoHandle: &comatproto.SyncSubscribeRepos_Handle{
528
+
Did: exu.Did,
529
+
Handle: handle,
530
+
Time: time.Now().Format(util.ISO8601),
531
+
},
532
+
}); err != nil {
533
+
// TODO: should we really error here? I'm leaning towards no
534
+
return nil, fmt.Errorf("failed to push handle update event: %s", err)
535
+
}
536
+
}
490
537
return exu, nil
491
538
}
492
539
+45
bgs/fedmgr.go
+45
bgs/fedmgr.go
···
175
175
176
176
return nil
177
177
},
178
+
RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error {
179
+
log.Infow("got remote handle update event", "host", host.Host, "did", evt.Did, "handle", evt.Handle)
180
+
if err := s.cb(context.TODO(), host, &events.XRPCStreamEvent{
181
+
RepoHandle: evt,
182
+
}); err != nil {
183
+
log.Errorf("failed handling event from %q (%d): %s", host.Host, evt.Seq, err)
184
+
}
185
+
*lastCursor = evt.Seq
186
+
187
+
if err := s.updateCursor(host, *lastCursor); err != nil {
188
+
return fmt.Errorf("updating cursor: %w", err)
189
+
}
190
+
191
+
return nil
192
+
},
193
+
RepoMigrate: func(evt *comatproto.SyncSubscribeRepos_Migrate) error {
194
+
log.Infow("got remote repo migrate event", "host", host.Host, "did", evt.Did, "migrateTo", evt.MigrateTo)
195
+
if err := s.cb(context.TODO(), host, &events.XRPCStreamEvent{
196
+
RepoMigrate: evt,
197
+
}); err != nil {
198
+
log.Errorf("failed handling event from %q (%d): %s", host.Host, evt.Seq, err)
199
+
}
200
+
*lastCursor = evt.Seq
201
+
202
+
if err := s.updateCursor(host, *lastCursor); err != nil {
203
+
return fmt.Errorf("updating cursor: %w", err)
204
+
}
205
+
206
+
return nil
207
+
},
208
+
RepoTombstone: func(evt *comatproto.SyncSubscribeRepos_Tombstone) error {
209
+
log.Infow("got remote repo tombstone event", "host", host.Host, "did", evt.Did)
210
+
if err := s.cb(context.TODO(), host, &events.XRPCStreamEvent{
211
+
RepoTombstone: evt,
212
+
}); err != nil {
213
+
log.Errorf("failed handling event from %q (%d): %s", host.Host, evt.Seq, err)
214
+
}
215
+
*lastCursor = evt.Seq
216
+
217
+
if err := s.updateCursor(host, *lastCursor); err != nil {
218
+
return fmt.Errorf("updating cursor: %w", err)
219
+
}
220
+
221
+
return nil
222
+
},
178
223
RepoInfo: func(info *comatproto.SyncSubscribeRepos_Info) error {
179
224
log.Infow("info event", "name", info.Name, "message", info.Message, "host", host.Host)
180
225
return nil
+2
-1
cmd/bigsky/main.go
+2
-1
cmd/bigsky/main.go
+11
events/consumer.go
+11
events/consumer.go
···
83
83
} else {
84
84
log.Warnf("received repo commit event with nil commit object (seq %d)", evt.Seq)
85
85
}
86
+
case "#handle":
87
+
var evt comatproto.SyncSubscribeRepos_Handle
88
+
if err := evt.UnmarshalCBOR(r); err != nil {
89
+
return err
90
+
}
91
+
92
+
if cbs.RepoHandle != nil {
93
+
if err := cbs.RepoHandle(&evt); err != nil {
94
+
return err
95
+
}
96
+
}
86
97
case "#info":
87
98
// TODO: this might also be a LabelInfo (as opposed to RepoInfo)
88
99
var evt comatproto.SyncSubscribeRepos_Info
+7
-2
indexer/indexer.go
+7
-2
indexer/indexer.go
···
206
206
207
207
return nil
208
208
}
209
+
209
210
func (ix *Indexer) handleRecordDeleteFeedLike(ctx context.Context, evt *repomgr.RepoEvent, op *repomgr.RepoOp) error {
210
211
var vr models.VoteRecord
211
212
if err := ix.db.Find(&vr, "voter = ? AND rkey = ?", evt.User, op.Rkey).Error; err != nil {
···
705
706
return &ai, nil
706
707
}
707
708
708
-
func (ix *Indexer) lookupUserByHandle(ctx context.Context, handle string) (*models.ActorInfo, error) {
709
+
func (ix *Indexer) LookupUserByHandle(ctx context.Context, handle string) (*models.ActorInfo, error) {
709
710
var ai models.ActorInfo
710
-
if err := ix.db.First(&ai, "handle = ?", handle).Error; err != nil {
711
+
if err := ix.db.Find(&ai, "handle = ?", handle).Error; err != nil {
711
712
return nil, err
713
+
}
714
+
715
+
if ai.ID == 0 {
716
+
return nil, gorm.ErrRecordNotFound
712
717
}
713
718
714
719
return &ai, nil
+1
-1
models/models.go
+1
-1
models/models.go
+10
-1
pds/handlers.go
+10
-1
pds/handlers.go
···
685
685
}
686
686
687
687
func (s *Server) handleComAtprotoIdentityUpdateHandle(ctx context.Context, body *comatprototypes.IdentityUpdateHandle_Input) error {
688
-
panic("nyi")
688
+
if err := s.validateHandle(body.Handle); err != nil {
689
+
return err
690
+
}
691
+
692
+
u, err := s.getUser(ctx)
693
+
if err != nil {
694
+
return err
695
+
}
696
+
697
+
return s.UpdateUserHandle(ctx, u, body.Handle)
689
698
}
690
699
691
700
func (s *Server) handleComAtprotoModerationCreateReport(ctx context.Context, body *comatprototypes.ModerationCreateReport_Input) (*comatprototypes.ModerationCreateReport_Output, error) {
+48
pds/server.go
+48
pds/server.go
···
20
20
"github.com/bluesky-social/indigo/notifs"
21
21
"github.com/bluesky-social/indigo/plc"
22
22
"github.com/bluesky-social/indigo/repomgr"
23
+
"github.com/bluesky-social/indigo/util"
23
24
bsutil "github.com/bluesky-social/indigo/util"
24
25
"github.com/bluesky-social/indigo/xrpc"
25
26
gojwt "github.com/golang-jwt/jwt"
···
30
31
"github.com/labstack/echo/v4/middleware"
31
32
"github.com/lestrrat-go/jwx/v2/jwt"
32
33
"github.com/whyrusleeping/go-did"
34
+
"golang.org/x/xerrors"
33
35
"gorm.io/gorm"
34
36
)
35
37
···
312
314
313
315
e.HTTPErrorHandler = func(err error, ctx echo.Context) {
314
316
fmt.Printf("HANDLER ERROR: (%s) %s\n", ctx.Path(), err)
317
+
318
+
// TODO: need to properly figure out where http error codes for error
319
+
// types get decided. This spot is reasonable, but maybe a bit weird.
320
+
// reviewers, please advise
321
+
if xerrors.Is(err, ErrNoSuchUser) {
322
+
ctx.Response().WriteHeader(404)
323
+
return
324
+
}
325
+
315
326
ctx.Response().WriteHeader(500)
316
327
}
317
328
···
629
640
630
641
return nil
631
642
}
643
+
644
+
func (s *Server) UpdateUserHandle(ctx context.Context, u *User, handle string) error {
645
+
if u.Handle == handle {
646
+
// no change? move on
647
+
log.Warnw("attempted to change handle to current handle", "did", u.Did, "handle", handle)
648
+
return nil
649
+
}
650
+
651
+
_, err := s.indexer.LookupUserByHandle(ctx, handle)
652
+
if err == nil {
653
+
return fmt.Errorf("handle %q is already in use", handle)
654
+
}
655
+
656
+
if err := s.plc.UpdateUserHandle(ctx, u.Did, handle); err != nil {
657
+
return fmt.Errorf("failed to update users handle on plc: %w", err)
658
+
}
659
+
660
+
if err := s.db.Model(models.ActorInfo{}).Where("uid = ?", u.ID).UpdateColumn("handle", handle).Error; err != nil {
661
+
return fmt.Errorf("failed to update handle: %w", err)
662
+
}
663
+
664
+
if err := s.db.Model(User{}).Where("id = ?", u.ID).UpdateColumn("handle", handle).Error; err != nil {
665
+
return fmt.Errorf("failed to update handle: %w", err)
666
+
}
667
+
668
+
if err := s.events.AddEvent(ctx, &events.XRPCStreamEvent{
669
+
RepoHandle: &comatproto.SyncSubscribeRepos_Handle{
670
+
Did: u.Did,
671
+
Handle: handle,
672
+
Time: time.Now().Format(util.ISO8601),
673
+
},
674
+
}); err != nil {
675
+
return fmt.Errorf("failed to push event: %s", err)
676
+
}
677
+
678
+
return nil
679
+
}
+8
plc/fakedid.go
+8
plc/fakedid.go
···
83
83
84
84
return d, nil
85
85
}
86
+
87
+
func (fd *FakeDid) UpdateUserHandle(ctx context.Context, did string, nhandle string) error {
88
+
if err := fd.db.Model(FakeDidMapping{}).Where("did = ?", did).UpdateColumn("handle", nhandle).Error; err != nil {
89
+
return err
90
+
}
91
+
92
+
return nil
93
+
}
+1
plc/plc.go
+1
plc/plc.go
···
9
9
type PLCClient interface {
10
10
DidResolver
11
11
CreateDID(ctx context.Context, sigkey *did.PrivKey, recovery string, handle string, service string) (string, error)
12
+
UpdateUserHandle(ctx context.Context, didstr string, nhandle string) error
12
13
}
13
14
14
15
type DidResolver interface {
+30
testing/integ_test.go
+30
testing/integ_test.go
···
230
230
t.Fatal(err)
231
231
}
232
232
}
233
+
234
+
func TestHandleChange(t *testing.T) {
235
+
//t.Skip("test too sleepy to run in CI for now")
236
+
assert := assert.New(t)
237
+
_ = assert
238
+
didr := testPLC(t)
239
+
p1 := mustSetupPDS(t, "localhost:5385", ".pdsuno", didr)
240
+
p1.Run(t)
241
+
242
+
b1 := mustSetupBGS(t, "localhost:8391", didr)
243
+
b1.Run(t)
244
+
245
+
p1.RequestScraping(t, b1)
246
+
time.Sleep(time.Millisecond * 50)
247
+
248
+
evts := b1.Events(t, -1)
249
+
250
+
u := p1.MustNewUser(t, usernames[0]+".pdsuno")
251
+
252
+
//socialSim(t, []*testUser{u}, 10, 0)
253
+
254
+
u.ChangeHandle(t, "catbear.pdsuno")
255
+
256
+
time.Sleep(time.Millisecond * 100)
257
+
258
+
initevt := evts.Next()
259
+
fmt.Println(initevt.RepoCommit)
260
+
hcevt := evts.Next()
261
+
fmt.Println(hcevt.RepoHandle)
262
+
}
+18
testing/utils.go
+18
testing/utils.go
···
305
305
return resp.Notifications
306
306
}
307
307
308
+
func (u *testUser) ChangeHandle(t *testing.T, nhandle string) {
309
+
t.Helper()
310
+
311
+
ctx := context.TODO()
312
+
if err := atproto.IdentityUpdateHandle(ctx, u.client, &atproto.IdentityUpdateHandle_Input{
313
+
Handle: nhandle,
314
+
}); err != nil {
315
+
t.Fatal(err)
316
+
}
317
+
}
318
+
308
319
func testPLC(t *testing.T) *plc.FakeDid {
309
320
// TODO: just do in memory...
310
321
tdir, err := os.MkdirTemp("", "plcserv")
···
444
455
fmt.Println("received event: ", evt.Seq, evt.Repo)
445
456
es.lk.Lock()
446
457
es.events = append(es.events, &events.XRPCStreamEvent{RepoCommit: evt})
458
+
es.lk.Unlock()
459
+
return nil
460
+
},
461
+
RepoHandle: func(evt *atproto.SyncSubscribeRepos_Handle) error {
462
+
fmt.Println("received handle event: ", evt.Seq, evt.Did)
463
+
es.lk.Lock()
464
+
es.events = append(es.events, &events.XRPCStreamEvent{RepoHandle: evt})
447
465
es.lk.Unlock()
448
466
return nil
449
467
},