From ba18ec20d332069db4e1c187f28191fb0d9ef2ed Mon Sep 17 00:00:00 2001
From: Lewis <lewis@tangled.org>
Date: Mon, 11 May 2026 23:13:18 +0300
Subject: [PATCH] appview: smarter pds rewrites per req

Lewis: May this revision serve well! <lewis@tangled.org>
---
 appview/db/migration.go        | 18 ++++++++++++
 appview/migration/migration.go | 51 ++++++++++++++++++++++++++++------
 appview/oauth/oauth.go         | 16 +++++++++++
 appview/state/router.go        |  6 ++--
 4 files changed, 79 insertions(+), 12 deletions(-)

diff --git a/appview/db/migration.go b/appview/db/migration.go
index f75c49cd..70b5e4df 100644
--- a/appview/db/migration.go
+++ b/appview/db/migration.go
@@ -58,6 +58,24 @@ func ListPendingPdsRecordMigrations(ctx context.Context, e Execer, user syntax.D
 	return migrations, nil
 }
 
+func HasPendingPdsRecordMigration(ctx context.Context, e Execer, user syntax.DID) (bool, error) {
+	var exists bool
+	err := e.QueryRowContext(ctx,
+		`select exists(
+			select 1 from pds_migration
+			where did = ?
+				and status = 'pending'
+				and retry_after < ?
+		)`,
+		user,
+		time.Now().Unix(),
+	).Scan(&exists)
+	if err != nil {
+		return false, err
+	}
+	return exists, nil
+}
+
 func EnqueuePdsRecordMigration(ctx context.Context, e Execer, name string, did syntax.DID, collection syntax.NSID, rkey syntax.RecordKey) error {
 	_, err := e.ExecContext(ctx,
 		`insert into pds_migration (name, did, collection, rkey)
diff --git a/appview/migration/migration.go b/appview/migration/migration.go
index 4c9eb135..fac4eab9 100644
--- a/appview/migration/migration.go
+++ b/appview/migration/migration.go
@@ -6,6 +6,7 @@ import (
 	"log/slog"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/bluesky-social/indigo/atproto/atclient"
@@ -17,16 +18,24 @@ import (
 	"tangled.org/core/appview/oauth"
 )
 
+const maxConcurrentMigrations = 8
+
 type Migration struct {
-	db     *db.DB
-	oauth  *oauth.OAuth
-	dir    identity.Directory
-	logger *slog.Logger
+	db       *db.DB
+	oauth    *oauth.OAuth
+	dir      identity.Directory
+	logger   *slog.Logger
+	inflight sync.Map
+	sem      chan struct{}
 }
 
 func NewMigration(db *db.DB, oauth *oauth.OAuth, dir identity.Directory, logger *slog.Logger) *Migration {
 	return &Migration{
-		db, oauth, dir, logger,
+		db:     db,
+		oauth:  oauth,
+		dir:    dir,
+		logger: logger,
+		sem:    make(chan struct{}, maxConcurrentMigrations),
 	}
 }
 
@@ -34,15 +43,39 @@ func (s *Migration) BackgroundMigrationMiddleware(next http.Handler) http.Handle
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		defer next.ServeHTTP(w, r)
 
-		client, err := s.oauth.AuthorizedClient(r)
-		if err != nil {
+		did := s.oauth.GetDidFromCookie(r)
+		if did == "" {
 			return
 		}
-		if client.AccountDID == nil {
+
+		hasPending, err := db.HasPendingPdsRecordMigration(r.Context(), s.db, did)
+		if err != nil || !hasPending {
+			return
+		}
+
+		if _, loaded := s.inflight.LoadOrStore(did, struct{}{}); loaded {
+			return
+		}
+
+		select {
+		case s.sem <- struct{}{}:
+		default:
+			s.inflight.Delete(did)
+			return
+		}
+
+		client, err := s.oauth.AuthorizedClient(r)
+		if err != nil || client.AccountDID == nil {
+			<-s.sem
+			s.inflight.Delete(did)
 			return
 		}
 
-		go s.runPendingMigrations(context.Background(), *client.AccountDID, client)
+		go func() {
+			defer s.inflight.Delete(did)
+			defer func() { <-s.sem }()
+			s.runPendingMigrations(context.Background(), *client.AccountDID, client)
+		}()
 	})
 }
 
diff --git a/appview/oauth/oauth.go b/appview/oauth/oauth.go
index 430c89a3..1af124d2 100644
--- a/appview/oauth/oauth.go
+++ b/appview/oauth/oauth.go
@@ -245,6 +245,22 @@ func (o *OAuth) GetDid(r *http.Request) string {
 	return ""
 }
 
+func (o *OAuth) GetDidFromCookie(r *http.Request) syntax.DID {
+	userSession, err := o.SessStore.Get(r, SessionName)
+	if err != nil || userSession.IsNew {
+		return ""
+	}
+	d, ok := userSession.Values[SessionDid].(string)
+	if !ok {
+		return ""
+	}
+	parsed, err := syntax.ParseDID(d)
+	if err != nil {
+		return ""
+	}
+	return parsed
+}
+
 func (o *OAuth) AuthorizedClient(r *http.Request) (*atclient.APIClient, error) {
 	session, err := o.ResumeSession(r)
 	if err != nil {
diff --git a/appview/state/router.go b/appview/state/router.go
index bcf6e519..6c46eb86 100644
--- a/appview/state/router.go
+++ b/appview/state/router.go
@@ -13,7 +13,7 @@ import (
 	"tangled.org/core/appview/labels"
 	"tangled.org/core/appview/metrics"
 	"tangled.org/core/appview/middleware"
-	// "tangled.org/core/appview/migration"
+	"tangled.org/core/appview/migration"
 	"tangled.org/core/appview/notifications"
 	"tangled.org/core/appview/pipelines"
 	"tangled.org/core/appview/pulls"
@@ -41,8 +41,8 @@ func (s *State) Router() http.Handler {
 
 	router.Use(metrics.Middleware)
 
-	// m := migration.NewMigration(s.db, s.oauth, s.idResolver.Directory(), s.logger)
-	// router.Use(m.BackgroundMigrationMiddleware)
+	m := migration.NewMigration(s.db, s.oauth, s.idResolver.Directory(), s.logger)
+	router.Use(m.BackgroundMigrationMiddleware)
 
 	router.Get("/pwa-manifest.json", s.WebAppManifest)
 	router.Get("/robots.txt", s.RobotsTxt)
-- 
2.51.2