bas.sh
forked from
tangled.org/core
Monorepo for Tangled
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 "use-feed-comment": m.migrateUseFeedComment,
50 "backfill-entity-state": m.backfillEntityState,
51 }
52 return m
53}
54
55func (s *Migration) BackgroundMigrationMiddleware(next http.Handler) http.Handler {
56 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
57 defer next.ServeHTTP(w, r)
58
59 did := s.oauth.GetDidFromCookie(r)
60 if did == "" {
61 return
62 }
63
64 hasPending, err := db.HasPendingPdsRecordMigration(r.Context(), s.db, did)
65 if err != nil || !hasPending {
66 return
67 }
68
69 if _, loaded := s.inflight.LoadOrStore(did, struct{}{}); loaded {
70 return
71 }
72
73 select {
74 case s.sem <- struct{}{}:
75 default:
76 s.inflight.Delete(did)
77 return
78 }
79
80 sessId := s.oauth.GetSessIdFromCookie(r)
81 client, err := s.oauth.AuthorizedClient(r)
82 if err != nil || client.AccountDID == nil {
83 <-s.sem
84 s.inflight.Delete(did)
85 return
86 }
87
88 go func() {
89 defer s.inflight.Delete(did)
90 defer func() { <-s.sem }()
91 s.runPendingMigrations(context.Background(), *client.AccountDID, sessId, client)
92 }()
93 })
94}
95
96func (s *Migration) runPendingMigrations(ctx context.Context, did syntax.DID, sessId string, client *atclient.APIClient) {
97 l := s.logger.With("did", did)
98 migrations, err := db.ListPendingPdsRecordMigrations(ctx, s.db, did)
99 if err != nil {
100 l.Error("failed to query pending migrations", "err", err)
101 return
102 }
103
104 for _, migration := range migrations {
105 if err := s.migrate(ctx, client, sessId, migration); err != nil {
106 l.Error("migration failed", "err", err)
107 }
108 }
109}
110
111func (s *Migration) migrate(ctx context.Context, client *atclient.APIClient, sessId string, migration *models.PDSMigration) error {
112 l := s.logger.With(
113 "name", migration.Name,
114 "aturi", migration.RecordAtUri(),
115 )
116
117 mig, ok := s.migrators[migration.Name]
118 if !ok {
119 return fmt.Errorf("unexpected migration name %s", migration.Name)
120 }
121 err := mig(ctx, client, migration.Did, migration.RecordAtUri())
122
123 if err == nil {
124 l.Info("migrated")
125 migration.Status = models.PDSMigrationStatusDone
126 migration.ErrorMsg = nil
127 migration.RetryCount = 0
128 migration.RetryAfter = 0
129 } else {
130 l.Warn("failed to migrate", "err", err)
131
132 errMsg := strings.ReplaceAll(err.Error(), "\x00", "")
133 migration.ErrorMsg = &errMsg
134 migration.RetryCount++
135
136 if s.onPermAuthErr(ctx, migration.Did, sessId, err) {
137 migration.Status = models.PDSMigrationStatusFailed
138 migration.RetryAfter = 0
139 } else {
140 migration.Status = models.PDSMigrationStatusPending
141 migration.RetryAfter = time.Now().Add(retryBackoff(migration.RetryCount)).Unix()
142 }
143 }
144 if err := db.UpdatePdsRecordMigration(ctx, s.db, migration); err != nil {
145 return fmt.Errorf("failed to update migration status: %w", err)
146 }
147 return nil
148}
149
150func retryBackoff(retries int) time.Duration {
151 return min(time.Duration(retries)*5*time.Second, time.Hour)
152}