appview: background pds data migration #1309

+22

appview/db/db.go

reviewed

··· 1409 1409 return err 1410 1410 }) 1411 1411 1412 1412 + orm.RunMigration(conn, logger, "add-pds-migration", func(tx *sql.Tx) error { 1413 1413 + _, err := tx.Exec(` 1414 1414 + create table if not exists pds_migration ( 1415 1415 + name text not null, 1416 1416 + 1417 1417 + -- record at_uri 1418 1418 + did text not null, 1419 1419 + collection text not null, 1420 1420 + rkey text not null, 1421 1421 + 1422 1422 + status text not null default 'pending', 1423 1423 + error_msg text, 1424 1424 + retry_count integer not null default 0, 1425 1425 + retry_after integer not null default 0, 1426 1426 + updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 1427 1427 + 1428 1428 + unique(name, did, collection, rkey) 1429 1429 + ); 1430 1430 + `) 1431 1431 + return err 1432 1432 + }) 1433 1433 + 1412 1434 return &DB{ 1413 1435 db, 1414 1436 logger,

+89

appview/db/migration.go

reviewed

··· 1 1 + package db 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "time" 6 6 + 7 7 + "github.com/bluesky-social/indigo/atproto/syntax" 8 8 + "tangled.org/core/appview/models" 9 9 + ) 10 10 + 11 11 + // "migration" for records stored in user's PDS, not AppView DB 12 12 + 13 13 + // ListPendingPdsRecordMigrations queries list of pending PDS migrations for given user. 14 14 + // Only pending migrations whose `retry_after` has elapsed are returned. 15 15 + func ListPendingPdsRecordMigrations(ctx context.Context, e Execer, user syntax.DID) ([]*models.PDSMigration, error) { 16 16 + rows, err := e.QueryContext(ctx, 17 17 + `with picked as ( 18 18 + select rowid 19 19 + from pds_migration 20 20 + where did = ? 21 21 + and status = 'pending' 22 22 + and retry_after < ? 23 23 + ) 24 24 + update pds_migration 25 25 + set status = ? 26 26 + where rowid in (select rowid from picked) 27 27 + returning name, did, collection, rkey, status, error_msg, retry_count, retry_after`, 28 28 + user, 29 29 + time.Now().Unix(), 30 30 + models.PDSMigrationStatusRunning, 31 31 + ) 32 32 + if err != nil { 33 33 + return nil, err 34 34 + } 35 35 + defer rows.Close() 36 36 + 37 37 + var migrations []*models.PDSMigration 38 38 + for rows.Next() { 39 39 + var migration models.PDSMigration 40 40 + if err := rows.Scan( 41 41 + &migration.Name, 42 42 + &migration.Did, 43 43 + &migration.Collection, 44 44 + &migration.Rkey, 45 45 + &migration.Status, 46 46 + &migration.ErrorMsg, 47 47 + &migration.RetryCount, 48 48 + &migration.RetryAfter, 49 49 + ); err != nil { 50 50 + return nil, err 51 51 + } 52 52 + migrations = append(migrations, &migration) 53 53 + } 54 54 + if err := rows.Err(); err != nil { 55 55 + return nil, err 56 56 + } 57 57 + 58 58 + return migrations, nil 59 59 + } 60 60 + 61 61 + func EnqueuePdsRecordMigration(ctx context.Context, e Execer, name string, did syntax.DID, collection syntax.NSID, rkey syntax.RecordKey) error { 62 62 + _, err := e.ExecContext(ctx, 63 63 + `insert into pds_migration (name, did, collection, rkey) 64 64 + values (?, ?, ?, ?)`, 65 65 + name, did, collection, rkey, 66 66 + ) 67 67 + return err 68 68 + } 69 69 + 70 70 + func UpdatePdsRecordMigration(ctx context.Context, e Execer, migration *models.PDSMigration) error { 71 71 + _, err := e.ExecContext(ctx, 72 72 + `update pds_migration 73 73 + set status = ?, 74 74 + error_msg = ?, 75 75 + retry_count = ?, 76 76 + retry_after = ?, 77 77 + updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') 78 78 + where name = ? and did = ? and collection = ? and rkey = ?`, 79 79 + migration.Status, 80 80 + migration.ErrorMsg, 81 81 + migration.RetryCount, 82 82 + migration.RetryAfter, 83 83 + migration.Name, 84 84 + migration.Did, 85 85 + migration.Collection, 86 86 + migration.Rkey, 87 87 + ) 88 88 + return err 89 89 + }

+151

appview/migration/migrate_add_repo_did.go

reviewed

··· 1 1 + package migration 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "strings" 7 7 + 8 8 + comatproto "github.com/bluesky-social/indigo/api/atproto" 9 9 + "github.com/bluesky-social/indigo/atproto/atclient" 10 10 + "github.com/bluesky-social/indigo/atproto/syntax" 11 11 + lexutil "github.com/bluesky-social/indigo/lex/util" 12 12 + "tangled.org/core/api/tangled" 13 13 + "tangled.org/core/appview/db" 14 14 + ) 15 15 + 16 16 + func (s *Migration) migrateAddRepoDid(ctx context.Context, client *atclient.APIClient, did syntax.DID, record syntax.ATURI) error { 17 17 + // TODO: use agnostic.RepoGetRecord instead 18 18 + ex, err := comatproto.RepoGetRecord(ctx, client, "", record.Collection().String(), did.String(), record.RecordKey().String()) 19 19 + if err != nil { 20 20 + return fmt.Errorf("pds: %w", err) 21 21 + } 22 22 + 23 23 + val := ex.Value.Val 24 24 + 25 25 + switch record.Collection() { 26 26 + case tangled.RepoNSID: 27 27 + rec, ok := val.(*tangled.Repo) 28 28 + if !ok { 29 29 + return fmt.Errorf("unexpected type for repo record") 30 30 + } 31 31 + repo, err := db.GetRepoByAtUri(s.db, record.String()) 32 32 + if err != nil { 33 33 + return fmt.Errorf("db: failed to query repo: %w", err) 34 34 + } 35 35 + rec.RepoDid = &repo.RepoDid 36 36 + 37 37 + case tangled.RepoIssueNSID: 38 38 + rec, ok := val.(*tangled.RepoIssue) 39 39 + if !ok { 40 40 + return fmt.Errorf("unexpected type for issue record") 41 41 + } 42 42 + if rec.Repo != nil { 43 43 + repoAt := *rec.Repo 44 44 + repo, err := db.GetRepoByAtUri(s.db, repoAt) 45 45 + if err != nil { 46 46 + return fmt.Errorf("db: failed to query repo: %w", err) 47 47 + } 48 48 + rec.RepoDid = &repo.RepoDid 49 49 + } 50 50 + 51 51 + case tangled.RepoPullNSID: 52 52 + rec, ok := val.(*tangled.RepoPull) 53 53 + if !ok { 54 54 + return fmt.Errorf("unexpected type for pull record") 55 55 + } 56 56 + if rec.Target != nil && rec.Target.Repo != nil { 57 57 + repoAt := *rec.Target.Repo 58 58 + repo, err := db.GetRepoByAtUri(s.db, repoAt) 59 59 + if err != nil { 60 60 + return fmt.Errorf("db: failed to query repo: %w", err) 61 61 + } 62 62 + rec.Target.RepoDid = &repo.RepoDid 63 63 + } 64 64 + if rec.Source != nil && rec.Source.Repo != nil { 65 65 + repoAt := *rec.Source.Repo 66 66 + repo, err := db.GetRepoByAtUri(s.db, repoAt) 67 67 + if err != nil { 68 68 + return fmt.Errorf("db: failed to query repo: %w", err) 69 69 + } 70 70 + rec.Source.RepoDid = &repo.RepoDid 71 71 + } 72 72 + 73 73 + case tangled.RepoCollaboratorNSID: 74 74 + rec, ok := val.(*tangled.RepoCollaborator) 75 75 + if !ok { 76 76 + return fmt.Errorf("unexpected type for collaborator record") 77 77 + } 78 78 + if rec.Repo != nil { 79 79 + repoAt := *rec.Repo 80 80 + repo, err := db.GetRepoByAtUri(s.db, repoAt) 81 81 + if err != nil { 82 82 + return fmt.Errorf("db: failed to query repo: %w", err) 83 83 + } 84 84 + rec.RepoDid = &repo.RepoDid 85 85 + } 86 86 + 87 87 + case tangled.RepoArtifactNSID: 88 88 + rec, ok := val.(*tangled.RepoArtifact) 89 89 + if !ok { 90 90 + return fmt.Errorf("unexpected type for artifact record") 91 91 + } 92 92 + if rec.Repo != nil { 93 93 + repoAt := *rec.Repo 94 94 + repo, err := db.GetRepoByAtUri(s.db, repoAt) 95 95 + if err != nil { 96 96 + return fmt.Errorf("db: failed to query repo: %w", err) 97 97 + } 98 98 + rec.RepoDid = &repo.RepoDid 99 99 + } 100 100 + 101 101 + case tangled.FeedStarNSID: 102 102 + rec, ok := val.(*tangled.FeedStar) 103 103 + if !ok { 104 104 + return fmt.Errorf("unexpected type for star record") 105 105 + } 106 106 + if rec.Subject != nil { 107 107 + repoAt := *rec.Subject 108 108 + repo, err := db.GetRepoByAtUri(s.db, repoAt) 109 109 + if err != nil { 110 110 + return fmt.Errorf("db: failed to query repo: %w", err) 111 111 + } 112 112 + rec.SubjectDid = &repo.RepoDid 113 113 + } 114 114 + 115 115 + case tangled.ActorProfileNSID: 116 116 + rec, ok := val.(*tangled.ActorProfile) 117 117 + if !ok { 118 118 + return fmt.Errorf("unexpected type for profile record") 119 119 + } 120 120 + rewritten := make([]string, 0, len(rec.PinnedRepositories)) 121 121 + for _, pin := range rec.PinnedRepositories { 122 122 + if strings.HasPrefix(pin, "did:") { 123 123 + rewritten = append(rewritten, pin) 124 124 + continue 125 125 + } 126 126 + repo, repoErr := db.GetRepoByAtUri(s.db, pin) 127 127 + if repoErr != nil || repo.RepoDid == "" { 128 128 + rewritten = append(rewritten, pin) 129 129 + continue 130 130 + } 131 131 + rewritten = append(rewritten, repo.RepoDid) 132 132 + } 133 133 + rec.PinnedRepositories = rewritten 134 134 + 135 135 + default: 136 136 + return fmt.Errorf("unexpected collection: '%s'", record.Collection()) 137 137 + } 138 138 + 139 139 + _, err = comatproto.RepoPutRecord(ctx, client, &comatproto.RepoPutRecord_Input{ 140 140 + Repo: did.String(), 141 141 + Collection: record.Collection().String(), 142 142 + Rkey: record.RecordKey().String(), 143 143 + SwapRecord: ex.Cid, 144 144 + Record: &lexutil.LexiconTypeDecoder{Val: val}, 145 145 + }) 146 146 + if err != nil { 147 147 + return fmt.Errorf("put record: %w", err) 148 148 + } 149 149 + 150 150 + return nil 151 151 + }

+100

appview/migration/migration.go

reviewed

··· 1 1 + package migration 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "log/slog" 7 7 + "net/http" 8 8 + "strings" 9 9 + "time" 10 10 + 11 11 + "github.com/bluesky-social/indigo/atproto/atclient" 12 12 + "github.com/bluesky-social/indigo/atproto/identity" 13 13 + "github.com/bluesky-social/indigo/atproto/syntax" 14 14 + 15 15 + "tangled.org/core/appview/db" 16 16 + "tangled.org/core/appview/models" 17 17 + "tangled.org/core/appview/oauth" 18 18 + ) 19 19 + 20 20 + type Migration struct { 21 21 + db *db.DB 22 22 + oauth *oauth.OAuth 23 23 + dir identity.Directory 24 24 + logger *slog.Logger 25 25 + } 26 26 + 27 27 + func NewMigration(db *db.DB, oauth *oauth.OAuth, dir identity.Directory, logger *slog.Logger) *Migration { 28 28 + return &Migration{ 29 29 + db, oauth, dir, logger, 30 30 + } 31 31 + } 32 32 + 33 33 + func (s *Migration) BackgroundMigrationMiddleware(next http.Handler) http.Handler { 34 34 + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { 35 35 + defer next.ServeHTTP(w, r) 36 36 + 37 37 + client, err := s.oauth.AuthorizedClient(r) 38 38 + if err != nil { 39 39 + return 40 40 + } 41 41 + if client.AccountDID == nil { 42 42 + return 43 43 + } 44 44 + 45 45 + go s.runPendingMigrations(context.Background(), *client.AccountDID, client) 46 46 + }) 47 47 + } 48 48 + 49 49 + func (s *Migration) runPendingMigrations(ctx context.Context, did syntax.DID, client *atclient.APIClient) { 50 50 + l := s.logger.With("did", did) 51 51 + migrations, err := db.ListPendingPdsRecordMigrations(ctx, s.db, did) 52 52 + if err != nil { 53 53 + l.Error("failed to query pending migrations", "err", err) 54 54 + return 55 55 + } 56 56 + 57 57 + for _, migration := range migrations { 58 58 + if err := s.migrate(ctx, client, migration); err != nil { 59 59 + l.Error("migration failed", "err", err) 60 60 + } 61 61 + } 62 62 + } 63 63 + 64 64 + func (s *Migration) migrate(ctx context.Context, client *atclient.APIClient, migration *models.PDSMigration) error { 65 65 + l := s.logger.With( 66 66 + "name", migration.Name, 67 67 + "aturi", migration.RecordAtUri(), 68 68 + ) 69 69 + 70 70 + var err error 71 71 + switch migration.Name { 72 72 + case "add-repo-did": 73 73 + err = s.migrateAddRepoDid(ctx, client, migration.Did, migration.RecordAtUri()) 74 74 + default: 75 75 + return fmt.Errorf("unexpected migration name %s", migration.Name) 76 76 + } 77 77 + 78 78 + if err == nil { 79 79 + l.Info("migrated") 80 80 + migration.Status = models.PDSMigrationStatusDone 81 81 + } else { 82 82 + l.Warn("failed to migrate", "err", err) 83 83 + 84 84 + errMsg := err.Error() 85 85 + var retryCount = migration.RetryCount + 1 86 86 + var retryAfter = time.Now().Add(3 * time.Second).Unix() 87 87 + 88 88 + // remove null bytes 89 89 + errMsg = strings.ReplaceAll(errMsg, "\x00", "") 90 90 + 91 91 + migration.Status = models.PDSMigrationStatusPending 92 92 + migration.ErrorMsg = &errMsg 93 93 + migration.RetryCount = retryCount 94 94 + migration.RetryAfter = retryAfter 95 95 + } 96 96 + if err := db.UpdatePdsRecordMigration(ctx, s.db, migration); err != nil { 97 97 + return fmt.Errorf("failed to update migration status: %w", err) 98 98 + } 99 99 + return nil 100 100 + }

+37

appview/models/migration.go

reviewed

··· 1 1 + package models 2 2 + 3 3 + import ( 4 4 + "fmt" 5 5 + 6 6 + "github.com/bluesky-social/indigo/atproto/syntax" 7 7 + ) 8 8 + 9 9 + type PDSRecordMigration struct { 10 10 + Did syntax.DID 11 11 + Name string // name of the migration 12 12 + Records []syntax.ATURI // records that need a migration 13 13 + ErrorMsg *string // error message from previous attempt 14 14 + } 15 15 + 16 16 + type PDSMigration struct { 17 17 + Name string // name of the migration 18 18 + Did syntax.DID // record owner 19 19 + Collection syntax.NSID // record collection 20 20 + Rkey syntax.RecordKey // record rkey 21 21 + Status PDSMigrationStatus 22 22 + ErrorMsg *string // error message from previous attempt 23 23 + RetryCount int 24 24 + RetryAfter int64 // Unix timestamp (seconds) 25 25 + } 26 26 + 27 27 + type PDSMigrationStatus string 28 28 + 29 29 + const ( 30 30 + PDSMigrationStatusPending PDSMigrationStatus = "pending" 31 31 + PDSMigrationStatusRunning PDSMigrationStatus = "running" 32 32 + PDSMigrationStatusDone PDSMigrationStatus = "done" 33 33 + ) 34 34 + 35 35 + func (m *PDSMigration) RecordAtUri() syntax.ATURI { 36 36 + return syntax.ATURI(fmt.Sprintf("at://%s/%s/%s", m.Did, m.Collection, m.Rkey)) 37 37 + }

+5

appview/state/router.go

reviewed

··· 12 12 "tangled.org/core/appview/knots" 13 13 "tangled.org/core/appview/labels" 14 14 "tangled.org/core/appview/middleware" 15 15 + "tangled.org/core/appview/migration" 15 16 "tangled.org/core/appview/notifications" 16 17 "tangled.org/core/appview/pipelines" 17 18 "tangled.org/core/appview/pulls" ··· 36 37 s.logger, 37 38 ) 38 39 40 40 + // TODO(boltless): merge this into BackgroundMigrationMiddleware 39 41 router.Use(s.oauth.PdsRewriteMiddleware) 40 42 43 43 + m := migration.NewMigration(s.db, s.oauth, s.idResolver.Directory(), s.logger) 44 44 + router.Use(m.BackgroundMigrationMiddleware) 45 45 + 41 46 router.Get("/pwa-manifest.json", s.WebAppManifest) 42 47 router.Get("/robots.txt", s.RobotsTxt) 43 48 router.Get("/.well-known/security.txt", s.SecurityTxt)