alice.pds.demo.boltless.dev
This repository has no description
1package migration
2
3import (
4 "context"
5 "fmt"
6
7 "github.com/bluesky-social/indigo/api/agnostic"
8 comatproto "github.com/bluesky-social/indigo/api/atproto"
9 "github.com/bluesky-social/indigo/atproto/atclient"
10 "github.com/bluesky-social/indigo/atproto/syntax"
11 "github.com/bluesky-social/indigo/lex/util"
12 "github.com/bluesky-social/indigo/xrpc"
13 "tangled.org/core/api/tangled"
14 "tangled.org/core/appview/db"
15 "tangled.org/core/orm"
16)
17
18func (s *Migration) migrateUseFeedComment(ctx context.Context, client *atclient.APIClient, did syntax.DID, record syntax.ATURI) error {
19 l := s.logger.With("aturi", record)
20 l.Debug("migrating record")
21
22 switch record.Collection() {
23 case tangled.RepoIssueCommentNSID:
24 case tangled.RepoPullCommentNSID:
25 default:
26 return fmt.Errorf("unexpected collection: '%s'", record.Collection())
27 }
28
29 comment, err := db.GetComment(s.db, orm.FilterEq("at_uri", record))
30 if err != nil {
31 return fmt.Errorf("db: %w", err)
32 }
33
34 comment.Collection = tangled.FeedCommentNSID
35
36 // only update from DB if comment is deleted
37 if comment.Deleted != nil {
38 l.Info("skipping pds migration for deleted record")
39
40 return nil
41 }
42
43 // fill missing reference CIDs
44 if comment.Subject.Cid == "" {
45 cid, err := s.getRecordCid(ctx, syntax.ATURI(comment.Subject.Uri))
46 if err != nil {
47 return fmt.Errorf("pds: getRecordCid for subject.uri: %w", err)
48 }
49 comment.Subject.Cid = cid.String()
50 }
51 if comment.ReplyTo != nil && comment.ReplyTo.Cid == "" {
52 uri, err := syntax.ParseATURI(comment.ReplyTo.Uri)
53 if err != nil {
54 return fmt.Errorf("invalid replyTo.uri: %w", err)
55 }
56
57 // assume parent comment is already migrated to `sh.tangled.feed.comment`.
58 // fail if it isn't ready
59 uri = syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", uri.Authority(), tangled.FeedCommentNSID, uri.RecordKey()))
60
61 cid, err := s.getRecordCid(ctx, uri)
62 if err != nil {
63 return fmt.Errorf("pds: getRecordCid for replyTo.uri: %w", err)
64 }
65 comment.ReplyTo.Uri = uri.String()
66 comment.ReplyTo.Cid = cid.String()
67 }
68
69 // use same rkey for new record
70 rkey := record.RecordKey().String()
71
72 if _, err := comatproto.RepoApplyWrites(ctx, client, &comatproto.RepoApplyWrites_Input{
73 Repo: did.String(),
74 Writes: []*comatproto.RepoApplyWrites_Input_Writes_Elem{
75 {RepoApplyWrites_Delete: &comatproto.RepoApplyWrites_Delete{
76 Collection: record.Collection().String(),
77 Rkey: rkey,
78 }},
79 {RepoApplyWrites_Create: &comatproto.RepoApplyWrites_Create{
80 Collection: tangled.FeedCommentNSID,
81 Rkey: &rkey,
82 Value: &util.LexiconTypeDecoder{Val: comment.AsRecord()},
83 }},
84 },
85 }); err != nil {
86 return fmt.Errorf("pds: applyWrites: %w", err)
87 }
88
89 return nil
90}
91
92func (s *Migration) getRecordCid(ctx context.Context, uri syntax.ATURI) (syntax.CID, error) {
93 ident, err := s.dir.Lookup(ctx, uri.Authority())
94 if err != nil {
95 return "", err
96 }
97
98 xrpcc := xrpc.Client{Host: ident.PDSEndpoint()}
99 out, err := agnostic.RepoGetRecord(ctx, &xrpcc, "", uri.Collection().String(), ident.DID.String(), uri.RecordKey().String())
100 if err != nil {
101 return "", err
102 }
103 if out.Cid == nil {
104 return "", fmt.Errorf("record CID is empty")
105 }
106
107 cid, err := syntax.ParseCID(*out.Cid)
108 if err != nil {
109 return "", err
110 }
111
112 return cid, nil
113}