alice.pds.demo.boltless.dev
This repository has no description
1package migration
2
3import (
4 "context"
5 "fmt"
6 "log/slog"
7 "net/http"
8 "strings"
9 "sync"
10 "time"
11
12 "github.com/bluesky-social/indigo/atproto/atclient"
13 "github.com/bluesky-social/indigo/atproto/identity"
14 "github.com/bluesky-social/indigo/atproto/syntax"
15
16 "tangled.org/core/appview/db"
17 "tangled.org/core/appview/models"
18 "tangled.org/core/appview/oauth"
19)
20
21const maxConcurrentMigrations = 8
22
23type migrator func(ctx context.Context, client *atclient.APIClient, did syntax.DID, aturi syntax.ATURI) error
24
25type permAuthErrHandler func(ctx context.Context, did syntax.DID, sessId string, err error) bool
26
27type Migration struct {
28 db *db.DB
29 oauth *oauth.OAuth
30 dir identity.Directory
31 logger *slog.Logger
32 inflight sync.Map
33 sem chan struct{}
34 migrators map[string]migrator
35 onPermAuthErr permAuthErrHandler
36}
37
38func NewMigration(db *db.DB, oauth *oauth.OAuth, dir identity.Directory, logger *slog.Logger) *Migration {
39 m := &Migration{
40 db: db,
41 oauth: oauth,
42 dir: dir,
43 logger: logger,
44 sem: make(chan struct{}, maxConcurrentMigrations),
45 onPermAuthErr: oauth.HandlePermanentAuthErr,
46 }
47 m.migrators = map[string]migrator{
48 "add-repo-did": m.migrateAddRepoDid,
49 }
50 return m
51}
52
53func (s *Migration) BackgroundMigrationMiddleware(next http.Handler) http.Handler {
54 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
55 defer next.ServeHTTP(w, r)
56
57 did := s.oauth.GetDidFromCookie(r)
58 if did == "" {
59 return
60 }
61
62 hasPending, err := db.HasPendingPdsRecordMigration(r.Context(), s.db, did)
63 if err != nil || !hasPending {
64 return
65 }
66
67 if _, loaded := s.inflight.LoadOrStore(did, struct{}{}); loaded {
68 return
69 }
70
71 select {
72 case s.sem <- struct{}{}:
73 default:
74 s.inflight.Delete(did)
75 return
76 }
77
78 sessId := s.oauth.GetSessIdFromCookie(r)
79 client, err := s.oauth.AuthorizedClient(r)
80 if err != nil || client.AccountDID == nil {
81 <-s.sem
82 s.inflight.Delete(did)
83 return
84 }
85
86 go func() {
87 defer s.inflight.Delete(did)
88 defer func() { <-s.sem }()
89 s.runPendingMigrations(context.Background(), *client.AccountDID, sessId, client)
90 }()
91 })
92}
93
94func (s *Migration) runPendingMigrations(ctx context.Context, did syntax.DID, sessId string, client *atclient.APIClient) {
95 l := s.logger.With("did", did)
96 migrations, err := db.ListPendingPdsRecordMigrations(ctx, s.db, did)
97 if err != nil {
98 l.Error("failed to query pending migrations", "err", err)
99 return
100 }
101
102 for _, migration := range migrations {
103 if err := s.migrate(ctx, client, sessId, migration); err != nil {
104 l.Error("migration failed", "err", err)
105 }
106 }
107}
108
109func (s *Migration) migrate(ctx context.Context, client *atclient.APIClient, sessId string, migration *models.PDSMigration) error {
110 l := s.logger.With(
111 "name", migration.Name,
112 "aturi", migration.RecordAtUri(),
113 )
114
115 mig, ok := s.migrators[migration.Name]
116 if !ok {
117 return fmt.Errorf("unexpected migration name %s", migration.Name)
118 }
119 err := mig(ctx, client, migration.Did, migration.RecordAtUri())
120
121 if err == nil {
122 l.Info("migrated")
123 migration.Status = models.PDSMigrationStatusDone
124 } else {
125 l.Warn("failed to migrate", "err", err)
126
127 errMsg := strings.ReplaceAll(err.Error(), "\x00", "")
128 migration.ErrorMsg = &errMsg
129 migration.RetryCount++
130
131 if s.onPermAuthErr(ctx, migration.Did, sessId, err) {
132 migration.Status = models.PDSMigrationStatusFailed
133 migration.RetryAfter = 0
134 } else {
135 migration.Status = models.PDSMigrationStatusPending
136 migration.RetryAfter = time.Now().Add(retryBackoff(migration.RetryCount)).Unix()
137 }
138 }
139 if err := db.UpdatePdsRecordMigration(ctx, s.db, migration); err != nil {
140 return fmt.Errorf("failed to update migration status: %w", err)
141 }
142 return nil
143}
144
145func retryBackoff(retries int) time.Duration {
146 return min(time.Duration(retries)*5*time.Second, time.Hour)
147}