From efabb677e4aad5563e333a78b2dc04547eaa5c0a Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Tue, 26 May 2026 20:19:19 -0500 Subject: [PATCH] clean up some functions to use indigo helpers. make repomgr more sync1.1 compliant --- go.mod | 2 +- pkg/appview/storage/profile.go | 6 +- pkg/atproto/directory.go | 37 +---- pkg/atproto/lexicon.go | 5 - pkg/atproto/lexicon_test.go | 92 ++++------- pkg/atproto/resolver.go | 4 +- pkg/auth/appview_token.go | 118 +++++++------- pkg/auth/appview_token_test.go | 151 ++++++++++++++++++ pkg/hold/admin/handlers_settings.go | 4 +- pkg/hold/labeler/subscriber_test.go | 4 +- pkg/hold/pds/appview_token_test.go | 186 ++++++++++++++++++++++ pkg/hold/pds/auth.go | 46 ++++-- pkg/hold/pds/events.go | 227 +++++++++++++++++++++++---- pkg/hold/pds/events_test.go | 230 ++++++++++++++++++++++++++-- pkg/hold/pds/jti_cache.go | 55 +++++++ pkg/hold/pds/jti_cache_test.go | 83 ++++++++++ pkg/hold/pds/repo.go | 4 + pkg/hold/pds/repo_operator.go | 3 +- pkg/hold/pds/repo_operator_test.go | 19 ++- pkg/hold/pds/repomgr.go | 122 ++++++++++----- pkg/hold/pds/xrpc.go | 9 +- 21 files changed, 1149 insertions(+), 258 deletions(-) create mode 100644 pkg/auth/appview_token_test.go create mode 100644 pkg/hold/pds/appview_token_test.go create mode 100644 pkg/hold/pds/jti_cache.go create mode 100644 pkg/hold/pds/jti_cache_test.go diff --git a/go.mod b/go.mod index d7a47a7..f558e09 100644 --- a/go.mod +++ b/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 diff --git a/pkg/appview/storage/profile.go b/pkg/appview/storage/profile.go index 72965f1..4775dee 100644 --- a/pkg/appview/storage/profile.go +++ b/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 { diff --git a/pkg/atproto/directory.go b/pkg/atproto/directory.go index 66e546f..9ad06e3 100644 --- a/pkg/atproto/directory.go +++ b/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) -} diff --git a/pkg/atproto/lexicon.go b/pkg/atproto/lexicon.go index 76ddd2b..ef918da 100644 --- a/pkg/atproto/lexicon.go +++ b/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 { diff --git a/pkg/atproto/lexicon_test.go b/pkg/atproto/lexicon_test.go index f27a8c6..1d224a0 100644 --- a/pkg/atproto/lexicon_test.go +++ b/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 := `{ diff --git a/pkg/atproto/resolver.go b/pkg/atproto/resolver.go index 056fbd3..13149bd 100644 --- a/pkg/atproto/resolver.go +++ b/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) } diff --git a/pkg/auth/appview_token.go b/pkg/auth/appview_token.go index 5b659d6..d4d772a 100644 --- a/pkg/auth/appview_token.go +++ b/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 diff --git a/pkg/auth/appview_token_test.go b/pkg/auth/appview_token_test.go new file mode 100644 index 0000000..0c13913 --- /dev/null +++ b/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) + } +} diff --git a/pkg/hold/admin/handlers_settings.go b/pkg/hold/admin/handlers_settings.go index f9c00ed..9e6b8db 100644 --- a/pkg/hold/admin/handlers_settings.go +++ b/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 } diff --git a/pkg/hold/labeler/subscriber_test.go b/pkg/hold/labeler/subscriber_test.go index 2baf55f..4b8d8eb 100644 --- a/pkg/hold/labeler/subscriber_test.go +++ b/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), }) diff --git a/pkg/hold/pds/appview_token_test.go b/pkg/hold/pds/appview_token_test.go new file mode 100644 index 0000000..66f686e --- /dev/null +++ b/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) + } +} diff --git a/pkg/hold/pds/auth.go b/pkg/hold/pds/auth.go index 2eabad8..677a990 100644 --- a/pkg/hold/pds/auth.go +++ b/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 diff --git a/pkg/hold/pds/events.go b/pkg/hold/pds/events.go index 11a3d76..1067d33 100644 --- a/pkg/hold/pds/events.go +++ b/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() } } diff --git a/pkg/hold/pds/events_test.go b/pkg/hold/pds/events_test.go index a1f8146..82919c0 100644 --- a/pkg/hold/pds/events_test.go +++ b/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") + } +} diff --git a/pkg/hold/pds/jti_cache.go b/pkg/hold/pds/jti_cache.go new file mode 100644 index 0000000..af17b96 --- /dev/null +++ b/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) +} diff --git a/pkg/hold/pds/jti_cache_test.go b/pkg/hold/pds/jti_cache_test.go new file mode 100644 index 0000000..b6852a6 --- /dev/null +++ b/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") + } +} diff --git a/pkg/hold/pds/repo.go b/pkg/hold/pds/repo.go index e1de29f..7232423 100644 --- a/pkg/hold/pds/repo.go +++ b/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) diff --git a/pkg/hold/pds/repo_operator.go b/pkg/hold/pds/repo_operator.go index 67ccf5b..a9847fa 100644 --- a/pkg/hold/pds/repo_operator.go +++ b/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 } diff --git a/pkg/hold/pds/repo_operator_test.go b/pkg/hold/pds/repo_operator_test.go index 8ef5024..fa9ed5a 100644 --- a/pkg/hold/pds/repo_operator_test.go +++ b/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) { diff --git a/pkg/hold/pds/repomgr.go b/pkg/hold/pds/repomgr.go index 68ad91f..1b61bda 100644 --- a/pkg/hold/pds/repomgr.go +++ b/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) diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index ac09478..5bd0138 100644 --- a/pkg/hold/pds/xrpc.go +++ b/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 }