at-container-registry/pkg/appview/storage/drain.go

package storage

import (
	"context"
	"encoding/json"
	"log/slog"
	"strings"
	"sync"
	"time"

	"atcr.io/pkg/atproto"
	"atcr.io/pkg/auth"
)

// drainLocks prevents concurrent drain operations per DID.
// If a drain is already running for a user (from login or push), skip.
var drainLocks sync.Map

// MigrateManifestsForSuccessor rewrites manifest records and profile
// when a user's defaultHold has a successor. Best-effort, runs in background.
//
// Steps:
// 1. Get user's sailor profile — check if defaultHold has a successor
// 2. Update profile.DefaultHold from oldHold → newHold
// 3. Walk all io.atcr.manifest records, rewrite holdDid from oldHold → newHold
// 4. Update appview's local manifests table to match
func MigrateManifestsForSuccessor(
	ctx context.Context,
	client *atproto.Client,
	authorizer auth.HoldAuthorizer,
	db HoldDIDLookup,
	did string,
) {
	// Lock per DID — skip if already running
	if _, loaded := drainLocks.LoadOrStore(did, true); loaded {
		return
	}
	defer drainLocks.Delete(did)

	// 1. Get user's profile
	profile, err := GetProfile(ctx, client)
	if err != nil {
		slog.Debug("Drain: failed to get profile", "component", "storage/drain", "did", did, "error", err)
		return
	}
	if profile == nil || profile.DefaultHold == "" {
		return
	}

	// 2. Check if their defaultHold has a successor
	oldHold := profile.DefaultHold
	captain, err := authorizer.GetCaptainRecord(ctx, oldHold)
	if err != nil {
		slog.Debug("Drain: failed to get captain record", "component", "storage/drain", "did", did, "hold", oldHold, "error", err)
		return
	}
	if captain == nil || captain.Successor == "" {
		return // No successor — nothing to drain
	}
	newHold := captain.Successor

	slog.Info("Starting hold drain", "component", "storage/drain", "did", did, "from", oldHold, "to", newHold)

	// 3. Update profile.DefaultHold
	profile.DefaultHold = newHold
	profile.UpdatedAt = time.Now()
	if err := UpdateProfile(ctx, client, profile); err != nil {
		slog.Warn("Drain: failed to update profile", "component", "storage/drain", "did", did, "error", err)
		// Continue — manifest rewrite is still valuable even if profile update fails
	} else {
		slog.Info("Drain: updated profile defaultHold", "component", "storage/drain", "did", did, "newHold", newHold)
	}

	// 4. Walk manifest records, rewrite holdDid
	cursor := ""
	rewritten := 0
	for {
		records, nextCursor, err := client.ListRecordsWithCursor(ctx, atproto.ManifestCollection, 100, cursor)
		if err != nil {
			slog.Warn("Drain: failed to list manifest records", "component", "storage/drain", "did", did, "error", err)
			break
		}

		for _, rec := range records {
			var manifest atproto.ManifestRecord
			if err := json.Unmarshal(rec.Value, &manifest); err != nil {
				slog.Debug("Drain: failed to unmarshal manifest", "component", "storage/drain", "uri", rec.URI, "error", err)
				continue
			}

			// Check if this manifest points to the old hold (via DID or legacy endpoint)
			needsRewrite := false
			if manifest.HoldDID == oldHold {
				needsRewrite = true
			} else if manifest.HoldEndpoint != "" {
				if resolvedDID, resolveErr := atproto.ResolveHoldDID(ctx, manifest.HoldEndpoint); resolveErr == nil && resolvedDID == oldHold {
					needsRewrite = true
				}
			}

			if !needsRewrite {
				continue
			}

			// Rewrite to new hold
			manifest.HoldDID = newHold
			manifest.HoldEndpoint = "" // Clear legacy field

			// Extract rkey from AT URI (at://did/collection/rkey)
			uriParts := strings.Split(rec.URI, "/")
			if len(uriParts) < 2 {
				continue
			}
			rkey := uriParts[len(uriParts)-1]

			if _, err := client.PutRecord(ctx, atproto.ManifestCollection, rkey, &manifest); err != nil {
				slog.Warn("Drain: failed to rewrite manifest", "component", "storage/drain", "uri", rec.URI, "error", err)
				continue
			}
			rewritten++
		}

		if nextCursor == "" {
			break
		}
		cursor = nextCursor
	}

	// 5. Update appview's local manifests table
	if db != nil {
		dbUpdated, err := db.UpdateManifestHoldDID(did, oldHold, newHold)
		if err != nil {
			slog.Warn("Drain: failed to update local DB", "component", "storage/drain", "did", did, "error", err)
		} else if dbUpdated > 0 {
			slog.Info("Drain: updated local DB manifests", "component", "storage/drain", "did", did, "rows", dbUpdated)
		}
	}

	slog.Info("Hold drain complete", "component", "storage/drain", "did", did, "from", oldHold, "to", newHold, "rewritten", rewritten)
}