clean up some functions to use indigo helpers. make repomgr more sync1.1 compliant

go.mod

@@ -14,7 +14,6 @@ require (
 	github.com/did-method-plc/go-didplc v0.0.0-20251009212921-7b7a252b8019
 	github.com/distribution/distribution/v3 v3.1.1
 	github.com/distribution/reference v0.6.0
-	github.com/earthboundkid/versioninfo/v2 v2.24.1
 	github.com/go-chi/chi/v5 v5.2.5
 	github.com/go-chi/render v1.0.3
 	github.com/go-viper/mapstructure/v2 v2.5.0
@@ -102,6 +101,7 @@ require (
 	github.com/docker/go-metrics v0.0.1 // indirect
 	github.com/docker/libtrust v0.0.0-20160708172513-aabc10ec26b7 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/earthboundkid/versioninfo/v2 v2.24.1 // indirect
 	github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect
 	github.com/fatih/color v1.19.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect

pkg/appview/storage/profile.go

@@ -10,6 +10,8 @@ import (
 	"time"
 
 	"atcr.io/pkg/atproto"
+
+	"github.com/bluesky-social/indigo/atproto/syntax"
 )
 
 // ProfileRKey is always "self" per lexicon
@@ -99,7 +101,7 @@ func GetProfile(ctx context.Context, client *atproto.Client) (*atproto.SailorPro
 
 	// Migrate old URL-based defaultHold to DID format
 	// This ensures backward compatibility with profiles created before DID migration
-	if profile.DefaultHold != "" && !atproto.IsDID(profile.DefaultHold) {
+	if _, parseErr := syntax.ParseDID(profile.DefaultHold); profile.DefaultHold != "" && parseErr != nil {
 		// Convert URL to DID by querying /.well-known/atproto-did
 		migratedDID, resolveErr := atproto.ResolveHoldDID(ctx, profile.DefaultHold)
 		if resolveErr != nil {
@@ -143,7 +145,7 @@ func GetProfile(ctx context.Context, client *atproto.Client) (*atproto.SailorPro
 func UpdateProfile(ctx context.Context, client *atproto.Client, profile *atproto.SailorProfileRecord) error {
 	// Normalize defaultHold to DID if it's a URL
 	// This ensures we always store DIDs, even if user provides a URL
-	if profile.DefaultHold != "" && !atproto.IsDID(profile.DefaultHold) {
+	if _, parseErr := syntax.ParseDID(profile.DefaultHold); profile.DefaultHold != "" && parseErr != nil {
 		if resolved, err := atproto.ResolveHoldDID(ctx, profile.DefaultHold); err != nil {
 			slog.Warn("Failed to resolve hold DID during profile update", "component", "profile", "defaultHold", profile.DefaultHold, "error", err)
 		} else {

pkg/atproto/directory.go

@@ -1,14 +1,9 @@
 package atproto
 
 import (
-	"context"
-	"net"
-	"net/http"
 	"sync"
-	"time"
 
 	"github.com/bluesky-social/indigo/atproto/identity"
-	"github.com/earthboundkid/versioninfo/v2"
 )
 
 var (
@@ -54,37 +49,7 @@ func GetDirectory() identity.Directory {
 	directoryMu.Lock()
 	defer directoryMu.Unlock()
 	if sharedDirectory == nil {
-		sharedDirectory = newDefaultDirectory()
+		sharedDirectory = identity.DefaultDirectory()
 	}
 	return sharedDirectory
 }
-
-func newDefaultDirectory() identity.Directory {
-	base := identity.BaseDirectory{
-		PLCURL: identity.DefaultPLCURL,
-		HTTPClient: http.Client{
-			Timeout: time.Second * 10,
-			Transport: &http.Transport{
-				// would want this around 100ms for services doing lots of handle resolution. Impacts PLC connections as well, but not too bad.
-				IdleConnTimeout: time.Millisecond * 1000,
-				MaxIdleConns:    100,
-			},
-		},
-		Resolver: net.Resolver{
-			Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
-				d := net.Dialer{Timeout: time.Second * 3}
-				return d.DialContext(ctx, network, address)
-			},
-		},
-		TryAuthoritativeDNS: true,
-		// primary Bluesky PDS instance only supports HTTP resolution method
-		SkipDNSDomainSuffixes: []string{".bsky.social"},
-		UserAgent:             "indigo-identity/" + versioninfo.Short(),
-	}
-	// Cache configuration:
-	// - capacity: 250,000 entries
-	// - hitTTL: 24 hours (event-driven invalidation via Jetstream provides freshness)
-	// - errTTL: 2 minutes
-	// - invalidHandleTTL: 5 minutes
-	return identity.NewCacheDirectory(&base, 250_000, time.Hour*24, time.Minute*2, time.Minute*5)
-}

pkg/atproto/lexicon.go

@@ -494,11 +494,6 @@ func ParseStarRecordKey(rkey string) (ownerDID, repository string, err error) {
 	return parts[0], parts[1], nil
 }
 
-// IsDID checks if a string is a DID (starts with "did:")
-func IsDID(s string) bool {
-	return len(s) > 4 && s[:4] == "did:"
-}
-
 // RepositoryTagToRKey converts a repository and tag to an ATProto record key
 // ATProto record keys must match: ^[a-zA-Z0-9._~-]{1,512}$
 func RepositoryTagToRKey(repository, tag string) string {

pkg/atproto/lexicon_test.go

@@ -6,8 +6,42 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/bluesky-social/indigo/atproto/syntax"
 )
 
+// TestDIDValidationParity asserts that the set of DIDs we accept/reject as
+// "looks like a DID" matches the syntax.ParseDID behavior we're switching to.
+// Includes the local-dev shapes we must not break.
+func TestDIDValidationParity(t *testing.T) {
+	tests := []struct {
+		name string
+		s    string
+		want bool
+	}{
+		{"valid did:web", "did:web:example.com", true},
+		{"valid did:web local dev percent-encoded port", "did:web:127.0.0.1%3A8000", true},
+		{"valid did:web localhost percent-encoded port", "did:web:localhost%3A8080", true},
+		{"valid did:plc short test fixture", "did:plc:abc123", true},
+		{"valid did:plc real 24-char", "did:plc:pddp4xt5lgnv2qsegbzzs4xg", true},
+		{"valid did:key", "did:key:z6Mkfriq", true},
+		{"reject empty method body", "did:plc:", false},
+		{"reject single-segment did", "did:", false},
+		{"reject no did prefix", "https://example.com", false},
+		{"reject plain text", "hello world", false},
+		{"reject empty", "", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := syntax.ParseDID(tt.s)
+			got := err == nil
+			if got != tt.want {
+				t.Errorf("syntax.ParseDID(%q) accepted=%v, want %v (err=%v)", tt.s, got, tt.want, err)
+			}
+		})
+	}
+}
+
 func TestNewManifestRecord(t *testing.T) {
 	validOCIManifest := `{
 		"schemaVersion": 2,
@@ -659,64 +693,6 @@ func TestParseStarRecordKey_Invalid(t *testing.T) {
 	}
 }
 
-func TestIsDID(t *testing.T) {
-	tests := []struct {
-		name string
-		s    string
-		want bool
-	}{
-		{
-			name: "valid did:web",
-			s:    "did:web:example.com",
-			want: true,
-		},
-		{
-			name: "valid did:plc",
-			s:    "did:plc:abc123",
-			want: true,
-		},
-		{
-			name: "valid did:key",
-			s:    "did:key:z6Mkfriq",
-			want: true,
-		},
-		{
-			name: "not a DID - URL",
-			s:    "https://example.com",
-			want: false,
-		},
-		{
-			name: "not a DID - short string",
-			s:    "did",
-			want: false,
-		},
-		{
-			name: "not a DID - empty",
-			s:    "",
-			want: false,
-		},
-		{
-			name: "not a DID - almost",
-			s:    "did:",
-			want: false,
-		},
-		{
-			name: "not a DID - plain text",
-			s:    "hello world",
-			want: false,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := IsDID(tt.s)
-			if got != tt.want {
-				t.Errorf("IsDID() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
 func TestManifestRecord_JSONSerialization(t *testing.T) {
 	// Create a manifest record
 	ociManifest := `{

pkg/atproto/resolver.go

@@ -44,7 +44,7 @@ func ResolveHoldDID(ctx context.Context, holdIdentifier string) (string, error)
 	}
 
 	// If already a DID, return as-is
-	if IsDID(holdIdentifier) {
+	if _, err := syntax.ParseDID(holdIdentifier); err == nil {
 		return holdIdentifier, nil
 	}
 
@@ -77,7 +77,7 @@ func ResolveHoldDID(ctx context.Context, holdIdentifier string) (string, error)
 	}
 
 	did := strings.TrimSpace(string(body))
-	if !IsDID(did) {
+	if _, err := syntax.ParseDID(did); err != nil {
 		return "", fmt.Errorf("hold at %s returned invalid DID: %q", holdURL, did)
 	}
 

pkg/auth/appview_token.go

@@ -1,77 +1,91 @@
 package auth
 
 import (
-	"crypto/ecdh"
-	"crypto/ecdsa"
-	"crypto/x509"
+	"crypto/rand"
+	"encoding/base64"
 	"fmt"
 	"time"
 
 	"github.com/bluesky-social/indigo/atproto/atcrypto"
+	indigoauth "github.com/bluesky-social/indigo/atproto/auth"
+	"github.com/bluesky-social/indigo/atproto/syntax"
 	"github.com/golang-jwt/jwt/v5"
 )
 
-// CreateAppviewServiceToken creates a short-lived ES256 JWT for appview→hold communication.
-// The token authenticates the appview when calling hold XRPC endpoints like updateCrewTier.
+func randomNonce() string {
+	buf := make([]byte, 16)
+	_, _ = rand.Read(buf)
+	return base64.RawURLEncoding.EncodeToString(buf)
+}
+
+// AppviewLxm is the lexicon method this token binds to. Hold doesn't enforce
+// lxm today (we just trust the signature), but appview sends it for forward
+// compatibility and so the JWT carries an unambiguous purpose.
+//
+// Kept as a typed NSID rather than a string so callers can't mistype it.
+var AppviewLxm = syntax.NSID("io.atcr.hold.updateCrewTier")
+
+// appviewServiceClaims extends RegisteredClaims with the `lxm` claim that
+// indigo's SignServiceAuth uses. We define our own struct because we need to
+// keep Subject (hold's verifier reads it as the acting user DID).
+type appviewServiceClaims struct {
+	jwt.RegisteredClaims
+	LexMethod string `json:"lxm,omitempty"`
+}
+
+// CreateAppviewServiceToken creates a short-lived JWT for appview→hold
+// communication, signed with the appview's private key.
+//
+// Supports both P-256 (ES256) and K-256/secp256k1 (ES256K) keys — the
+// algorithm is selected from the concrete key type.
 //
 // Claims:
-//   - iss: appview DID (e.g. did:web:atcr.io)
-//   - aud: hold DID (e.g. did:web:hold01.atcr.io)
+//   - iss: appview DID
+//   - aud: hold DID
 //   - sub: user DID being acted upon
 //   - exp: now + 60s
 //   - iat: now
-func CreateAppviewServiceToken(privateKey *atcrypto.PrivateKeyP256, appviewDID, holdDID, userDID string) (string, error) {
+//   - jti: random 16-byte nonce (replay prevention)
+//   - lxm: lexicon method this token authorizes
+//
+// Signing uses indigo's atproto JWT signing method so the atcrypto key is
+// used natively (no PKCS8 round-trip).
+func CreateAppviewServiceToken(privateKey atcrypto.PrivateKey, appviewDID, holdDID, userDID string) (string, error) {
+	var alg string
+	switch privateKey.(type) {
+	case *atcrypto.PrivateKeyP256:
+		alg = "ES256"
+	case *atcrypto.PrivateKeyK256:
+		alg = "ES256K"
+	default:
+		return "", fmt.Errorf("unsupported appview key type %T (expected P-256 or K-256)", privateKey)
+	}
+
 	now := time.Now()
-
-	claims := jwt.RegisteredClaims{
-		Issuer:    appviewDID,
-		Audience:  jwt.ClaimStrings{holdDID},
-		Subject:   userDID,
-		ExpiresAt: jwt.NewNumericDate(now.Add(60 * time.Second)),
-		IssuedAt:  jwt.NewNumericDate(now),
+	claims := appviewServiceClaims{
+		RegisteredClaims: jwt.RegisteredClaims{
+			Issuer:    appviewDID,
+			Audience:  jwt.ClaimStrings{holdDID},
+			Subject:   userDID,
+			ExpiresAt: jwt.NewNumericDate(now.Add(60 * time.Second)),
+			IssuedAt:  jwt.NewNumericDate(now),
+			ID:        randomNonce(),
+		},
+		LexMethod: AppviewLxm.String(),
 	}
 
-	token := jwt.NewWithClaims(jwt.SigningMethodES256, claims)
-
-	ecKey, err := P256ToECDSA(privateKey)
-	if err != nil {
-		return "", fmt.Errorf("failed to extract ECDSA key: %w", err)
+	sm := jwt.GetSigningMethod(alg)
+	if sm == nil {
+		return "", fmt.Errorf("%s signing method not registered", alg)
 	}
-
-	signed, err := token.SignedString(ecKey)
+	token := jwt.NewWithClaims(sm, claims)
+	signed, err := token.SignedString(privateKey)
 	if err != nil {
 		return "", fmt.Errorf("failed to sign token: %w", err)
 	}
-
 	return signed, nil
 }
 
-// P256ToECDSA converts an atcrypto P-256 private key to a stdlib *ecdsa.PrivateKey.
-// This is needed because golang-jwt requires stdlib crypto types, while atcrypto
-// wraps them in its own types. We re-parse via PKCS8 encoding round-trip.
-func P256ToECDSA(key *atcrypto.PrivateKeyP256) (*ecdsa.PrivateKey, error) {
-	rawBytes := key.Bytes() // 32-byte raw scalar
-
-	// Parse raw bytes as ecdh key, then convert via PKCS8 round-trip (same as atcrypto does)
-	ecdhKey, err := ecdh.P256().NewPrivateKey(rawBytes)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse P-256 raw bytes: %w", err)
-	}
-
-	pkcs8, err := x509.MarshalPKCS8PrivateKey(ecdhKey)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal PKCS8: %w", err)
-	}
-
-	parsed, err := x509.ParsePKCS8PrivateKey(pkcs8)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse PKCS8: %w", err)
-	}
-
-	ecdsaKey, ok := parsed.(*ecdsa.PrivateKey)
-	if !ok {
-		return nil, fmt.Errorf("parsed key is not ECDSA")
-	}
-
-	return ecdsaKey, nil
-}
+// keep indigoauth referenced so its init() runs and overrides ES256/ES256K
+// signing methods to accept atcrypto.PrivateKey directly.
+var _ = indigoauth.SignServiceAuth

pkg/auth/appview_token_test.go

@@ -0,0 +1,151 @@
+package auth
+
+import (
+	"testing"
+	"time"
+
+	"github.com/bluesky-social/indigo/atproto/atcrypto"
+	"github.com/golang-jwt/jwt/v5"
+)
+
+// TestCreateAppviewServiceToken_Claims locks the claim shape so we can detect
+// regressions when swapping the underlying signer.
+func TestCreateAppviewServiceToken_Claims(t *testing.T) {
+	priv, err := atcrypto.GeneratePrivateKeyP256()
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+
+	const (
+		appviewDID = "did:web:atcr.io"
+		holdDID    = "did:web:hold01.atcr.io"
+		userDID    = "did:plc:pddp4xt5lgnv2qsegbzzs4xg"
+	)
+
+	before := time.Now()
+	tokenStr, err := CreateAppviewServiceToken(priv, appviewDID, holdDID, userDID)
+	if err != nil {
+		t.Fatalf("CreateAppviewServiceToken: %v", err)
+	}
+	after := time.Now()
+
+	parser := jwt.NewParser(jwt.WithoutClaimsValidation())
+	parsed, _, err := parser.ParseUnverified(tokenStr, jwt.MapClaims{})
+	if err != nil {
+		t.Fatalf("parse token: %v", err)
+	}
+
+	if parsed.Method.Alg() != "ES256" {
+		t.Errorf("alg = %q, want ES256", parsed.Method.Alg())
+	}
+
+	claims, ok := parsed.Claims.(jwt.MapClaims)
+	if !ok {
+		t.Fatalf("claims type %T", parsed.Claims)
+	}
+
+	if got, _ := claims["iss"].(string); got != appviewDID {
+		t.Errorf("iss = %q, want %q", got, appviewDID)
+	}
+	aud := claims["aud"]
+	switch a := aud.(type) {
+	case string:
+		if a != holdDID {
+			t.Errorf("aud = %q, want %q", a, holdDID)
+		}
+	case []any:
+		if len(a) != 1 || a[0] != holdDID {
+			t.Errorf("aud = %v, want [%q]", a, holdDID)
+		}
+	default:
+		t.Errorf("aud unexpected type %T = %v", aud, aud)
+	}
+
+	exp, err := claims.GetExpirationTime()
+	if err != nil || exp == nil {
+		t.Fatalf("exp missing: %v", err)
+	}
+	delta := exp.Sub(before)
+	if delta < 55*time.Second || delta > 65*time.Second {
+		t.Errorf("exp delta from sign time = %v, want ~60s", delta)
+	}
+
+	iat, err := claims.GetIssuedAt()
+	if err != nil || iat == nil {
+		t.Fatalf("iat missing: %v", err)
+	}
+	if iat.Before(before.Add(-1*time.Second)) || iat.After(after.Add(1*time.Second)) {
+		t.Errorf("iat = %v, outside sign window [%v, %v]", iat.Time, before, after)
+	}
+
+	if got, _ := claims["sub"].(string); got != userDID {
+		t.Errorf("sub = %q, want %q", got, userDID)
+	}
+	if got, _ := claims["lxm"].(string); got != AppviewLxm.String() {
+		t.Errorf("lxm = %q, want %q", got, AppviewLxm.String())
+	}
+	if got, _ := claims["jti"].(string); got == "" {
+		t.Error("jti claim missing — required for replay protection")
+	}
+}
+
+// TestCreateAppviewServiceToken_K256UsesES256K — when given a secp256k1 key,
+// the token must be signed with ES256K so K-256-only appviews (legacy ATProto
+// PDS keys) can still authenticate to hold.
+func TestCreateAppviewServiceToken_K256UsesES256K(t *testing.T) {
+	priv, err := atcrypto.GeneratePrivateKeyK256()
+	if err != nil {
+		t.Fatalf("generate K-256: %v", err)
+	}
+	tok, err := CreateAppviewServiceToken(priv, "did:web:atcr.io", "did:web:hold01.atcr.io", "did:plc:pddp4xt5lgnv2qsegbzzs4xg")
+	if err != nil {
+		t.Fatalf("sign: %v", err)
+	}
+	parser := jwt.NewParser(jwt.WithoutClaimsValidation())
+	parsed, _, err := parser.ParseUnverified(tok, jwt.MapClaims{})
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if parsed.Method.Alg() != "ES256K" {
+		t.Errorf("alg = %q, want ES256K", parsed.Method.Alg())
+	}
+}
+
+// TestCreateAppviewServiceToken_HasJTI — after the swap to indigo's
+// SignServiceAuth, every token carries a non-empty `jti` (random nonce) so
+// hold can reject replays. Two consecutive tokens get distinct jtis.
+func TestCreateAppviewServiceToken_HasJTI(t *testing.T) {
+	priv, err := atcrypto.GeneratePrivateKeyP256()
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+	const (
+		appviewDID = "did:web:atcr.io"
+		holdDID    = "did:web:hold01.atcr.io"
+		userDID    = "did:plc:pddp4xt5lgnv2qsegbzzs4xg"
+	)
+	parser := jwt.NewParser(jwt.WithoutClaimsValidation())
+
+	jti := func(t *testing.T) string {
+		t.Helper()
+		s, err := CreateAppviewServiceToken(priv, appviewDID, holdDID, userDID)
+		if err != nil {
+			t.Fatalf("sign: %v", err)
+		}
+		tok, _, err := parser.ParseUnverified(s, jwt.MapClaims{})
+		if err != nil {
+			t.Fatalf("parse: %v", err)
+		}
+		claims, _ := tok.Claims.(jwt.MapClaims)
+		val, _ := claims["jti"].(string)
+		return val
+	}
+	a := jti(t)
+	if a == "" {
+		t.Fatal("jti claim missing or empty — needed for replay prevention")
+	}
+	b := jti(t)
+	if a == b {
+		t.Errorf("two tokens share jti %q; expected distinct nonces", a)
+	}
+}

pkg/hold/admin/handlers_settings.go

@@ -6,7 +6,7 @@ import (
 	"net/http"
 	"strings"
 
-	"atcr.io/pkg/atproto"
+	"github.com/bluesky-social/indigo/atproto/syntax"
 	"github.com/spf13/viper"
 )
 
@@ -106,7 +106,7 @@ func (ui *AdminUI) handleSettingsUpdate(w http.ResponseWriter, r *http.Request)
 
 	// Validate successor DID format if provided
 	if successor != "" {
-		if !atproto.IsDID(successor) || (!strings.HasPrefix(successor, "did:web:") && !strings.HasPrefix(successor, "did:plc:")) {
+		if _, err := syntax.ParseDID(successor); err != nil || (!strings.HasPrefix(successor, "did:web:") && !strings.HasPrefix(successor, "did:plc:")) {
 			respond("error", "Successor must be a valid did:web: or did:plc: DID", http.StatusBadRequest)
 			return
 		}

pkg/hold/labeler/subscriber_test.go

@@ -40,8 +40,6 @@ func (s *stubPurger) snapshot() (manifests, users []string) {
 	return append([]string(nil), s.manifestCalls...), append([]string(nil), s.userLevelCalls...)
 }
 
-func ptrBool(b bool) *bool { return &b }
-
 func TestApplyLabelManifestTakedownPurges(t *testing.T) {
 	cache := newTestCache(t)
 	purger := &stubPurger{}
@@ -109,7 +107,7 @@ func TestApplyLabelNegationDropsCacheNoPurge(t *testing.T) {
 		Src: "did:web:labeler.example.com",
 		Uri: uri,
 		Val: TakedownLabelValue,
-		Neg: ptrBool(true),
+		Neg: new(true),
 		Cts: time.Now().UTC().Format(time.RFC3339),
 	})
 

pkg/hold/pds/appview_token_test.go

@@ -0,0 +1,186 @@
+package pds
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+
+	"atcr.io/pkg/atproto"
+	"atcr.io/pkg/auth"
+
+	"github.com/bluesky-social/indigo/atproto/atcrypto"
+)
+
+// appviewTestEnv stands up an httptest server that serves a did:web DID
+// document for a generated keypair, returns the matching appview DID (with
+// percent-encoded port for local-dev), and resets the shared jti cache
+// between tests.
+type appviewTestEnv struct {
+	server     *httptest.Server
+	priv       atcrypto.PrivateKey
+	appviewDID string
+	holdDID    string
+}
+
+func newAppviewTestEnv(t *testing.T) *appviewTestEnv {
+	t.Helper()
+	priv, err := atcrypto.GeneratePrivateKeyP256()
+	if err != nil {
+		t.Fatalf("generate P-256: %v", err)
+	}
+	return newAppviewTestEnvWithKey(t, priv)
+}
+
+func newAppviewTestEnvWithKey(t *testing.T, priv atcrypto.PrivateKey) *appviewTestEnv {
+	t.Helper()
+	pub, err := priv.PublicKey()
+	if err != nil {
+		t.Fatalf("public key: %v", err)
+	}
+
+	doc := map[string]any{
+		"@context": []string{"https://www.w3.org/ns/did/v1"},
+		"verificationMethod": []map[string]string{{
+			"id":                 "#atproto",
+			"type":               "Multikey",
+			"publicKeyMultibase": pub.Multibase(),
+		}},
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/.well-known/did.json", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(doc)
+	})
+	server := httptest.NewServer(mux)
+	t.Cleanup(server.Close)
+
+	u, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatalf("parse server URL: %v", err)
+	}
+	// did:web requires percent-encoded ":" for ports
+	appviewDID := "did:web:" + strings.ReplaceAll(u.Host, ":", "%3A")
+
+	// fetchAppviewPublicKey downgrades to http when test mode is on
+	atproto.SetTestMode(true)
+	t.Cleanup(func() { atproto.SetTestMode(false) })
+
+	// Reset shared jti cache so other tests don't pollute this one and vice versa.
+	*sharedAppviewJTICache = *newJTIReplayCache()
+	t.Cleanup(func() { *sharedAppviewJTICache = *newJTIReplayCache() })
+
+	return &appviewTestEnv{
+		server:     server,
+		priv:       priv,
+		appviewDID: appviewDID,
+		holdDID:    "did:web:hold01.atcr.io",
+	}
+}
+
+func (e *appviewTestEnv) newRequestWithToken(t *testing.T, token string) *http.Request {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodPost, "/test", nil)
+	req.Header.Set("Authorization", "Bearer "+token)
+	return req
+}
+
+func (e *appviewTestEnv) sign(t *testing.T, userDID string) string {
+	t.Helper()
+	tok, err := auth.CreateAppviewServiceToken(e.priv, e.appviewDID, e.holdDID, userDID)
+	if err != nil {
+		t.Fatalf("CreateAppviewServiceToken: %v", err)
+	}
+	return tok
+}
+
+func TestValidateAppviewToken_AcceptsK256SignedToken(t *testing.T) {
+	priv, err := atcrypto.GeneratePrivateKeyK256()
+	if err != nil {
+		t.Fatalf("generate K-256: %v", err)
+	}
+	env := newAppviewTestEnvWithKey(t, priv)
+	tok := env.sign(t, "did:plc:pddp4xt5lgnv2qsegbzzs4xg")
+
+	got, err := ValidateAppviewToken(env.newRequestWithToken(t, tok), env.appviewDID, env.holdDID)
+	if err != nil {
+		t.Fatalf("ValidateAppviewToken (K-256): %v", err)
+	}
+	if got != "did:plc:pddp4xt5lgnv2qsegbzzs4xg" {
+		t.Errorf("returned sub = %q", got)
+	}
+}
+
+func TestValidateAppviewToken_AcceptsFreshToken(t *testing.T) {
+	env := newAppviewTestEnv(t)
+	tok := env.sign(t, "did:plc:pddp4xt5lgnv2qsegbzzs4xg")
+
+	got, err := ValidateAppviewToken(env.newRequestWithToken(t, tok), env.appviewDID, env.holdDID)
+	if err != nil {
+		t.Fatalf("ValidateAppviewToken: %v", err)
+	}
+	if got != "did:plc:pddp4xt5lgnv2qsegbzzs4xg" {
+		t.Errorf("returned sub = %q", got)
+	}
+}
+
+func TestValidateAppviewToken_RejectsReplayedJTI(t *testing.T) {
+	env := newAppviewTestEnv(t)
+	tok := env.sign(t, "did:plc:pddp4xt5lgnv2qsegbzzs4xg")
+
+	if _, err := ValidateAppviewToken(env.newRequestWithToken(t, tok), env.appviewDID, env.holdDID); err != nil {
+		t.Fatalf("first validate failed: %v", err)
+	}
+	_, err := ValidateAppviewToken(env.newRequestWithToken(t, tok), env.appviewDID, env.holdDID)
+	if err == nil {
+		t.Fatal("expected replay rejection on second submission, got nil")
+	}
+	if !strings.Contains(err.Error(), "replay") && err != ErrTokenReplayed {
+		t.Errorf("error %v does not mention replay", err)
+	}
+}
+
+// TODO: re-enable when jti becomes mandatory. Today the hold verifier
+// accepts tokens without a jti claim (with a warning) so a partial appview
+// rollout doesn't break tier updates. Once all appview deployments are
+// confirmed to emit jti, restore this test and the corresponding
+// ErrMissingJTIClaim check in ValidateAppviewToken.
+//
+// func TestValidateAppviewToken_RejectsMissingJTI(t *testing.T) {
+// 	env := newAppviewTestEnv(t)
+// 	header := `{"alg":"ES256","typ":"JWT"}`
+// 	now := time.Now().Unix()
+// 	body := fmt.Sprintf(`{"iss":%q,"aud":[%q],"sub":%q,"exp":%d,"iat":%d}`,
+// 		env.appviewDID, env.holdDID, "did:plc:pddp4xt5lgnv2qsegbzzs4xg", now+60, now)
+// 	h := base64.RawURLEncoding.EncodeToString([]byte(header))
+// 	p := base64.RawURLEncoding.EncodeToString([]byte(body))
+// 	signingInput := h + "." + p
+// 	sig, err := env.priv.HashAndSign([]byte(signingInput))
+// 	if err != nil {
+// 		t.Fatalf("sign: %v", err)
+// 	}
+// 	tok := signingInput + "." + base64.RawURLEncoding.EncodeToString(sig)
+//
+// 	_, err = ValidateAppviewToken(env.newRequestWithToken(t, tok), env.appviewDID, env.holdDID)
+// 	if !errors.Is(err, ErrMissingJTIClaim) {
+// 		t.Errorf("expected ErrMissingJTIClaim, got %v", err)
+// 	}
+// }
+
+func TestValidateAppviewToken_AcceptsDistinctJTIs(t *testing.T) {
+	env := newAppviewTestEnv(t)
+	tokA := env.sign(t, "did:plc:pddp4xt5lgnv2qsegbzzs4xg")
+	tokB := env.sign(t, "did:plc:pddp4xt5lgnv2qsegbzzs4xg")
+	if tokA == tokB {
+		t.Fatal("two signed tokens are byte-identical")
+	}
+	if _, err := ValidateAppviewToken(env.newRequestWithToken(t, tokA), env.appviewDID, env.holdDID); err != nil {
+		t.Fatalf("validate A: %v", err)
+	}
+	if _, err := ValidateAppviewToken(env.newRequestWithToken(t, tokB), env.appviewDID, env.holdDID); err != nil {
+		t.Errorf("validate B (distinct jti) rejected: %v", err)
+	}
+}

pkg/hold/pds/auth.go

@@ -34,8 +34,14 @@ var (
 	ErrMissingISSClaim  = errors.New("missing 'iss' claim in token")
 	ErrMissingSubClaim  = errors.New("missing 'sub' claim in token")
 	ErrTokenExpired     = errors.New("token has expired")
+	ErrTokenReplayed    = errors.New("token jti already seen (replay)")
 )
 
+// sharedAppviewJTICache deduplicates appview JTIs across all
+// ValidateAppviewToken calls in this process. A per-process cache is enough
+// because every appview-issued token names a single hold as its audience.
+var sharedAppviewJTICache = newJTIReplayCache()
+
 // AuthError provides structured authorization error information
 type AuthError struct {
 	Action   string   // The action being attempted: "blob:read", "blob:write", "crew:admin"
@@ -678,13 +684,16 @@ func ValidateAppviewToken(r *http.Request, appviewDID, holdDID string) (string,
 		return "", ErrMissingSubClaim
 	}
 
-	// Fetch P-256 public key from appview DID document
-	pubKey, err := fetchP256PublicKeyFromDID(r.Context(), appviewDID)
+	// Fetch appview public key from its DID document. Either P-256 (ES256)
+	// or K-256 (ES256K) is accepted — the curve is encoded in the multikey
+	// prefix, and HashAndVerifyLenient rejects signatures from the wrong
+	// curve automatically.
+	pubKey, err := fetchAppviewPublicKey(r.Context(), appviewDID)
 	if err != nil {
 		return "", fmt.Errorf("failed to fetch appview public key: %w", err)
 	}
 
-	// Verify JWT signature with P-256 key
+	// Verify JWT signature
 	signedData := []byte(tokenParts[0] + "." + tokenParts[1])
 	signature, err := base64.RawURLEncoding.DecodeString(tokenParts[2])
 	if err != nil {
@@ -695,18 +704,30 @@ func ValidateAppviewToken(r *http.Request, appviewDID, holdDID string) (string,
 		return "", fmt.Errorf("signature verification failed: %w", err)
 	}
 
+	// Replay check: when `jti` is present, refuse to honour the same one
+	// twice within its TTL.
+	// TODO: hard-require jti once all appview deployments are confirmed to
+	// emit it. For now we accept missing jti with a warning so a partial
+	// rollout doesn't break tier updates.
+	if claims.ID == "" {
+		slog.Warn("Appview token missing jti claim; accepting for now",
+			"appviewDID", appviewDID, "userDID", subject)
+	} else if exp != nil && sharedAppviewJTICache.Seen(claims.ID, exp.Time) {
+		return "", ErrTokenReplayed
+	}
+
 	slog.Debug("Validated appview service token", "appviewDID", appviewDID, "userDID", subject)
 	return subject, nil
 }
 
-// fetchP256PublicKeyFromDID fetches a P-256 public key from a did:web DID document.
-// It resolves the DID document and looks for a Multikey verification method with P-256 prefix.
-func fetchP256PublicKeyFromDID(ctx context.Context, did string) (*atcrypto.PublicKeyP256, error) {
+// fetchAppviewPublicKey fetches the appview's verification key from its
+// did:web DID document. Accepts any Multikey atcrypto can parse (P-256 or
+// K-256). Returns the first one found.
+func fetchAppviewPublicKey(ctx context.Context, did string) (atcrypto.PublicKey, error) {
 	if !strings.HasPrefix(did, "did:web:") {
 		return nil, fmt.Errorf("only did:web is supported for appview DID, got %s", did)
 	}
 
-	// Resolve did:web to URL
 	host := strings.TrimPrefix(did, "did:web:")
 	host = strings.ReplaceAll(host, "%3A", ":")
 	scheme := "https"
@@ -741,25 +762,18 @@ func fetchP256PublicKeyFromDID(ctx context.Context, did string) (*atcrypto.Publi
 		return nil, fmt.Errorf("failed to decode DID document: %w", err)
 	}
 
-	// Find a Multikey verification method with P-256 public key
 	for _, vm := range doc.VerificationMethod {
 		if vm.Type != "Multikey" || vm.PublicKeyMultibase == "" {
 			continue
 		}
-
-		// Try parsing as P-256 key via atcrypto's multibase parser
 		pubKey, err := atcrypto.ParsePublicMultibase(vm.PublicKeyMultibase)
 		if err != nil {
 			continue
 		}
-
-		p256Key, ok := pubKey.(*atcrypto.PublicKeyP256)
-		if ok {
-			return p256Key, nil
-		}
+		return pubKey, nil
 	}
 
-	return nil, fmt.Errorf("no P-256 public key found in DID document for %s", did)
+	return nil, fmt.Errorf("no Multikey verification method found in DID document for %s", did)
 }
 
 // fetchPublicKeyFromDID fetches the public key from a DID document

pkg/hold/pds/events.go

@@ -6,6 +6,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log/slog"
 	"strings"
 	"sync"
@@ -35,10 +36,12 @@ type EventBroadcaster struct {
 	ownsDB       bool    // true when this broadcaster opened the connection itself
 }
 
-// Subscriber represents a WebSocket client subscribed to the firehose
+// Subscriber represents a WebSocket client subscribed to the firehose.
+// The send channel carries any subscribeRepos message type (commit, identity, …)
+// so the writer loop can serialize all of them uniformly.
 type Subscriber struct {
 	conn   *websocket.Conn
-	send   chan *RepoCommitEvent
+	send   chan firehoseMsg
 	cursor int64 // Last sequence number this subscriber has seen
 }
 
@@ -48,6 +51,38 @@ type HistoricalEvent struct {
 	Event *RepoCommitEvent
 }
 
+// firehoseMsg is anything we can hand to a subscriber's writer goroutine —
+// commits, identity events, etc. Each implementation knows its #type header,
+// its sequence number, and how to CBOR-marshal its body to the wire.
+type firehoseMsg interface {
+	msgType() string
+	seq() int64
+	marshalBody(w io.Writer) error
+}
+
+// commitMsg wraps a #commit event for the subscriber channel.
+type commitMsg struct{ ev *RepoCommitEvent }
+
+func (m *commitMsg) msgType() string { return "#commit" }
+func (m *commitMsg) seq() int64      { return m.ev.Seq }
+func (m *commitMsg) marshalBody(w io.Writer) error {
+	indigoEvent := convertToIndigoCommit(m.ev)
+	var obj lexutil.CBOR = indigoEvent
+	return obj.MarshalCBOR(w)
+}
+
+// identityMsg wraps a #identity event for the subscriber channel.
+type identityMsg struct {
+	ev *atproto.SyncSubscribeRepos_Identity
+}
+
+func (m *identityMsg) msgType() string { return "#identity" }
+func (m *identityMsg) seq() int64      { return m.ev.Seq }
+func (m *identityMsg) marshalBody(w io.Writer) error {
+	var obj lexutil.CBOR = m.ev
+	return obj.MarshalCBOR(w)
+}
+
 // RepoCommitEvent represents a #commit event in subscribeRepos
 type RepoCommitEvent struct {
 	Seq      int64                                `json:"seq" cborgen:"seq"`
@@ -162,6 +197,15 @@ func (b *EventBroadcaster) initSchema() error {
 			created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
 		)`,
 		`CREATE INDEX IF NOT EXISTS idx_firehose_events_rev ON firehose_events(rev)`,
+		// Identity events live in a separate table so the commit-events schema
+		// stays unchanged. Both tables share the broadcaster's seq counter, and
+		// backfill UNIONs them on seq.
+		`CREATE TABLE IF NOT EXISTS firehose_identity_events (
+			seq INTEGER PRIMARY KEY,
+			did TEXT NOT NULL,
+			handle TEXT,
+			created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+		)`,
 	}
 
 	for _, stmt := range stmts {
@@ -174,9 +218,15 @@ func (b *EventBroadcaster) initSchema() error {
 	// Intentionally ignore error — fails with "duplicate column" if already present.
 	_, _ = b.db.Exec("ALTER TABLE firehose_events ADD COLUMN prev_data TEXT")
 
-	// Load last sequence number from database
+	// Load last sequence number from database — take MAX across both event tables
+	// since they share a sequence space.
 	var lastSeq sql.NullInt64
-	err := b.db.QueryRow("SELECT MAX(seq) FROM firehose_events").Scan(&lastSeq)
+	err := b.db.QueryRow(`
+		SELECT MAX(seq) FROM (
+			SELECT MAX(seq) AS seq FROM firehose_events
+			UNION ALL
+			SELECT MAX(seq) AS seq FROM firehose_identity_events
+		)`).Scan(&lastSeq)
 	if err != nil {
 		slog.Warn("Failed to load last event sequence", "error", err)
 	} else if lastSeq.Valid {
@@ -382,7 +432,7 @@ func (b *EventBroadcaster) Close() error {
 func (b *EventBroadcaster) Subscribe(conn *websocket.Conn, cursor int64, userAgent string) *Subscriber {
 	sub := &Subscriber{
 		conn:   conn,
-		send:   make(chan *RepoCommitEvent, 10), // Buffer 10 events
+		send:   make(chan firehoseMsg, 10), // Buffer 10 events
 		cursor: cursor,
 	}
 
@@ -503,9 +553,10 @@ func (b *EventBroadcaster) Broadcast(ctx context.Context, event *RepoEvent) {
 	b.addToHistory(seq, commitEvent)
 
 	// Broadcast to all subscribers
+	msg := &commitMsg{ev: commitEvent}
 	for sub := range b.subscribers {
 		select {
-		case sub.send <- commitEvent:
+		case sub.send <- msg:
 			// Sent successfully
 		default:
 			// Subscriber's buffer is full, skip (they'll get disconnected for being too slow)
@@ -514,6 +565,51 @@ func (b *EventBroadcaster) Broadcast(ctx context.Context, event *RepoEvent) {
 	}
 }
 
+// BroadcastIdentity emits a Sync 1.1 #identity event to all subscribers and
+// persists it for backfill. Use this when the hold's DID document changes
+// (handle rotation, signing-key rotation, PLC update) so downstream relays
+// know to refresh their identity cache.
+//
+// handle may be nil if the new handle is unknown or unchanged — the spec
+// allows omitting it.
+func (b *EventBroadcaster) BroadcastIdentity(ctx context.Context, handle *string) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.eventSeq++
+	seq := b.eventSeq
+
+	ev := &atproto.SyncSubscribeRepos_Identity{
+		Did:    b.holdDID,
+		Handle: handle,
+		Seq:    seq,
+		Time:   time.Now().Format(time.RFC3339),
+	}
+
+	if b.db != nil {
+		var h sql.NullString
+		if handle != nil {
+			h = sql.NullString{String: *handle, Valid: true}
+		}
+		_, err := b.db.Exec(
+			`INSERT INTO firehose_identity_events (seq, did, handle) VALUES (?, ?, ?)`,
+			seq, b.holdDID, h,
+		)
+		if err != nil {
+			slog.Warn("Failed to persist identity event", "seq", seq, "error", err)
+		}
+	}
+
+	msg := &identityMsg{ev: ev}
+	for sub := range b.subscribers {
+		select {
+		case sub.send <- msg:
+		default:
+			slog.Warn("Subscriber buffer full, skipping identity event", "seq", seq)
+		}
+	}
+}
+
 // persistEvent stores an event in the database
 func (b *EventBroadcaster) persistEvent(event *RepoCommitEvent) error {
 	// Serialize ops to JSON
@@ -551,17 +647,26 @@ func (b *EventBroadcaster) convertToCommitEvent(event *RepoEvent, seq int64) *Re
 		action := string(op.Kind) // "create", "update", "delete"
 		path := op.Collection + "/" + op.Rkey
 
-		// Convert CID to LexLink if present
+		// Convert new-record CID to LexLink if present (nil for delete).
 		var cidLink *lexutil.LexLink
 		if op.RecCid != nil {
 			link := lexutil.LexLink(*op.RecCid)
 			cidLink = &link
 		}
 
+		// Convert previous-record CID to LexLink if present (Sync 1.1 — required
+		// for update and delete ops; must be absent for create).
+		var prevLink *lexutil.LexLink
+		if op.Prev != nil {
+			link := lexutil.LexLink(*op.Prev)
+			prevLink = &link
+		}
+
 		ops[i] = &atproto.SyncSubscribeRepos_RepoOp{
 			Action: action,
 			Path:   path,
 			Cid:    cidLink,
+			Prev:   prevLink,
 		}
 	}
 
@@ -621,6 +726,14 @@ func (b *EventBroadcaster) backfillSubscriber(sub *Subscriber, cursor int64) {
 
 // backfillFromDatabase queries events from database and sends to subscriber
 func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) error {
+	// Identity events are rare, so load them up front and interleave with the
+	// commit stream by seq. Stays correct even if seq spaces drift.
+	identityQueue, err := b.loadIdentityEvents(cursor)
+	if err != nil {
+		slog.Warn("Failed to load identity events for backfill", "error", err)
+		// Continue without identity backfill — commits are more important
+	}
+
 	// Query events where seq > cursor, ordered by seq
 	// Include created_at to preserve original event timestamp
 	query := `
@@ -653,6 +766,14 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e
 			continue
 		}
 
+		// Flush any identity events that sit before this commit's seq.
+		for len(identityQueue) > 0 && identityQueue[0].ev.Seq < seq {
+			if !sendBackfillMsg(sub, identityQueue[0], identityQueue[0].ev.Seq) {
+				return nil
+			}
+			identityQueue = identityQueue[1:]
+		}
+
 		// Deserialize ops from JSON
 		var ops []*atproto.SyncSubscribeRepos_RepoOp
 		if err := json.Unmarshal(opsJSON, &ops); err != nil {
@@ -685,12 +806,14 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e
 		}
 
 		// Send to subscriber
-		select {
-		case sub.send <- event:
-			// Sent successfully
-		case <-time.After(5 * time.Second):
-			// Timeout, subscriber too slow
-			slog.Warn("Backfill timeout for subscriber", "seq", seq)
+		if !sendBackfillMsg(sub, &commitMsg{ev: event}, seq) {
+			return nil
+		}
+	}
+
+	// Drain any identity events that come after the last commit.
+	for _, m := range identityQueue {
+		if !sendBackfillMsg(sub, m, m.ev.Seq) {
 			return nil
 		}
 	}
@@ -698,6 +821,60 @@ func (b *EventBroadcaster) backfillFromDatabase(sub *Subscriber, cursor int64) e
 	return rows.Err()
 }
 
+// loadIdentityEvents pulls persisted #identity events with seq > cursor into a
+// slice for interleaved backfill. Identity events are rare, so loading them
+// fully into memory is fine.
+func (b *EventBroadcaster) loadIdentityEvents(cursor int64) ([]*identityMsg, error) {
+	rows, err := b.db.Query(`
+		SELECT seq, did, handle, created_at
+		FROM firehose_identity_events
+		WHERE seq > ?
+		ORDER BY seq ASC
+	`, cursor)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []*identityMsg
+	for rows.Next() {
+		var (
+			seq       int64
+			did       string
+			handle    sql.NullString
+			createdAt time.Time
+		)
+		if err := rows.Scan(&seq, &did, &handle, &createdAt); err != nil {
+			slog.Error("Error scanning identity event row", "error", err)
+			continue
+		}
+		var h *string
+		if handle.Valid {
+			h = &handle.String
+		}
+		out = append(out, &identityMsg{ev: &atproto.SyncSubscribeRepos_Identity{
+			Did:    did,
+			Handle: h,
+			Seq:    seq,
+			Time:   createdAt.Format(time.RFC3339),
+		}})
+	}
+	return out, rows.Err()
+}
+
+// sendBackfillMsg pushes a message into the subscriber's send channel with a
+// 5-second slow-subscriber timeout. Returns false if the timeout fired and the
+// caller should abandon the backfill.
+func sendBackfillMsg(sub *Subscriber, msg firehoseMsg, seq int64) bool {
+	select {
+	case sub.send <- msg:
+		return true
+	case <-time.After(5 * time.Second):
+		slog.Warn("Backfill timeout for subscriber", "seq", seq)
+		return false
+	}
+}
+
 // backfillFromMemory sends events from in-memory ring buffer (fallback)
 func (b *EventBroadcaster) backfillFromMemory(sub *Subscriber, cursor int64) {
 	b.mu.RLock()
@@ -706,7 +883,7 @@ func (b *EventBroadcaster) backfillFromMemory(sub *Subscriber, cursor int64) {
 	for _, he := range b.eventHistory {
 		if he.Seq > cursor {
 			select {
-			case sub.send <- he.Event:
+			case sub.send <- &commitMsg{ev: he.Event}:
 				// Sent
 			case <-time.After(5 * time.Second):
 				// Timeout, subscriber too slow
@@ -744,46 +921,36 @@ func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) {
 		sub.conn.Close()
 	}()
 
-	for event := range sub.send {
-		// Create event header (ATProto firehose format)
+	for msg := range sub.send {
 		header := events.EventHeader{
 			Op:      events.EvtKindMessage,
-			MsgType: "#commit",
+			MsgType: msg.msgType(),
 		}
 
-		// Get a writer for this message
 		wc, err := sub.conn.NextWriter(websocket.BinaryMessage)
 		if err != nil {
 			slog.Error("Failed to get websocket writer", "error", err)
 			return
 		}
 
-		// Write header as CBOR
 		if err := header.MarshalCBOR(wc); err != nil {
-			slog.Error("Failed to write event header", "error", err)
+			slog.Error("Failed to write event header", "error", err, "type", msg.msgType())
 			wc.Close()
 			return
 		}
 
-		// Convert our RepoCommitEvent to indigo's SyncSubscribeRepos_Commit
-		indigoEvent := convertToIndigoCommit(event)
-
-		// Write the event as CBOR
-		var obj lexutil.CBOR = indigoEvent
-		if err := obj.MarshalCBOR(wc); err != nil {
-			slog.Error("Failed to write event body", "error", err)
+		if err := msg.marshalBody(wc); err != nil {
+			slog.Error("Failed to write event body", "error", err, "type", msg.msgType())
 			wc.Close()
 			return
 		}
 
-		// Close the writer to flush the message
 		if err := wc.Close(); err != nil {
 			slog.Error("Failed to close websocket writer", "error", err)
 			return
 		}
 
-		// Update cursor
-		sub.cursor = event.Seq
+		sub.cursor = msg.seq()
 	}
 }
 

pkg/hold/pds/events_test.go

@@ -1,6 +1,7 @@
 package pds
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"testing"
@@ -185,6 +186,7 @@ func TestConvertToCommitEvent(t *testing.T) {
 
 	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")
 	prevDataCID, _ := cid.Decode("bafyreie5737gdxlw5i64mnsc7x35mha5ee4w5vqnivxceag4pfd2mhqtsu")
+	prevRecCID, _ := cid.Decode("bafyreigh2akiscaildcqabsyg3dfr6chu3fgpregiymsck7e7aqa4s52zy")
 	since := "prev-rev"
 
 	event := &RepoEvent{
@@ -205,12 +207,14 @@ func TestConvertToCommitEvent(t *testing.T) {
 				Collection: "io.atcr.hold.captain",
 				Rkey:       "self",
 				RecCid:     &testCID,
+				Prev:       &prevRecCID,
 			},
 			{
 				Kind:       EvtKindDeleteRecord,
 				Collection: "io.atcr.hold.crew",
 				Rkey:       "oldmember",
 				RecCid:     nil, // Deletes don't have CIDs
+				Prev:       &prevRecCID,
 			},
 		},
 	}
@@ -278,6 +282,10 @@ func TestConvertToCommitEvent(t *testing.T) {
 	if createOp.Cid == nil {
 		t.Error("Expected non-nil CID for create op")
 	}
+	// Sync 1.1: create ops must NOT carry a prev CID.
+	if createOp.Prev != nil {
+		t.Errorf("Expected nil Prev on create op, got %s", cid.Cid(*createOp.Prev))
+	}
 
 	// Check update op
 	updateOp := commitEvent.Ops[1]
@@ -287,6 +295,12 @@ func TestConvertToCommitEvent(t *testing.T) {
 	if updateOp.Path != "io.atcr.hold.captain/self" {
 		t.Errorf("Expected path=io.atcr.hold.captain/self, got %s", updateOp.Path)
 	}
+	// Sync 1.1: update ops must carry the prior record CID.
+	if updateOp.Prev == nil {
+		t.Error("Expected non-nil Prev on update op")
+	} else if cid.Cid(*updateOp.Prev) != prevRecCID {
+		t.Errorf("Expected Prev=%s on update op, got %s", prevRecCID, cid.Cid(*updateOp.Prev))
+	}
 
 	// Check delete op
 	deleteOp := commitEvent.Ops[2]
@@ -299,6 +313,12 @@ func TestConvertToCommitEvent(t *testing.T) {
 	if deleteOp.Cid != nil {
 		t.Error("Expected nil CID for delete op")
 	}
+	// Sync 1.1: delete ops must carry the prior record CID.
+	if deleteOp.Prev == nil {
+		t.Error("Expected non-nil Prev on delete op")
+	} else if cid.Cid(*deleteOp.Prev) != prevRecCID {
+		t.Errorf("Expected Prev=%s on delete op, got %s", prevRecCID, cid.Cid(*deleteOp.Prev))
+	}
 }
 
 // TestConvertToCommitEvent_NoSince tests event without since field
@@ -488,8 +508,8 @@ func TestSubscribe_CursorZeroBackfill(t *testing.T) {
 	// Test backfillSubscriber directly with cursor=0
 	// Create a subscriber manually (conn not needed for backfill test)
 	sub := &Subscriber{
-		conn:   nil,                              // Not used in backfillSubscriber
-		send:   make(chan *RepoCommitEvent, 100), // Large buffer for testing
+		conn:   nil,                         // Not used in backfillSubscriber
+		send:   make(chan firehoseMsg, 100), // Large buffer for testing
 		cursor: 0,
 	}
 
@@ -509,8 +529,8 @@ func TestSubscribe_CursorZeroBackfill(t *testing.T) {
 	for i := 1; i <= 5; i++ {
 		select {
 		case event := <-sub.send:
-			if event.Seq != int64(i) {
-				t.Errorf("Expected event seq=%d, got %d", i, event.Seq)
+			if event.seq() != int64(i) {
+				t.Errorf("Expected event seq=%d, got %d", i, event.seq())
 			}
 		default:
 			t.Errorf("Expected event %d but channel was empty", i)
@@ -538,8 +558,8 @@ func TestSubscribe_MidCursorBackfill(t *testing.T) {
 
 	// Test backfillSubscriber with cursor=5 (conn not needed for backfill test)
 	sub := &Subscriber{
-		conn:   nil,                              // Not used in backfillSubscriber
-		send:   make(chan *RepoCommitEvent, 100), // Large buffer for testing
+		conn:   nil,                         // Not used in backfillSubscriber
+		send:   make(chan firehoseMsg, 100), // Large buffer for testing
 		cursor: 5,
 	}
 
@@ -559,8 +579,8 @@ func TestSubscribe_MidCursorBackfill(t *testing.T) {
 	for i := 6; i <= 10; i++ {
 		select {
 		case event := <-sub.send:
-			if event.Seq != int64(i) {
-				t.Errorf("Expected event seq=%d, got %d", i, event.Seq)
+			if event.seq() != int64(i) {
+				t.Errorf("Expected event seq=%d, got %d", i, event.seq())
 			}
 		default:
 			t.Errorf("Expected event %d but channel was empty", i)
@@ -589,7 +609,7 @@ func TestSubscribe_NegativeCursorNoBackfill(t *testing.T) {
 	// Create subscriber with cursor=-1 (no backfill, conn not needed)
 	sub := &Subscriber{
 		conn:   nil, // Not used in this test
-		send:   make(chan *RepoCommitEvent, 100),
+		send:   make(chan firehoseMsg, 100),
 		cursor: -1,
 	}
 
@@ -610,3 +630,195 @@ func TestSubscribe_NegativeCursorNoBackfill(t *testing.T) {
 		t.Errorf("Expected 0 events with cursor=-1 (no backfill), got %d", len(sub.send))
 	}
 }
+
+// TestBroadcastIdentity verifies that BroadcastIdentity increments the shared
+// seq counter and pushes an #identity message to subscribers.
+func TestBroadcastIdentity(t *testing.T) {
+	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "")
+	ctx := context.Background()
+
+	sub := &Subscriber{
+		conn:   nil,
+		send:   make(chan firehoseMsg, 10),
+		cursor: 0,
+	}
+	broadcaster.mu.Lock()
+	broadcaster.subscribers[sub] = true
+	broadcaster.mu.Unlock()
+
+	handle := "hold.example.com"
+	broadcaster.BroadcastIdentity(ctx, &handle)
+
+	if got := broadcaster.GetCurrentSeq(); got != 1 {
+		t.Errorf("expected seq=1 after BroadcastIdentity, got %d", got)
+	}
+
+	select {
+	case msg := <-sub.send:
+		idMsg, ok := msg.(*identityMsg)
+		if !ok {
+			t.Fatalf("expected *identityMsg, got %T", msg)
+		}
+		if idMsg.msgType() != "#identity" {
+			t.Errorf("expected msgType=#identity, got %s", idMsg.msgType())
+		}
+		if idMsg.ev.Did != "did:web:hold.example.com" {
+			t.Errorf("expected Did=did:web:hold.example.com, got %s", idMsg.ev.Did)
+		}
+		if idMsg.ev.Handle == nil || *idMsg.ev.Handle != handle {
+			t.Errorf("expected Handle=%s, got %v", handle, idMsg.ev.Handle)
+		}
+		if idMsg.ev.Seq != 1 {
+			t.Errorf("expected Seq=1, got %d", idMsg.ev.Seq)
+		}
+		if idMsg.ev.Time == "" {
+			t.Error("expected non-empty Time")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("did not receive identity event")
+	}
+}
+
+// TestBroadcastIdentity_NilHandle verifies the spec-allowed nil-handle path.
+func TestBroadcastIdentity_NilHandle(t *testing.T) {
+	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "")
+	ctx := context.Background()
+
+	sub := &Subscriber{
+		conn:   nil,
+		send:   make(chan firehoseMsg, 10),
+		cursor: 0,
+	}
+	broadcaster.mu.Lock()
+	broadcaster.subscribers[sub] = true
+	broadcaster.mu.Unlock()
+
+	broadcaster.BroadcastIdentity(ctx, nil)
+
+	select {
+	case msg := <-sub.send:
+		idMsg, ok := msg.(*identityMsg)
+		if !ok {
+			t.Fatalf("expected *identityMsg, got %T", msg)
+		}
+		if idMsg.ev.Handle != nil {
+			t.Errorf("expected nil Handle, got %v", idMsg.ev.Handle)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("did not receive identity event")
+	}
+}
+
+// TestBroadcastIdentity_SharedSeqWithCommits verifies that identity events and
+// commit events draw from the same monotonic seq counter — required for
+// consistent cursor semantics on the firehose.
+func TestBroadcastIdentity_SharedSeqWithCommits(t *testing.T) {
+	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, "")
+	ctx := context.Background()
+	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")
+
+	commit := &RepoEvent{
+		NewRoot:   testCID,
+		Rev:       "rev-1",
+		RepoSlice: []byte("car"),
+		Ops:       []RepoOp{},
+	}
+	broadcaster.Broadcast(ctx, commit)      // seq=1
+	broadcaster.BroadcastIdentity(ctx, nil) // seq=2
+	broadcaster.Broadcast(ctx, commit)      // seq=3
+	broadcaster.BroadcastIdentity(ctx, nil) // seq=4
+
+	if got := broadcaster.GetCurrentSeq(); got != 4 {
+		t.Errorf("expected seq=4 after mixed broadcasts, got %d", got)
+	}
+}
+
+// TestBackfill_InterleavesIdentityAndCommits verifies that backfill replays
+// identity events at the correct seq position relative to commits.
+func TestBackfill_InterleavesIdentityAndCommits(t *testing.T) {
+	dbPath := t.TempDir() + "/events.db"
+	broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100, dbPath)
+	t.Cleanup(func() { _ = broadcaster.Close() })
+
+	ctx := context.Background()
+	testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke")
+	commit := &RepoEvent{
+		NewRoot:   testCID,
+		Rev:       "rev",
+		RepoSlice: []byte("car"),
+		Ops:       []RepoOp{},
+	}
+
+	broadcaster.Broadcast(ctx, commit)      // seq=1
+	broadcaster.Broadcast(ctx, commit)      // seq=2
+	broadcaster.BroadcastIdentity(ctx, nil) // seq=3
+	broadcaster.Broadcast(ctx, commit)      // seq=4
+	broadcaster.BroadcastIdentity(ctx, nil) // seq=5
+
+	sub := &Subscriber{
+		conn:   nil,
+		send:   make(chan firehoseMsg, 16),
+		cursor: 0,
+	}
+
+	if err := broadcaster.backfillFromDatabase(sub, 0); err != nil {
+		t.Fatalf("backfillFromDatabase: %v", err)
+	}
+	close(sub.send)
+
+	var got []struct {
+		typ string
+		seq int64
+	}
+	for msg := range sub.send {
+		got = append(got, struct {
+			typ string
+			seq int64
+		}{msg.msgType(), msg.seq()})
+	}
+
+	want := []struct {
+		typ string
+		seq int64
+	}{
+		{"#commit", 1},
+		{"#commit", 2},
+		{"#identity", 3},
+		{"#commit", 4},
+		{"#identity", 5},
+	}
+	if len(got) != len(want) {
+		t.Fatalf("expected %d messages, got %d (%+v)", len(want), len(got), got)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Errorf("at index %d: want %+v, got %+v", i, want[i], got[i])
+		}
+	}
+}
+
+// TestIdentityMsg_MarshalBody verifies the wire-format CBOR marshal path.
+func TestIdentityMsg_MarshalBody(t *testing.T) {
+	handle := "hold.example.com"
+	msg := &identityMsg{ev: &atproto.SyncSubscribeRepos_Identity{
+		Did:    "did:web:hold.example.com",
+		Handle: &handle,
+		Seq:    42,
+		Time:   time.Now().UTC().Format(time.RFC3339),
+	}}
+
+	if msg.msgType() != "#identity" {
+		t.Errorf("expected #identity, got %s", msg.msgType())
+	}
+	if msg.seq() != 42 {
+		t.Errorf("expected seq=42, got %d", msg.seq())
+	}
+
+	var buf bytes.Buffer
+	if err := msg.marshalBody(&buf); err != nil {
+		t.Fatalf("marshalBody: %v", err)
+	}
+	if buf.Len() == 0 {
+		t.Error("expected non-empty CBOR body")
+	}
+}

pkg/hold/pds/jti_cache.go

@@ -0,0 +1,55 @@
+package pds
+
+import (
+	"sync"
+	"time"
+)
+
+// jtiReplayCache tracks JWT IDs of recently-seen appview tokens so the same
+// token can't be replayed within its TTL. Entries auto-expire at the token's
+// claimed expiration time (no background sweeper needed — eviction happens
+// lazily on Seen()).
+type jtiReplayCache struct {
+	mu      sync.Mutex
+	entries map[string]time.Time
+	clock   func() time.Time
+}
+
+func newJTIReplayCache() *jtiReplayCache {
+	return &jtiReplayCache{entries: make(map[string]time.Time), clock: time.Now}
+}
+
+// Seen records `jti` with the given expiration and returns true if the jti was
+// already present (i.e., this is a replay). It opportunistically evicts any
+// already-expired entries it encounters.
+func (c *jtiReplayCache) Seen(jti string, exp time.Time) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	now := c.clock()
+	// lazy eviction: drop the entry we're about to look at if it's expired
+	if prev, ok := c.entries[jti]; ok {
+		if !prev.After(now) {
+			delete(c.entries, jti)
+		} else {
+			return true
+		}
+	}
+	// opportunistic sweep — keeps the map bounded under steady-state load
+	// without needing a goroutine. O(n) but n is small in practice.
+	if len(c.entries) > 0 && len(c.entries)%64 == 0 {
+		for k, e := range c.entries {
+			if !e.After(now) {
+				delete(c.entries, k)
+			}
+		}
+	}
+	c.entries[jti] = exp
+	return false
+}
+
+// size returns the number of tracked entries (test-only helper).
+func (c *jtiReplayCache) size() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return len(c.entries)
+}

pkg/hold/pds/jti_cache_test.go

@@ -0,0 +1,83 @@
+package pds
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestJTIReplayCache_FirstSeenAccepted(t *testing.T) {
+	c := newJTIReplayCache()
+	if c.Seen("a", time.Now().Add(time.Minute)) {
+		t.Fatal("first observation of jti reported as replay")
+	}
+}
+
+func TestJTIReplayCache_DuplicateRejected(t *testing.T) {
+	c := newJTIReplayCache()
+	exp := time.Now().Add(time.Minute)
+	_ = c.Seen("dup", exp)
+	if !c.Seen("dup", exp) {
+		t.Error("second observation of same jti not flagged as replay")
+	}
+}
+
+func TestJTIReplayCache_DistinctJTIsAccepted(t *testing.T) {
+	c := newJTIReplayCache()
+	exp := time.Now().Add(time.Minute)
+	if c.Seen("a", exp) || c.Seen("b", exp) {
+		t.Error("distinct jtis should both be accepted")
+	}
+}
+
+func TestJTIReplayCache_ExpiredEntryReusable(t *testing.T) {
+	c := newJTIReplayCache()
+	now := time.Unix(1_000_000, 0)
+	c.clock = func() time.Time { return now }
+
+	if c.Seen("a", now.Add(time.Second)) {
+		t.Fatal("first seen flagged as replay")
+	}
+	// advance past expiration
+	now = now.Add(2 * time.Second)
+	if c.Seen("a", now.Add(time.Second)) {
+		t.Error("after exp, jti should be reusable (entry evicted)")
+	}
+}
+
+func TestJTIReplayCache_DoesNotGrowUnbounded(t *testing.T) {
+	c := newJTIReplayCache()
+	now := time.Unix(1_000_000, 0)
+	c.clock = func() time.Time { return now }
+
+	for i := range 1024 {
+		_ = c.Seen(string(rune('A'+i%64))+string(rune('0'+i/64)), now.Add(time.Second))
+	}
+	// Jump past all expirations.
+	now = now.Add(time.Hour)
+	// One more insert triggers a sweep at the 64-multiple boundary.
+	for range 200 {
+		_ = c.Seen("trigger-"+time.Now().String(), now.Add(time.Second))
+	}
+	if got := c.size(); got > 300 {
+		t.Errorf("cache size after eviction = %d, expected sweep to drop expired entries", got)
+	}
+}
+
+func TestJTIReplayCache_ConcurrentSafe(t *testing.T) {
+	c := newJTIReplayCache()
+	exp := time.Now().Add(time.Minute)
+	var wg sync.WaitGroup
+	for i := range 32 {
+		wg.Add(1)
+		go func(n int) {
+			defer wg.Done()
+			c.Seen("concurrent", exp)
+			c.Seen(string(rune(n)), exp)
+		}(i)
+	}
+	wg.Wait()
+	if c.size() == 0 {
+		t.Error("expected entries after concurrent inserts")
+	}
+}

pkg/hold/pds/repo.go

@@ -112,6 +112,10 @@ func (d *DirectRepoOperator) commitWrite(ctx context.Context, ws *writeSession,
 		return cid.Undef, "", err
 	}
 
+	if err := fillPrevCIDs(ctx, ws.r, ws.head, ops); err != nil {
+		d.log.Warn("fillPrevCIDs failed; emitting without prev", "err", err)
+	}
+
 	rslice, err := ws.ds.CloseWithRoot(ctx, nroot, nrev)
 	if err != nil {
 		return cid.Undef, "", fmt.Errorf("close with root: %w", err)

pkg/hold/pds/repo_operator.go

@@ -74,7 +74,8 @@ type RepoOp struct {
 	Kind       EventKind
 	Collection string
 	Rkey       string
-	RecCid     *cid.Cid
+	RecCid     *cid.Cid // new record CID — nil for delete
+	Prev       *cid.Cid // previous record CID — set for update/delete (Sync 1.1)
 	Record     any
 	ActorInfo  *ActorInfo
 }

pkg/hold/pds/repo_operator_test.go

@@ -587,7 +587,7 @@ func runRepoOperatorTests(t *testing.T, setup func(t *testing.T) (RepoOperator,
 
 		// Create first (no event handler yet)
 		rec := newCrewRecord("did:plc:evt-update")
-		_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, "evtupdate", rec)
+		_, origCID, err := op.PutRecord(ctx, uid, atproto.CrewCollection, "evtupdate", rec)
 		if err != nil {
 			t.Fatalf("PutRecord: %v", err)
 		}
@@ -625,6 +625,13 @@ func runRepoOperatorTests(t *testing.T, setup func(t *testing.T) (RepoOperator,
 		if evt.Ops[0].Kind != EvtKindUpdateRecord {
 			t.Errorf("expected kind=update, got %q", evt.Ops[0].Kind)
 		}
+		// Sync 1.1: update op must carry the prior record CID.
+		if evt.Ops[0].Prev == nil {
+			t.Fatal("expected non-nil Prev on update op")
+		}
+		if *evt.Ops[0].Prev != origCID {
+			t.Errorf("expected Prev=%s, got %s", origCID, evt.Ops[0].Prev)
+		}
 	})
 
 	t.Run("EventEmission_Delete", func(t *testing.T) {
@@ -632,7 +639,7 @@ func runRepoOperatorTests(t *testing.T, setup func(t *testing.T) (RepoOperator,
 		ctx := context.Background()
 
 		rec := newCrewRecord("did:plc:evt-delete")
-		_, _, err := op.PutRecord(ctx, uid, atproto.CrewCollection, "evtdelete", rec)
+		_, origCID, err := op.PutRecord(ctx, uid, atproto.CrewCollection, "evtdelete", rec)
 		if err != nil {
 			t.Fatalf("PutRecord: %v", err)
 		}
@@ -660,6 +667,14 @@ func runRepoOperatorTests(t *testing.T, setup func(t *testing.T) (RepoOperator,
 		if evt.Ops[0].RecCid != nil {
 			t.Error("expected nil RecCid for delete op")
 		}
+		// Sync 1.1: delete op must carry the prior record CID — this is the
+		// exact gap that caused "delete op ... missing prev CID" relay errors.
+		if evt.Ops[0].Prev == nil {
+			t.Fatal("expected non-nil Prev on delete op")
+		}
+		if *evt.Ops[0].Prev != origCID {
+			t.Errorf("expected Prev=%s, got %s", origCID, evt.Ops[0].Prev)
+		}
 	})
 
 	t.Run("EventEmission_Hydrate", func(t *testing.T) {

pkg/hold/pds/repomgr.go

@@ -98,6 +98,39 @@ func (rm *RepoManager) lockUser(ctx context.Context, user models.Uid) func() {
 	}
 }
 
+// fillPrevCIDs uses the MST diff between oldRoot and the current repo state to
+// stamp Prev on existing RepoOps in-place. Required for Sync 1.1 inductive
+// firehose: update/delete ops must carry the previous record CID.
+//
+// Call after r.Commit() and before ds.CloseWithRoot() — the delta session still
+// holds both old and new blocks at that point. No-op for first commits.
+func fillPrevCIDs(ctx context.Context, r *repo.Repo, oldRoot cid.Cid, ops []RepoOp) error {
+	if !oldRoot.Defined() || len(ops) == 0 {
+		return nil
+	}
+	diff, err := r.DiffSince(ctx, oldRoot)
+	if err != nil {
+		return fmt.Errorf("DiffSince: %w", err)
+	}
+	prevByPath := make(map[string]cid.Cid, len(diff))
+	for _, d := range diff {
+		// mst.DiffOp.Op is "add" | "mut" | "del" — only the latter two have an OldCid.
+		if d.Op == "mut" || d.Op == "del" {
+			prevByPath[d.Rpath] = d.OldCid
+		}
+	}
+	for i := range ops {
+		if ops[i].Kind == EvtKindCreateRecord {
+			continue
+		}
+		path := ops[i].Collection + "/" + ops[i].Rkey
+		if pc, ok := prevByPath[path]; ok {
+			ops[i].Prev = &pc
+		}
+	}
+	return nil
+}
+
 func (rm *RepoManager) CreateRecord(ctx context.Context, user models.Uid, collection string, rec cbg.CBORMarshaler) (string, cid.Cid, error) {
 	ctx, span := otel.Tracer("repoman").Start(ctx, "CreateRecord")
 	defer span.End()
@@ -212,6 +245,20 @@ func (rm *RepoManager) UpdateRecord(ctx context.Context, user models.Uid, collec
 		return cid.Undef, err
 	}
 
+	op := RepoOp{
+		Kind:       EvtKindUpdateRecord,
+		Collection: collection,
+		Rkey:       rkey,
+		RecCid:     &cc,
+	}
+	if rm.hydrateRecords {
+		op.Record = rec
+	}
+	ops := []RepoOp{op}
+	if err := fillPrevCIDs(ctx, r, head, ops); err != nil {
+		rm.log.Warn("fillPrevCIDs failed; emitting without prev", "err", err)
+	}
+
 	rslice, err := ds.CloseWithRoot(ctx, nroot, nrev)
 	if err != nil {
 		return cid.Undef, fmt.Errorf("close with root: %w", err)
@@ -223,17 +270,6 @@ func (rm *RepoManager) UpdateRecord(ctx context.Context, user models.Uid, collec
 	}
 
 	if rm.events != nil {
-		op := RepoOp{
-			Kind:       EvtKindUpdateRecord,
-			Collection: collection,
-			Rkey:       rkey,
-			RecCid:     &cc,
-		}
-
-		if rm.hydrateRecords {
-			op.Record = rec
-		}
-
 		rm.events(ctx, &RepoEvent{
 			User:      user,
 			OldRoot:   oldroot,
@@ -241,7 +277,7 @@ func (rm *RepoManager) UpdateRecord(ctx context.Context, user models.Uid, collec
 			PrevData:  prevData,
 			Rev:       nrev,
 			Since:     &rev,
-			Ops:       []RepoOp{op},
+			Ops:       ops,
 			RepoSlice: rslice,
 		})
 	}
@@ -390,6 +426,20 @@ func (rm *RepoManager) UpsertRecord(ctx context.Context, user models.Uid, collec
 		return "", cid.Undef, false, err
 	}
 
+	op := RepoOp{
+		Kind:       evtKind,
+		Collection: collection,
+		Rkey:       rkey,
+		RecCid:     &cc,
+	}
+	if rm.hydrateRecords {
+		op.Record = rec
+	}
+	ops := []RepoOp{op}
+	if err := fillPrevCIDs(ctx, r, head, ops); err != nil {
+		rm.log.Warn("fillPrevCIDs failed; emitting without prev", "err", err)
+	}
+
 	rslice, err := ds.CloseWithRoot(ctx, nroot, nrev)
 	if err != nil {
 		return "", cid.Undef, false, fmt.Errorf("close with root: %w", err)
@@ -401,17 +451,6 @@ func (rm *RepoManager) UpsertRecord(ctx context.Context, user models.Uid, collec
 	}
 
 	if rm.events != nil {
-		op := RepoOp{
-			Kind:       evtKind,
-			Collection: collection,
-			Rkey:       rkey,
-			RecCid:     &cc,
-		}
-
-		if rm.hydrateRecords {
-			op.Record = rec
-		}
-
 		rm.events(ctx, &RepoEvent{
 			User:      user,
 			OldRoot:   oldroot,
@@ -419,7 +458,7 @@ func (rm *RepoManager) UpsertRecord(ctx context.Context, user models.Uid, collec
 			PrevData:  prevData,
 			Rev:       nrev,
 			Since:     &rev,
-			Ops:       []RepoOp{op},
+			Ops:       ops,
 			RepoSlice: rslice,
 		})
 	}
@@ -467,6 +506,15 @@ func (rm *RepoManager) DeleteRecord(ctx context.Context, user models.Uid, collec
 		return err
 	}
 
+	ops := []RepoOp{{
+		Kind:       EvtKindDeleteRecord,
+		Collection: collection,
+		Rkey:       rkey,
+	}}
+	if err := fillPrevCIDs(ctx, r, head, ops); err != nil {
+		rm.log.Warn("fillPrevCIDs failed; emitting without prev", "err", err)
+	}
+
 	rslice, err := ds.CloseWithRoot(ctx, nroot, nrev)
 	if err != nil {
 		return fmt.Errorf("close with root: %w", err)
@@ -479,17 +527,13 @@ func (rm *RepoManager) DeleteRecord(ctx context.Context, user models.Uid, collec
 
 	if rm.events != nil {
 		rm.events(ctx, &RepoEvent{
-			User:     user,
-			OldRoot:  oldroot,
-			NewRoot:  nroot,
-			PrevData: prevData,
-			Rev:      nrev,
-			Since:    &rev,
-			Ops: []RepoOp{{
-				Kind:       EvtKindDeleteRecord,
-				Collection: collection,
-				Rkey:       rkey,
-			}},
+			User:      user,
+			OldRoot:   oldroot,
+			NewRoot:   nroot,
+			PrevData:  prevData,
+			Rev:       nrev,
+			Since:     &rev,
+			Ops:       ops,
 			RepoSlice: rslice,
 		})
 	}
@@ -733,6 +777,10 @@ func (rm *RepoManager) BatchWrite(ctx context.Context, user models.Uid, writes [
 		return err
 	}
 
+	if err := fillPrevCIDs(ctx, r, head, ops); err != nil {
+		rm.log.Warn("fillPrevCIDs failed; emitting without prev", "err", err)
+	}
+
 	rslice, err := ds.CloseWithRoot(ctx, nroot, nrev)
 	if err != nil {
 		return fmt.Errorf("close with root: %w", err)
@@ -825,6 +873,10 @@ func (rm *RepoManager) BulkUpsert(ctx context.Context, user models.Uid, records
 		return err
 	}
 
+	if err := fillPrevCIDs(ctx, r, head, ops); err != nil {
+		rm.log.Warn("fillPrevCIDs failed; emitting without prev", "err", err)
+	}
+
 	rslice, err := ds.CloseWithRoot(ctx, nroot, nrev)
 	if err != nil {
 		return fmt.Errorf("close with root: %w", err)

pkg/hold/pds/xrpc.go

@@ -11,6 +11,7 @@ import (
 	"atcr.io/pkg/hold/quota"
 	"atcr.io/pkg/s3"
 	"github.com/bluesky-social/indigo/api/bsky"
+	"github.com/bluesky-social/indigo/atproto/syntax"
 	lexutil "github.com/bluesky-social/indigo/lex/util"
 	"github.com/bluesky-social/indigo/repo"
 	"github.com/go-chi/chi/v5"
@@ -294,7 +295,7 @@ func (h *XRPCHandler) HandleGetProfile(w http.ResponseWriter, r *http.Request) {
 
 	// Normalize actor to DID
 	actorDID := actor
-	if !atproto.IsDID(actor) {
+	if _, err := syntax.ParseDID(actor); err != nil {
 		// It's a handle, resolve to DID
 		expectedHandle := didWebHandle(h.pds.DID())
 		if actor == expectedHandle {
@@ -334,7 +335,7 @@ func (h *XRPCHandler) HandleGetProfiles(w http.ResponseWriter, r *http.Request)
 	for _, actor := range actors {
 		// Normalize actor to DID
 		actorDID := actor
-		if !atproto.IsDID(actor) {
+		if _, err := syntax.ParseDID(actor); err != nil {
 			// It's a handle, check if it matches
 			if actor == expectedHandle {
 				actorDID = h.pds.DID()
@@ -1691,7 +1692,7 @@ func (h *XRPCHandler) HandleGetQuota(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Validate DID format
-	if !atproto.IsDID(userDID) {
+	if _, err := syntax.ParseDID(userDID); err != nil {
 		http.Error(w, "invalid userDid format", http.StatusBadRequest)
 		return
 	}
@@ -1765,7 +1766,7 @@ func (h *XRPCHandler) HandleUpdateCrewTier(w http.ResponseWriter, r *http.Reques
 		userDID = req.UserDID
 	}
 
-	if !atproto.IsDID(userDID) {
+	if _, err := syntax.ParseDID(userDID); err != nil {
 		http.Error(w, "invalid userDid format", http.StatusBadRequest)
 		return
 	}