alice.pds.demo.boltless.dev
This repository has no description
4.9 kB
193 lines
1package accountmigration
2
3import (
4 "context"
5 "fmt"
6 "log/slog"
7 "sync"
8 "time"
9
10 comatproto "github.com/bluesky-social/indigo/api/atproto"
11 "github.com/bluesky-social/indigo/atproto/atclient"
12 "github.com/bluesky-social/indigo/atproto/syntax"
13 lexutil "github.com/bluesky-social/indigo/lex/util"
14 "github.com/bluesky-social/indigo/xrpc"
15
16 "tangled.org/core/api/tangled"
17 "tangled.org/core/appview/db"
18 "tangled.org/core/appview/models"
19 "tangled.org/core/appview/oauth"
20 "tangled.org/core/xrpc/xrpcclient"
21 "tangled.org/core/orm"
22)
23
24const defaultConcurrency = 4
25
26type Worker struct {
27 db *db.DB
28 oauth *oauth.OAuth
29 dev bool
30 logger *slog.Logger
31 concurrency int
32
33 startMu sync.Mutex
34 started bool
35}
36
37func NewWorker(d *db.DB, o *oauth.OAuth, dev bool, logger *slog.Logger) *Worker {
38 return &Worker{
39 db: d,
40 oauth: o,
41 dev: dev,
42 logger: logger,
43 concurrency: defaultConcurrency,
44 }
45}
46
47func (w *Worker) Start(ctx context.Context) {
48 w.startMu.Lock()
49 defer w.startMu.Unlock()
50 if w.started {
51 return
52 }
53 w.started = true
54
55 for i := 0; i < w.concurrency; i++ {
56 go w.runWorker(ctx, i)
57 }
58}
59
60func (w *Worker) runWorker(ctx context.Context, id int) {
61 l := w.logger.With("worker", id)
62 for {
63 select {
64 case <-ctx.Done():
65 return
66 default:
67 }
68
69 row, ok, err := db.ClaimNextPending(ctx, w.db)
70 if err != nil {
71 l.Error("claim failed", "err", err)
72 time.Sleep(time.Second)
73 continue
74 }
75 if !ok {
76 time.Sleep(time.Second)
77 continue
78 }
79 l.Info("migrating repo", "row", row)
80
81 w.importRepo(ctx, l, row)
82 }
83}
84
85func (w *Worker) importRepo(ctx context.Context, l *slog.Logger, row *db.GitRepoMigration) {
86 l = l.With("id", row.ID, "owner", row.OwnerDid, "name", row.Name, "knot", row.Knot)
87
88 err := w.doImportRepo(ctx, row)
89 if err == nil {
90 if err := db.MarkGitRepoMigrationDone(ctx, w.db, row.ID); err != nil {
91 l.Error("mark done failed", "err", err)
92 }
93 return
94 }
95
96 l.Warn("job failed", "err", err)
97
98 if uerr := db.MarkGitRepoMigrationFailed(ctx, w.db, row.ID, err.Error()); uerr != nil {
99 l.Error("mark failed failed", "err", uerr)
100 }
101}
102
103func (w *Worker) doImportRepo(ctx context.Context, row *db.GitRepoMigration) error {
104 atpClient, err := w.oauth.ClientForDidSession(ctx, row.OwnerDid, row.SessionID)
105 if err != nil {
106 return fmt.Errorf("resume session: %w", err)
107 }
108
109 existing, _ := db.GetRepo(w.db,
110 orm.FilterEq("did", row.OwnerDid),
111 orm.FilterEq("rkey", row.Name),
112 )
113 if existing != nil || rkeyOccupied(ctx, atpClient, row.OwnerDid.String(), row.Name) {
114 return fmt.Errorf("rkey %q already exists", row.Name)
115 }
116
117 sc, err := w.oauth.ServiceClientForDidSession(ctx, row.OwnerDid, row.SessionID,
118 oauth.WithService(row.Knot),
119 oauth.WithLxm(tangled.RepoCreateNSID),
120 oauth.WithDev(w.dev),
121 oauth.WithTimeout(10*time.Minute),
122 )
123 if err != nil {
124 return fmt.Errorf("knot service client: %w", err)
125 }
126
127 defaultBranch := "main"
128 createResp, err := tangled.RepoCreate(ctx, sc, &tangled.RepoCreate_Input{
129 Rkey: row.Name,
130 Name: row.Name,
131 DefaultBranch: &defaultBranch,
132 Source: &row.CloneUrl,
133 })
134 if xrpcerr := xrpcclient.HandleXrpcErr(err); xrpcerr != nil {
135 return fmt.Errorf("knot RepoCreate: %w", err)
136 }
137 if err != nil {
138 return fmt.Errorf("knot RepoCreate: %w", err)
139 }
140 var repoDid string
141 if createResp != nil && createResp.RepoDid != nil {
142 repoDid = *createResp.RepoDid
143 }
144 if repoDid == "" {
145 return fmt.Errorf("knot returned empty repo DID")
146 }
147
148 repo := &models.Repo{
149 Did: row.OwnerDid.String(),
150 Name: row.Name,
151 Knot: row.Knot,
152 Rkey: row.Name,
153 Description: row.Description,
154 Created: time.Now(),
155 RepoDid: repoDid,
156 }
157 record := repo.AsRecord()
158
159 rkey := row.Name
160 _, err = comatproto.RepoCreateRecord(ctx, atpClient, &comatproto.RepoCreateRecord_Input{
161 Collection: tangled.RepoNSID,
162 Repo: row.OwnerDid.String(),
163 Rkey: &rkey,
164 Record: &lexutil.LexiconTypeDecoder{
165 Val: &record,
166 },
167 })
168 if err != nil {
169 w.cleanupKnotRepo(ctx, sc, row.OwnerDid, row.Knot, row.Name)
170 return fmt.Errorf("PDS PutRecord: %w", err)
171 }
172
173 return nil
174}
175
176func (w *Worker) cleanupKnotRepo(ctx context.Context, sc *xrpc.Client, did syntax.DID, knot, name string) {
177 cctx, cancel := context.WithTimeout(ctx, 30*time.Second)
178 defer cancel()
179 if err := tangled.RepoDelete(cctx, sc, &tangled.RepoDelete_Input{
180 Did: did.String(),
181 Name: name,
182 Rkey: name,
183 }); err != nil {
184 w.logger.Error("cleanup: RepoDelete failed", "knot", knot, "name", name, "err", err)
185 }
186}
187
188func rkeyOccupied(ctx context.Context, client *atclient.APIClient, did, rkey string) bool {
189 probeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
190 defer cancel()
191 resp, err := comatproto.RepoGetRecord(probeCtx, client, "", tangled.RepoNSID, did, rkey)
192 return err == nil && resp != nil
193}