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 Migration struct {
24 db *db.DB
25 oauth *oauth.OAuth
26 dir identity.Directory
27 logger *slog.Logger
28 inflight sync.Map
29 sem chan struct{}
30}
31
32func NewMigration(db *db.DB, oauth *oauth.OAuth, dir identity.Directory, logger *slog.Logger) *Migration {
33 return &Migration{
34 db: db,
35 oauth: oauth,
36 dir: dir,
37 logger: logger,
38 sem: make(chan struct{}, maxConcurrentMigrations),
39 }
40}
41
42func (s *Migration) BackgroundMigrationMiddleware(next http.Handler) http.Handler {
43 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
44 defer next.ServeHTTP(w, r)
45
46 did := s.oauth.GetDidFromCookie(r)
47 if did == "" {
48 return
49 }
50
51 hasPending, err := db.HasPendingPdsRecordMigration(r.Context(), s.db, did)
52 if err != nil || !hasPending {
53 return
54 }
55
56 if _, loaded := s.inflight.LoadOrStore(did, struct{}{}); loaded {
57 return
58 }
59
60 select {
61 case s.sem <- struct{}{}:
62 default:
63 s.inflight.Delete(did)
64 return
65 }
66
67 client, err := s.oauth.AuthorizedClient(r)
68 if err != nil || client.AccountDID == nil {
69 <-s.sem
70 s.inflight.Delete(did)
71 return
72 }
73
74 go func() {
75 defer s.inflight.Delete(did)
76 defer func() { <-s.sem }()
77 s.runPendingMigrations(context.Background(), *client.AccountDID, client)
78 }()
79 })
80}
81
82func (s *Migration) runPendingMigrations(ctx context.Context, did syntax.DID, client *atclient.APIClient) {
83 l := s.logger.With("did", did)
84 migrations, err := db.ListPendingPdsRecordMigrations(ctx, s.db, did)
85 if err != nil {
86 l.Error("failed to query pending migrations", "err", err)
87 return
88 }
89
90 for _, migration := range migrations {
91 if err := s.migrate(ctx, client, migration); err != nil {
92 l.Error("migration failed", "err", err)
93 }
94 }
95}
96
97func (s *Migration) migrate(ctx context.Context, client *atclient.APIClient, migration *models.PDSMigration) error {
98 l := s.logger.With(
99 "name", migration.Name,
100 "aturi", migration.RecordAtUri(),
101 )
102
103 var err error
104 switch migration.Name {
105 case "add-repo-did":
106 err = s.migrateAddRepoDid(ctx, client, migration.Did, migration.RecordAtUri())
107 default:
108 return fmt.Errorf("unexpected migration name %s", migration.Name)
109 }
110
111 if err == nil {
112 l.Info("migrated")
113 migration.Status = models.PDSMigrationStatusDone
114 } else {
115 l.Warn("failed to migrate", "err", err)
116
117 errMsg := err.Error()
118 var retryCount = migration.RetryCount + 1
119 var retryAfter = time.Now().Add(3 * time.Second).Unix()
120
121 // remove null bytes
122 errMsg = strings.ReplaceAll(errMsg, "\x00", "")
123
124 migration.Status = models.PDSMigrationStatusPending
125 migration.ErrorMsg = &errMsg
126 migration.RetryCount = retryCount
127 migration.RetryAfter = retryAfter
128 }
129 if err := db.UpdatePdsRecordMigration(ctx, s.db, migration); err != nil {
130 return fmt.Errorf("failed to update migration status: %w", err)
131 }
132 return nil
133}