diff --git a/cmd/hold/main.go b/cmd/hold/main.go index d29fb87..68efed1 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -147,6 +147,7 @@ func main() { // Update status post to "online" after server starts if holdPDS != nil { ctx := context.Background() + if err := holdPDS.SetStatus(ctx, "online"); err != nil { log.Printf("Warning: Failed to set status post to online: %v", err) } else { diff --git a/go.mod b/go.mod index 8584335..acb6e27 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.24.7 require ( github.com/aws/aws-sdk-go v1.55.5 - github.com/bluesky-social/indigo v0.0.0-20251014222321-1e8718ae9f33 + github.com/bluesky-social/indigo v0.0.0-20251021193747-543ab1124beb github.com/distribution/distribution/v3 v3.0.0 github.com/distribution/reference v0.6.0 github.com/go-chi/chi/v5 v5.2.3 @@ -34,6 +34,7 @@ require ( ) require ( + github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b // indirect github.com/aymerick/douceur v0.2.0 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/bshuster-repo/logrus-logstash-hook v1.0.0 // indirect @@ -46,7 +47,6 @@ require ( github.com/docker/go-metrics v0.0.1 // indirect github.com/earthboundkid/versioninfo/v2 v2.24.1 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect - github.com/go-chi/cors v1.2.2 // indirect github.com/go-jose/go-jose/v4 v4.1.2 // indirect github.com/go-logr/logr v1.4.2 // indirect github.com/go-logr/stdr v1.2.2 // indirect diff --git a/go.sum b/go.sum index 63cf36c..f2e37e0 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,8 @@ github.com/AdaLogics/go-fuzz-headers v0.0.0-20221103172237-443f56ff4ba8 h1:d+pBUmsteW5tM87xmVXHZ4+LibHRFn40SPAoZJOg2ak= github.com/AdaLogics/go-fuzz-headers v0.0.0-20221103172237-443f56ff4ba8/go.mod h1:i9fr2JpcEcY/IHEvzCM3qXUZYOQHgR89dt4es1CgMhc= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b h1:5/++qT1/z812ZqBvqQt6ToRswSuPZ/B33m6xVHRzADU= +github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b/go.mod h1:4+EPqMRApwwE/6yo6CxiHoSnBzjRr3jsqer7frxP8y4= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM= @@ -18,8 +20,8 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932 h1:mXoPYz/Ul5HYEDvkta6I8/rnYM5gSdSV2tJ6XbZuEtY= github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932/go.mod h1:NOuUCSz6Q9T7+igc/hlvDOUdtWKryOrtFyIVABv/p7k= -github.com/bluesky-social/indigo v0.0.0-20251014222321-1e8718ae9f33 h1:x06Y6VyYUCvqWl2AS4/3NBBbRf8wWNMd3YrI44NTHS8= -github.com/bluesky-social/indigo v0.0.0-20251014222321-1e8718ae9f33/go.mod h1:GuGAU33qKulpZCZNPcUeIQ4RW6KzNvOy7s8MSUXbAng= +github.com/bluesky-social/indigo v0.0.0-20251021193747-543ab1124beb h1:zzyqB1W/itfdIA5cnOZ7IFCJ6QtqwOsXltmLunL4sHw= +github.com/bluesky-social/indigo v0.0.0-20251021193747-543ab1124beb/go.mod h1:GuGAU33qKulpZCZNPcUeIQ4RW6KzNvOy7s8MSUXbAng= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4= github.com/bshuster-repo/logrus-logstash-hook v1.0.0 h1:e+C0SB5R1pu//O4MQ3f9cFuPGoOVeF2fE4Og9otCc70= @@ -68,8 +70,6 @@ github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHk github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/go-chi/chi/v5 v5.2.3 h1:WQIt9uxdsAbgIYgid+BpYc+liqQZGMHRaUwp0JUcvdE= github.com/go-chi/chi/v5 v5.2.3/go.mod h1:L2yAIGWB3H+phAw1NxKwWM+7eUH/lU8pOMm5hHcoops= -github.com/go-chi/cors v1.2.2 h1:Jmey33TE+b+rB7fT8MUy1u0I4L+NARQlK6LhzKPSyQE= -github.com/go-chi/cors v1.2.2/go.mod h1:sSbTewc+6wYHBBCW7ytsFSn836hqM7JxpglAy2Vzc58= github.com/go-jose/go-jose/v4 v4.1.2 h1:TK/7NqRQZfgAh+Td8AlsrvtPoUyiHh0LqVvokh+1vHI= github.com/go-jose/go-jose/v4 v4.1.2/go.mod h1:22cg9HWM1pOlnRiY+9cQYJ9XHmya1bYW8OeDM6Ku6Oo= github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= @@ -80,6 +80,8 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg= +github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0= github.com/gocql/gocql v1.7.0 h1:O+7U7/1gSN7QTEAaMEsJc1Oq2QHXvCWoF3DFK9HDHus= diff --git a/pkg/hold/oci/xrpc_test.go b/pkg/hold/oci/xrpc_test.go index 8b8dc6c..dfbd4c7 100644 --- a/pkg/hold/oci/xrpc_test.go +++ b/pkg/hold/oci/xrpc_test.go @@ -4,9 +4,11 @@ import ( "bytes" "context" "encoding/json" + "fmt" "io" "net/http" "net/http/httptest" + "os" "path/filepath" "strconv" "testing" @@ -18,6 +20,36 @@ import ( _ "github.com/distribution/distribution/v3/registry/storage/driver/filesystem" ) +// Shared test resources for OCI package +var ( + sharedTestKeyPath string + sharedTestKey []byte +) + +// TestMain sets up shared resources for all OCI tests +func TestMain(m *testing.M) { + // Create a temporary directory for shared test key + tmpDir, err := os.MkdirTemp("", "oci-test-shared-*") + if err != nil { + panic(fmt.Sprintf("Failed to create temp dir: %v", err)) + } + defer os.RemoveAll(tmpDir) + + // Generate one signing key to be reused across all tests + sharedTestKeyPath = filepath.Join(tmpDir, "shared-signing-key") + privateKey, err := pds.GenerateOrLoadKey(sharedTestKeyPath) + if err != nil { + panic(fmt.Sprintf("Failed to generate shared signing key: %v", err)) + } + + // Store the key bytes so tests can copy them + sharedTestKey = privateKey.Bytes() + + // Run tests + code := m.Run() + os.Exit(code) +} + // Test setup helpers // mockPDSClient implements pds.HTTPClient for testing @@ -54,19 +86,38 @@ func setupTestOCIHandler(t *testing.T) (*XRPCHandler, context.Context) { } // Create minimal PDS for DID/auth - dbPath := filepath.Join(tmpDir, "pds.db") + // Use in-memory database for speed + dbPath := ":memory:" keyPath := filepath.Join(tmpDir, "signing-key") holdDID := "did:web:hold.example.com" publicURL := "https://hold.example.com" + // Copy shared signing key instead of generating a new one + if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil { + t.Fatalf("Failed to copy shared signing key: %v", err) + } + holdPDS, err := pds.NewHoldPDS(ctx, holdDID, publicURL, dbPath, keyPath) if err != nil { t.Fatalf("Failed to create PDS: %v", err) } - // Bootstrap PDS + // Bootstrap PDS, suppressing stdout to avoid log spam ownerDID := "did:plc:owner123" - if err := holdPDS.Bootstrap(ctx, nil, ownerDID, true, false, ""); err != nil { + + // Redirect stdout to suppress bootstrap logging + oldStdout := os.Stdout + r, w, _ := os.Pipe() + os.Stdout = w + + err = holdPDS.Bootstrap(ctx, nil, ownerDID, true, false, "") + + // Restore stdout + w.Close() + os.Stdout = oldStdout + io.ReadAll(r) // Drain the pipe + + if err != nil { t.Fatalf("Failed to bootstrap PDS: %v", err) } diff --git a/pkg/hold/pds/apppassword.go b/pkg/hold/pds/apppassword.go deleted file mode 100644 index 4652892..0000000 --- a/pkg/hold/pds/apppassword.go +++ /dev/null @@ -1,138 +0,0 @@ -package pds - -import ( - "crypto/rand" - "encoding/base32" - "fmt" - "strings" - - "golang.org/x/crypto/bcrypt" -) - -// GenerateAppPassword creates a random app password in the format: abcd-efgh-ijkl-mnop -// Uses base32 encoding for readable characters (no ambiguous chars like 0/O, 1/l) -func GenerateAppPassword() (string, error) { - // Generate 20 random bytes (160 bits of entropy) - // Base32 encoding gives us 32 characters, we'll format as 4 groups of 4 - randomBytes := make([]byte, 20) - if _, err := rand.Read(randomBytes); err != nil { - return "", fmt.Errorf("failed to generate random bytes: %w", err) - } - - // Encode as base32 and lowercase (base32 alphabet: a-z, 2-7) - encoded := base32.StdEncoding.EncodeToString(randomBytes) - encoded = strings.ToLower(encoded) - - // Remove padding and take first 16 characters - encoded = strings.TrimRight(encoded, "=") - if len(encoded) > 16 { - encoded = encoded[:16] - } - - // Format as: xxxx-xxxx-xxxx-xxxx - parts := []string{ - encoded[0:4], - encoded[4:8], - encoded[8:12], - encoded[12:16], - } - - return strings.Join(parts, "-"), nil -} - -// HashAppPassword hashes an app password using bcrypt -// Cost is set to 12 for good security without excessive CPU usage -func HashAppPassword(password string) (string, error) { - hash, err := bcrypt.GenerateFromPassword([]byte(password), 12) - if err != nil { - return "", fmt.Errorf("failed to hash password: %w", err) - } - return string(hash), nil -} - -// ValidateAppPassword compares a plaintext password with a bcrypt hash -func ValidateAppPassword(password, hash string) bool { - err := bcrypt.CompareHashAndPassword([]byte(hash), []byte(password)) - return err == nil -} - -// CreateAppPassword generates and stores a new app password -func (p *HoldPDS) CreateAppPassword(name string) (string, error) { - // Generate random password - password, err := GenerateAppPassword() - if err != nil { - return "", fmt.Errorf("failed to generate password: %w", err) - } - - // Hash password - hash, err := HashAppPassword(password) - if err != nil { - return "", fmt.Errorf("failed to hash password: %w", err) - } - - // Store in database - if err := p.authDB.CreateAppPassword(name, hash); err != nil { - return "", fmt.Errorf("failed to store password: %w", err) - } - - return password, nil -} - -// ValidateAppPasswordByName checks if a password matches the stored hash for a given name -func (p *HoldPDS) ValidateAppPasswordByName(name, password string) error { - // Get app password from database - ap, err := p.authDB.GetAppPassword(name) - if err != nil { - return fmt.Errorf("app password not found: %w", err) - } - - // Validate password - if !ValidateAppPassword(password, ap.PasswordHash) { - return fmt.Errorf("invalid password") - } - - // Update last used timestamp - if err := p.authDB.UpdateLastUsed(name); err != nil { - // Log but don't fail - this is not critical - fmt.Printf("Warning: failed to update last used timestamp: %v\n", err) - } - - return nil -} - -// ValidateAnyAppPassword checks if a password matches any stored app password -// Returns the name of the matching app password, or error if none match -func (p *HoldPDS) ValidateAnyAppPassword(password string) (string, error) { - // List all app passwords - passwords, err := p.authDB.ListAppPasswords() - if err != nil { - return "", fmt.Errorf("failed to list app passwords: %w", err) - } - - // Try each one - for _, ap := range passwords { - // Get full record with hash - fullAP, err := p.authDB.GetAppPassword(ap.Name) - if err != nil { - continue - } - - if ValidateAppPassword(password, fullAP.PasswordHash) { - // Update last used - p.authDB.UpdateLastUsed(ap.Name) - return ap.Name, nil - } - } - - return "", fmt.Errorf("invalid app password") -} - -// ListAppPasswords returns a list of app password names (without hashes) -func (p *HoldPDS) ListAppPasswords() ([]AppPassword, error) { - return p.authDB.ListAppPasswords() -} - -// RevokeAppPassword deletes an app password -func (p *HoldPDS) RevokeAppPassword(name string) error { - return p.authDB.DeleteAppPassword(name) -} diff --git a/pkg/hold/pds/auth.go b/pkg/hold/pds/auth.go index daa30bc..a6d9889 100644 --- a/pkg/hold/pds/auth.go +++ b/pkg/hold/pds/auth.go @@ -528,33 +528,3 @@ func fetchPublicKeyFromDID(ctx context.Context, did string, httpClient HTTPClien return publicKey, nil } - -// ValidateJWTAuth validates a request with a JWT access token from createSession -// This is used for authenticated repo operations (createRecord, etc.) -// Returns the validated user DID -func ValidateJWTAuth(r *http.Request, pds *HoldPDS) (*ValidatedUser, error) { - // Extract Authorization header - authHeader := r.Header.Get("Authorization") - if authHeader == "" { - return nil, fmt.Errorf("missing Authorization header") - } - - // Remove "Bearer " prefix - accessToken := strings.TrimPrefix(authHeader, "Bearer ") - if accessToken == authHeader { - return nil, fmt.Errorf("invalid authorization header format (expected Bearer)") - } - - // Validate access token - claims, err := pds.ValidateAccessToken(accessToken) - if err != nil { - return nil, fmt.Errorf("invalid access token: %w", err) - } - - return &ValidatedUser{ - DID: claims.DID, - Handle: claims.Handle, - PDS: "", - Authorized: true, - }, nil -} diff --git a/pkg/hold/pds/auth_test.go b/pkg/hold/pds/auth_test.go index 08ece18..087af70 100644 --- a/pkg/hold/pds/auth_test.go +++ b/pkg/hold/pds/auth_test.go @@ -506,16 +506,10 @@ func TestValidateServiceToken_WrongScheme(t *testing.T) { // TestValidateBlobWriteAccess_ServiceToken_Owner tests owner write access via service token func TestValidateBlobWriteAccess_ServiceToken_Owner(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" holdDID := "did:web:hold01.atcr.io" - // Bootstrap with owner - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + _, _ = setupTestPDSWithBootstrap(t, ownerDID, true, false) // Create service token for owner helper, err := NewServiceTokenTestHelper(ownerDID, holdDID) @@ -539,20 +533,14 @@ func TestValidateBlobWriteAccess_ServiceToken_Owner(t *testing.T) { // TestValidateBlobWriteAccess_ServiceToken_CrewWithPermission tests crew write access via service token func TestValidateBlobWriteAccess_ServiceToken_CrewWithPermission(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" writerDID := "did:plc:writer123" holdDID := "did:web:hold01.atcr.io" - // Bootstrap - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, true, false) // Add crew member with blob:write permission - _, err = pds.AddCrewMember(ctx, writerDID, "writer", []string{"blob:write"}) + _, err := pds.AddCrewMember(ctx, writerDID, "writer", []string{"blob:write"}) if err != nil { t.Fatalf("Failed to add crew member: %v", err) } @@ -598,20 +586,14 @@ func TestValidateBlobWriteAccess_ServiceToken_CrewWithPermission(t *testing.T) { // TestValidateBlobWriteAccess_ServiceToken_CrewWithoutPermission tests that crew without permission is rejected func TestValidateBlobWriteAccess_ServiceToken_CrewWithoutPermission(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" readerDID := "did:plc:reader123" holdDID := "did:web:hold01.atcr.io" - // Bootstrap - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, true, false) // Add crew member with blob:read permission only (no blob:write) - _, err = pds.AddCrewMember(ctx, readerDID, "reader", []string{"blob:read"}) + _, err := pds.AddCrewMember(ctx, readerDID, "reader", []string{"blob:read"}) if err != nil { t.Fatalf("Failed to add crew member: %v", err) } @@ -645,15 +627,9 @@ func TestValidateBlobWriteAccess_ServiceToken_CrewWithoutPermission(t *testing.T // TestValidateBlobWriteAccess_Owner tests that the hold owner has write access func TestValidateBlobWriteAccess_Owner(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" - // Bootstrap with owner - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, _ := setupTestPDSWithBootstrap(t, ownerDID, true, false) // Create DPoP helper for owner dpopHelper, err := NewDPoPTestHelper(ownerDID, "https://test-pds.example.com") @@ -691,19 +667,13 @@ func TestValidateBlobWriteAccess_Owner(t *testing.T) { // TestValidateBlobWriteAccess_CrewPermissions tests crew permission checking func TestValidateBlobWriteAccess_CrewPermissions(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" - // Bootstrap - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, true, false) // Add crew member with blob:write permission writerDID := "did:plc:writer123" - _, err = pds.AddCrewMember(ctx, writerDID, "writer", []string{"blob:write"}) + _, err := pds.AddCrewMember(ctx, writerDID, "writer", []string{"blob:write"}) if err != nil { t.Fatalf("Failed to add crew member: %v", err) } @@ -764,15 +734,9 @@ func TestValidateBlobWriteAccess_CrewPermissions(t *testing.T) { // TestValidateBlobReadAccess_PublicHold tests public hold access func TestValidateBlobReadAccess_PublicHold(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" - // Bootstrap with public=true - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, true, false) // Verify captain record has public=true _, captain, err := pds.GetCaptainRecord(ctx) @@ -801,18 +765,12 @@ func TestValidateBlobReadAccess_PublicHold(t *testing.T) { // TestValidateBlobReadAccess_PrivateHold tests private hold access func TestValidateBlobReadAccess_PrivateHold(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" - // Bootstrap with public=false - err := pds.Bootstrap(ctx, nil, ownerDID, false, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, false, false) // Update captain to be private - _, err = pds.UpdateCaptainRecord(ctx, false, false) + _, err := pds.UpdateCaptainRecord(ctx, false, false) if err != nil { t.Fatalf("Failed to update captain record: %v", err) } @@ -843,19 +801,13 @@ func TestValidateBlobReadAccess_PrivateHold(t *testing.T) { // TestValidateOwnerOrCrewAdmin tests admin permission checking func TestValidateOwnerOrCrewAdmin(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" - // Bootstrap - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, true, false) // Add crew member with crew:admin permission adminDID := "did:plc:admin123" - _, err = pds.AddCrewMember(ctx, adminDID, "admin", []string{"crew:admin", "blob:write", "blob:read"}) + _, err := pds.AddCrewMember(ctx, adminDID, "admin", []string{"crew:admin", "blob:write", "blob:read"}) if err != nil { t.Fatalf("Failed to add crew admin: %v", err) } @@ -911,15 +863,9 @@ func TestValidateOwnerOrCrewAdmin(t *testing.T) { // TestCrewPermissions tests various permission combinations func TestCrewPermissions(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" - // Bootstrap - err := pds.Bootstrap(ctx, nil, ownerDID, true, false, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, true, false) tests := []struct { name string @@ -1035,15 +981,9 @@ func TestCaptainRecordSettings(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - pds, ctx := setupTestPDS(t) - ownerDID := "did:plc:owner123" - // Bootstrap with specified settings - err := pds.Bootstrap(ctx, nil, ownerDID, tt.public, tt.allowAllCrew, "") - if err != nil { - t.Fatalf("Failed to bootstrap PDS: %v", err) - } + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, tt.public, tt.allowAllCrew) // Verify captain record has expected settings _, captain, err := pds.GetCaptainRecord(ctx) diff --git a/pkg/hold/pds/captain_test.go b/pkg/hold/pds/captain_test.go index 4db3fec..6fcef82 100644 --- a/pkg/hold/pds/captain_test.go +++ b/pkg/hold/pds/captain_test.go @@ -3,6 +3,8 @@ package pds import ( "bytes" "context" + "io" + "os" "path/filepath" "strings" "testing" @@ -17,9 +19,15 @@ func setupTestPDS(t *testing.T) (*HoldPDS, context.Context) { ctx := context.Background() tmpDir := t.TempDir() - dbPath := filepath.Join(tmpDir, "pds.db") + // Use in-memory database for speed + dbPath := ":memory:" keyPath := filepath.Join(tmpDir, "signing-key") + // Copy shared signing key instead of generating a new one + if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil { + t.Fatalf("Failed to copy shared signing key: %v", err) + } + pds, err := NewHoldPDS(ctx, "did:web:hold.example.com", "https://hold.example.com", dbPath, keyPath) if err != nil { t.Fatalf("Failed to create test PDS: %v", err) @@ -35,6 +43,31 @@ func setupTestPDS(t *testing.T) (*HoldPDS, context.Context) { return pds, ctx } +// setupTestPDSWithBootstrap creates a test PDS and bootstraps it with suppressed output +// This is a convenience function for tests that need a fully initialized PDS +func setupTestPDSWithBootstrap(t *testing.T, ownerDID string, public, allowAllCrew bool) (*HoldPDS, context.Context) { + t.Helper() + + pds, ctx := setupTestPDS(t) + + // Bootstrap with suppressed output + oldStdout := os.Stdout + r, w, _ := os.Pipe() + os.Stdout = w + + err := pds.Bootstrap(ctx, nil, ownerDID, public, allowAllCrew, "") + + w.Close() + os.Stdout = oldStdout + io.ReadAll(r) // Drain the pipe + + if err != nil { + t.Fatalf("Failed to bootstrap PDS: %v", err) + } + + return pds, ctx +} + // TestCreateCaptainRecord tests creating a captain record with various settings func TestCreateCaptainRecord(t *testing.T) { tests := []struct { diff --git a/pkg/hold/pds/database.go b/pkg/hold/pds/database.go deleted file mode 100644 index 9903716..0000000 --- a/pkg/hold/pds/database.go +++ /dev/null @@ -1,232 +0,0 @@ -package pds - -import ( - "database/sql" - "fmt" - "os" - "path/filepath" - "time" - - _ "github.com/mattn/go-sqlite3" -) - -// Database manages app passwords and sessions for the hold PDS -type Database struct { - db *sql.DB -} - -// NewDatabase creates or opens a database for app passwords and sessions -// dbPath should be the directory path (same as carstore) -// It creates a separate "auth.db" file for authentication data -func NewDatabase(dbPath string) (*Database, error) { - // Ensure directory exists - if err := os.MkdirAll(dbPath, 0755); err != nil { - return nil, fmt.Errorf("failed to create database directory: %w", err) - } - - // Create auth database file alongside carstore database - authDBFile := filepath.Join(dbPath, "auth.db") - - db, err := sql.Open("sqlite3", authDBFile) - if err != nil { - return nil, fmt.Errorf("failed to open database: %w", err) - } - - // Create tables - if err := createTables(db); err != nil { - db.Close() - return nil, fmt.Errorf("failed to create tables: %w", err) - } - - return &Database{db: db}, nil -} - -// createTables creates the database schema -func createTables(db *sql.DB) error { - schema := ` - CREATE TABLE IF NOT EXISTS app_passwords ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL UNIQUE, - password_hash TEXT NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_used_at TIMESTAMP - ); - - CREATE INDEX IF NOT EXISTS idx_app_passwords_name ON app_passwords(name); - - CREATE TABLE IF NOT EXISTS refresh_tokens ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - token_hash TEXT NOT NULL UNIQUE, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - expires_at TIMESTAMP NOT NULL, - last_used_at TIMESTAMP - ); - - CREATE INDEX IF NOT EXISTS idx_refresh_tokens_hash ON refresh_tokens(token_hash); - CREATE INDEX IF NOT EXISTS idx_refresh_tokens_expires ON refresh_tokens(expires_at); - ` - - _, err := db.Exec(schema) - return err -} - -// Close closes the database connection -func (d *Database) Close() error { - return d.db.Close() -} - -// AppPassword represents an app password record -type AppPassword struct { - ID int64 - Name string - PasswordHash string - CreatedAt time.Time - LastUsedAt *time.Time -} - -// CreateAppPassword stores a new app password -func (d *Database) CreateAppPassword(name, passwordHash string) error { - query := `INSERT INTO app_passwords (name, password_hash) VALUES (?, ?)` - _, err := d.db.Exec(query, name, passwordHash) - if err != nil { - return fmt.Errorf("failed to create app password: %w", err) - } - return nil -} - -// GetAppPassword retrieves an app password by name -func (d *Database) GetAppPassword(name string) (*AppPassword, error) { - query := `SELECT id, name, password_hash, created_at, last_used_at FROM app_passwords WHERE name = ?` - - var ap AppPassword - var lastUsedAt sql.NullTime - - err := d.db.QueryRow(query, name).Scan( - &ap.ID, - &ap.Name, - &ap.PasswordHash, - &ap.CreatedAt, - &lastUsedAt, - ) - - if err == sql.ErrNoRows { - return nil, fmt.Errorf("app password not found") - } - if err != nil { - return nil, fmt.Errorf("failed to get app password: %w", err) - } - - if lastUsedAt.Valid { - ap.LastUsedAt = &lastUsedAt.Time - } - - return &ap, nil -} - -// ListAppPasswords returns all app passwords (without hashes) -func (d *Database) ListAppPasswords() ([]AppPassword, error) { - query := `SELECT id, name, created_at, last_used_at FROM app_passwords ORDER BY created_at DESC` - - rows, err := d.db.Query(query) - if err != nil { - return nil, fmt.Errorf("failed to list app passwords: %w", err) - } - defer rows.Close() - - var passwords []AppPassword - for rows.Next() { - var ap AppPassword - var lastUsedAt sql.NullTime - - if err := rows.Scan(&ap.ID, &ap.Name, &ap.CreatedAt, &lastUsedAt); err != nil { - return nil, fmt.Errorf("failed to scan row: %w", err) - } - - if lastUsedAt.Valid { - ap.LastUsedAt = &lastUsedAt.Time - } - - passwords = append(passwords, ap) - } - - return passwords, rows.Err() -} - -// UpdateLastUsed updates the last used timestamp for an app password -func (d *Database) UpdateLastUsed(name string) error { - query := `UPDATE app_passwords SET last_used_at = CURRENT_TIMESTAMP WHERE name = ?` - _, err := d.db.Exec(query, name) - return err -} - -// DeleteAppPassword removes an app password -func (d *Database) DeleteAppPassword(name string) error { - query := `DELETE FROM app_passwords WHERE name = ?` - result, err := d.db.Exec(query, name) - if err != nil { - return fmt.Errorf("failed to delete app password: %w", err) - } - - rows, err := result.RowsAffected() - if err != nil { - return fmt.Errorf("failed to check rows affected: %w", err) - } - - if rows == 0 { - return fmt.Errorf("app password not found") - } - - return nil -} - -// CreateRefreshToken stores a refresh token -func (d *Database) CreateRefreshToken(tokenHash string, expiresAt time.Time) error { - query := `INSERT INTO refresh_tokens (token_hash, expires_at) VALUES (?, ?)` - _, err := d.db.Exec(query, tokenHash, expiresAt) - if err != nil { - return fmt.Errorf("failed to create refresh token: %w", err) - } - return nil -} - -// ValidateRefreshToken checks if a refresh token exists and is not expired -func (d *Database) ValidateRefreshToken(tokenHash string) (bool, error) { - query := `SELECT expires_at FROM refresh_tokens WHERE token_hash = ?` - - var expiresAt time.Time - err := d.db.QueryRow(query, tokenHash).Scan(&expiresAt) - - if err == sql.ErrNoRows { - return false, nil - } - if err != nil { - return false, fmt.Errorf("failed to validate refresh token: %w", err) - } - - // Check if expired - if time.Now().After(expiresAt) { - // Delete expired token - d.DeleteRefreshToken(tokenHash) - return false, nil - } - - // Update last used - updateQuery := `UPDATE refresh_tokens SET last_used_at = CURRENT_TIMESTAMP WHERE token_hash = ?` - d.db.Exec(updateQuery, tokenHash) - - return true, nil -} - -// DeleteRefreshToken removes a refresh token -func (d *Database) DeleteRefreshToken(tokenHash string) error { - query := `DELETE FROM refresh_tokens WHERE token_hash = ?` - _, err := d.db.Exec(query, tokenHash) - return err -} - -// CleanupExpiredTokens removes all expired refresh tokens -func (d *Database) CleanupExpiredTokens() error { - query := `DELETE FROM refresh_tokens WHERE expires_at < CURRENT_TIMESTAMP` - _, err := d.db.Exec(query) - return err -} diff --git a/pkg/hold/pds/events.go b/pkg/hold/pds/events.go index a062b2a..8570efd 100644 --- a/pkg/hold/pds/events.go +++ b/pkg/hold/pds/events.go @@ -8,8 +8,10 @@ import ( "time" atproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/events" lexutil "github.com/bluesky-social/indigo/lex/util" "github.com/gorilla/websocket" + "github.com/ipfs/go-cid" ) // EventBroadcaster manages WebSocket connections and broadcasts repo events @@ -77,7 +79,9 @@ func (b *EventBroadcaster) Subscribe(conn *websocket.Conn, cursor int64) *Subscr b.mu.Unlock() // Send historical events if cursor is provided and < current seq - if cursor > 0 && cursor < currentSeq { + // cursor=0 means "replay all events from the beginning" + // cursor >= 0 triggers backfill, negative cursor means "no backfill" + if cursor >= 0 && cursor < currentSeq { go b.backfillSubscriber(sub, cursor) } @@ -205,17 +209,40 @@ func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) { }() for event := range sub.send { - // Encode as CBOR - cborBytes, err := encodeCBOR(event) - if err != nil { - log.Printf("Failed to encode event as CBOR: %v", err) - continue + // Create event header (ATProto firehose format) + header := events.EventHeader{ + Op: events.EvtKindMessage, + MsgType: "#commit", } - // Write CBOR message to WebSocket - err = sub.conn.WriteMessage(websocket.BinaryMessage, cborBytes) + // Get a writer for this message + wc, err := sub.conn.NextWriter(websocket.BinaryMessage) if err != nil { - log.Printf("Failed to write to websocket: %v", err) + log.Printf("Failed to get websocket writer: %v", err) + return + } + + // Write header as CBOR + if err := header.MarshalCBOR(wc); err != nil { + log.Printf("Failed to write event header: %v", err) + 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 { + log.Printf("Failed to write event body: %v", err) + wc.Close() + return + } + + // Close the writer to flush the message + if err := wc.Close(); err != nil { + log.Printf("Failed to close websocket writer: %v", err) return } @@ -224,14 +251,42 @@ func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) { } } -// encodeCBOR encodes an event as CBOR -func encodeCBOR(event *RepoCommitEvent) ([]byte, error) { - // For now, use JSON encoding wrapped in CBOR envelope - // In production, you'd use proper CBOR encoding - // The atproto spec requires DAG-CBOR with specific header +// convertToIndigoCommit converts our RepoCommitEvent to indigo's SyncSubscribeRepos_Commit +// which has proper CBOR marshaling methods generated +func convertToIndigoCommit(event *RepoCommitEvent) *atproto.SyncSubscribeRepos_Commit { + // Parse commit CID string to cid.Cid, then convert to LexLink + commitCID, err := cid.Decode(event.Commit) + if err != nil { + log.Printf("Warning: failed to parse commit CID %s: %v", event.Commit, err) + // Create an empty CID as fallback + commitCID = cid.Undef + } - // Simple approach: encode as JSON for MVP - // Real implementation needs proper CBOR-gen types + // Convert cid.Cid to LexLink + commitLink := lexutil.LexLink(commitCID) + + // Convert blocks to LexBytes + blocks := lexutil.LexBytes(event.Blocks) + + return &atproto.SyncSubscribeRepos_Commit{ + Seq: event.Seq, + Repo: event.Repo, + Commit: commitLink, + Rev: event.Rev, + Since: event.Since, + Blocks: blocks, + Ops: event.Ops, + Time: event.Time, + Blobs: []lexutil.LexLink{}, // Empty for now, we don't track blob refs in our simplified model + Rebase: false, // DEPRECATED field + TooBig: false, // Not implementing tooBig for now + } +} + +// encodeCBOR encodes an event as CBOR (DEPRECATED - kept for tests) +func encodeCBOR(event *RepoCommitEvent) ([]byte, error) { + // For backward compatibility with tests, encode as JSON + // Production code uses convertToIndigoCommit + CBOR marshaling in handleSubscriber return json.Marshal(event) } diff --git a/pkg/hold/pds/events_test.go b/pkg/hold/pds/events_test.go index 161e176..8cd9f5b 100644 --- a/pkg/hold/pds/events_test.go +++ b/pkg/hold/pds/events_test.go @@ -382,3 +382,167 @@ func TestEncodeCBOR(t *testing.T) { t.Errorf("Expected decoded seq=1, got %d", decoded.Seq) } } + +// TestSubscribe_CursorZeroBackfill tests that cursor=0 replays all events +func TestSubscribe_CursorZeroBackfill(t *testing.T) { + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100) + ctx := context.Background() + + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + + // Broadcast 5 events before subscribing + for i := 1; i <= 5; i++ { + event := &RepoEvent{ + NewRoot: testCID, + Rev: "test-rev", + RepoSlice: []byte("test CAR data"), + Ops: []RepoOp{}, + } + broadcaster.Broadcast(ctx, event) + } + + // Verify we have 5 events in history + if broadcaster.eventSeq != 5 { + t.Fatalf("Expected eventSeq=5, got %d", broadcaster.eventSeq) + } + + // Create mock websocket connection (we won't actually use it) + // We just need to verify backfillSubscriber is called + // For this test, we'll check the history directly + if len(broadcaster.eventHistory) != 5 { + t.Errorf("Expected 5 events in history, got %d", len(broadcaster.eventHistory)) + } + + // Verify all events have sequential sequence numbers + for i, he := range broadcaster.eventHistory { + expectedSeq := int64(i + 1) + if he.Seq != expectedSeq { + t.Errorf("Expected history[%d].Seq=%d, got %d", i, expectedSeq, he.Seq) + } + } + + // 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 + cursor: 0, + } + + // Run backfill in a goroutine + go broadcaster.backfillSubscriber(sub, 0) + + // Wait for events to be sent + time.Sleep(100 * time.Millisecond) + + // Should receive all 5 events + receivedCount := len(sub.send) + if receivedCount != 5 { + t.Errorf("Expected to receive 5 events with cursor=0, got %d", receivedCount) + } + + // Verify events are in order + 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) + } + default: + t.Errorf("Expected event %d but channel was empty", i) + } + } +} + +// TestSubscribe_MidCursorBackfill tests that cursor=N only gets events after N +func TestSubscribe_MidCursorBackfill(t *testing.T) { + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100) + ctx := context.Background() + + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + + // Broadcast 10 events before subscribing + for i := 1; i <= 10; i++ { + event := &RepoEvent{ + NewRoot: testCID, + Rev: "test-rev", + RepoSlice: []byte("test CAR data"), + Ops: []RepoOp{}, + } + broadcaster.Broadcast(ctx, event) + } + + // 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 + cursor: 5, + } + + // Run backfill + go broadcaster.backfillSubscriber(sub, 5) + + // Wait for events to be sent + time.Sleep(100 * time.Millisecond) + + // Should receive events 6-10 (5 events after cursor=5) + receivedCount := len(sub.send) + if receivedCount != 5 { + t.Errorf("Expected to receive 5 events with cursor=5, got %d", receivedCount) + } + + // Verify events start at seq=6 + 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) + } + default: + t.Errorf("Expected event %d but channel was empty", i) + } + } +} + +// TestSubscribe_NegativeCursorNoBackfill tests that negative cursor means no backfill +func TestSubscribe_NegativeCursorNoBackfill(t *testing.T) { + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 100) + ctx := context.Background() + + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + + // Broadcast 5 events before subscribing + for i := 1; i <= 5; i++ { + event := &RepoEvent{ + NewRoot: testCID, + Rev: "test-rev", + RepoSlice: []byte("test CAR data"), + Ops: []RepoOp{}, + } + broadcaster.Broadcast(ctx, event) + } + + // Create subscriber with cursor=-1 (no backfill, conn not needed) + sub := &Subscriber{ + conn: nil, // Not used in this test + send: make(chan *RepoCommitEvent, 100), + cursor: -1, + } + + // Subscribe should not trigger backfill + broadcaster.mu.Lock() + currentSeq := broadcaster.eventSeq + broadcaster.mu.Unlock() + + // Check the condition: cursor >= 0 && cursor < currentSeq + // For cursor=-1, this should be false + shouldBackfill := -1 >= 0 && -1 < currentSeq + if shouldBackfill { + t.Error("Expected shouldBackfill=false for cursor=-1, but condition evaluated to true") + } + + // Verify no events in send channel (no backfill happened) + if len(sub.send) != 0 { + t.Errorf("Expected 0 events with cursor=-1 (no backfill), got %d", len(sub.send)) + } +} diff --git a/pkg/hold/pds/jwt.go b/pkg/hold/pds/jwt.go deleted file mode 100644 index 4038798..0000000 --- a/pkg/hold/pds/jwt.go +++ /dev/null @@ -1,249 +0,0 @@ -package pds - -import ( - "crypto/sha256" - "encoding/base64" - "encoding/hex" - "encoding/json" - "fmt" - "time" -) - -// Session token types -const ( - TokenTypeAccess = "access" - TokenTypeRefresh = "refresh" -) - -// Token expiration durations -const ( - AccessTokenDuration = 2 * time.Hour // Short-lived access token - RefreshTokenDuration = 90 * 24 * time.Hour // Long-lived refresh token (90 days) -) - -// SessionClaims represents JWT claims for ATProto sessions -type SessionClaims struct { - DID string `json:"sub"` // Subject (DID) - Issuer string `json:"iss"` // Issuer (PDS DID) - Handle string `json:"handle,omitempty"` - Scope string `json:"scope"` - TokenType string `json:"token_type"` - IssuedAt int64 `json:"iat"` // Unix timestamp - ExpiresAt int64 `json:"exp"` // Unix timestamp -} - -// IssueAccessToken creates a new access JWT for a session -func (p *HoldPDS) IssueAccessToken(did, handle string) (string, error) { - now := time.Now() - claims := &SessionClaims{ - DID: did, - Issuer: p.did, - Handle: handle, - Scope: "com.atproto.access", - TokenType: TokenTypeAccess, - IssuedAt: now.Unix(), - ExpiresAt: now.Add(AccessTokenDuration).Unix(), - } - - return p.signJWT(claims) -} - -// IssueRefreshToken creates a new refresh JWT for a session -func (p *HoldPDS) IssueRefreshToken(did, handle string) (string, error) { - now := time.Now() - claims := &SessionClaims{ - DID: did, - Issuer: p.did, - Handle: handle, - Scope: "com.atproto.refresh", - TokenType: TokenTypeRefresh, - IssuedAt: now.Unix(), - ExpiresAt: now.Add(RefreshTokenDuration).Unix(), - } - - signedToken, err := p.signJWT(claims) - if err != nil { - return "", err - } - - // Store refresh token hash in database for validation/revocation - tokenHash := hashToken(signedToken) - expiresAt := now.Add(RefreshTokenDuration) - if err := p.authDB.CreateRefreshToken(tokenHash, expiresAt); err != nil { - return "", fmt.Errorf("failed to store refresh token: %w", err) - } - - return signedToken, nil -} - -// ValidateAccessToken validates an access JWT and returns the claims -func (p *HoldPDS) ValidateAccessToken(tokenString string) (*SessionClaims, error) { - return p.validateToken(tokenString, TokenTypeAccess) -} - -// ValidateRefreshToken validates a refresh JWT and returns the claims -// Also checks the database to ensure the token hasn't been revoked -func (p *HoldPDS) ValidateRefreshToken(tokenString string) (*SessionClaims, error) { - // First validate signature and claims - claims, err := p.validateToken(tokenString, TokenTypeRefresh) - if err != nil { - return nil, err - } - - // Check if token is in database (not revoked) - tokenHash := hashToken(tokenString) - valid, err := p.authDB.ValidateRefreshToken(tokenHash) - if err != nil { - return nil, fmt.Errorf("failed to validate refresh token in database: %w", err) - } - if !valid { - return nil, fmt.Errorf("refresh token has been revoked or expired") - } - - return claims, nil -} - -// validateToken validates a JWT token and returns the claims -func (p *HoldPDS) validateToken(tokenString, expectedType string) (*SessionClaims, error) { - // Split token into parts - parts := splitJWT(tokenString) - if len(parts) != 3 { - return nil, fmt.Errorf("invalid JWT format") - } - - // Decode header - headerBytes, err := base64.RawURLEncoding.DecodeString(parts[0]) - if err != nil { - return nil, fmt.Errorf("failed to decode header: %w", err) - } - - var header map[string]interface{} - if err := json.Unmarshal(headerBytes, &header); err != nil { - return nil, fmt.Errorf("failed to parse header: %w", err) - } - - // Verify algorithm - alg, ok := header["alg"].(string) - if !ok || alg != "ES256K" { - return nil, fmt.Errorf("unsupported algorithm: %v", alg) - } - - // Decode claims - claimsBytes, err := base64.RawURLEncoding.DecodeString(parts[1]) - if err != nil { - return nil, fmt.Errorf("failed to decode claims: %w", err) - } - - var claims SessionClaims - if err := json.Unmarshal(claimsBytes, &claims); err != nil { - return nil, fmt.Errorf("failed to parse claims: %w", err) - } - - // Verify token type - if claims.TokenType != expectedType { - return nil, fmt.Errorf("invalid token type: expected %s, got %s", expectedType, claims.TokenType) - } - - // Verify issuer - if claims.Issuer != p.did { - return nil, fmt.Errorf("invalid issuer: expected %s, got %s", p.did, claims.Issuer) - } - - // Verify subject matches this hold - if claims.DID != p.did { - return nil, fmt.Errorf("invalid subject: expected %s, got %s", p.did, claims.DID) - } - - // Verify expiration - if time.Now().Unix() > claims.ExpiresAt { - return nil, fmt.Errorf("token has expired") - } - - // Verify signature - signedData := []byte(parts[0] + "." + parts[1]) - signature, err := base64.RawURLEncoding.DecodeString(parts[2]) - if err != nil { - return nil, fmt.Errorf("failed to decode signature: %w", err) - } - - publicKey, err := p.signingKey.PublicKey() - if err != nil { - return nil, fmt.Errorf("failed to get public key: %w", err) - } - - if err := publicKey.HashAndVerify(signedData, signature); err != nil { - return nil, fmt.Errorf("signature verification failed: %w", err) - } - - return &claims, nil -} - -// RevokeRefreshToken revokes a refresh token by removing it from the database -func (p *HoldPDS) RevokeRefreshToken(tokenString string) error { - tokenHash := hashToken(tokenString) - return p.authDB.DeleteRefreshToken(tokenHash) -} - -// hashToken creates a SHA-256 hash of a token for storage -func hashToken(token string) string { - hash := sha256.Sum256([]byte(token)) - return hex.EncodeToString(hash[:]) -} - -// signJWT creates and signs a JWT using the hold's private key -func (p *HoldPDS) signJWT(claims *SessionClaims) (string, error) { - // Create header - header := map[string]interface{}{ - "typ": "JWT", - "alg": "ES256K", - } - - headerJSON, err := json.Marshal(header) - if err != nil { - return "", fmt.Errorf("failed to marshal header: %w", err) - } - - // Create payload - payloadJSON, err := json.Marshal(claims) - if err != nil { - return "", fmt.Errorf("failed to marshal claims: %w", err) - } - - // Base64url encode header and payload - headerEncoded := base64.RawURLEncoding.EncodeToString(headerJSON) - payloadEncoded := base64.RawURLEncoding.EncodeToString(payloadJSON) - - // Create signing input - signingInput := headerEncoded + "." + payloadEncoded - - // Sign with private key - signature, err := p.signingKey.HashAndSign([]byte(signingInput)) - if err != nil { - return "", fmt.Errorf("failed to sign JWT: %w", err) - } - - // Base64url encode signature - signatureEncoded := base64.RawURLEncoding.EncodeToString(signature) - - // Combine into final JWT - jwt := signingInput + "." + signatureEncoded - - return jwt, nil -} - -// splitJWT splits a JWT string into its three parts -func splitJWT(token string) []string { - // JWT format: header.payload.signature - parts := make([]string, 0, 3) - start := 0 - for i := 0; i < len(token); i++ { - if token[i] == '.' { - parts = append(parts, token[start:i]) - start = i + 1 - } - } - if start < len(token) { - parts = append(parts, token[start:]) - } - return parts -} diff --git a/pkg/hold/pds/server.go b/pkg/hold/pds/server.go index 8892619..f8f0b7f 100644 --- a/pkg/hold/pds/server.go +++ b/pkg/hold/pds/server.go @@ -36,28 +36,37 @@ type HoldPDS struct { dbPath string uid models.Uid signingKey *atcrypto.PrivateKeyK256 - authDB *Database // Authentication database for app passwords and sessions } // NewHoldPDS creates or opens a hold PDS with SQLite carstore func NewHoldPDS(ctx context.Context, did, publicURL, dbPath, keyPath string) (*HoldPDS, error) { - // Ensure directory exists - dir := filepath.Dir(dbPath) - if err := os.MkdirAll(dir, 0755); err != nil { - return nil, fmt.Errorf("failed to create database directory: %w", err) - } - // Generate or load signing key signingKey, err := GenerateOrLoadKey(keyPath) if err != nil { return nil, fmt.Errorf("failed to initialize signing key: %w", err) } - // Create and open SQLite-backed carstore - // dbPath is the directory, carstore creates and opens db.sqlite3 inside it - sqlStore, err := carstore.NewSqliteStore(dbPath) - if err != nil { - return nil, fmt.Errorf("failed to create sqlite store: %w", err) + // Create SQLite-backed carstore + var sqlStore *carstore.SQLiteStore + + if dbPath == ":memory:" { + // In-memory mode for tests: create carstore manually and open with :memory: + sqlStore = new(carstore.SQLiteStore) + if err := sqlStore.Open(":memory:"); err != nil { + return nil, fmt.Errorf("failed to open in-memory sqlite store: %w", err) + } + } else { + // File mode for production: create directory and use NewSqliteStore + dir := filepath.Dir(dbPath) + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, fmt.Errorf("failed to create database directory: %w", err) + } + + // dbPath is the directory, carstore creates and opens db.sqlite3 inside it + sqlStore, err = carstore.NewSqliteStore(dbPath) + if err != nil { + return nil, fmt.Errorf("failed to create sqlite store: %w", err) + } } // Use SQLiteStore directly, not the CarStore() wrapper @@ -84,12 +93,6 @@ func NewHoldPDS(ctx context.Context, did, publicURL, dbPath, keyPath string) (*H fmt.Printf("New hold repo - will be initialized in Bootstrap\n") } - // Create or open authentication database - authDB, err := NewDatabase(dbPath) - if err != nil { - return nil, fmt.Errorf("failed to create auth database: %w", err) - } - return &HoldPDS{ did: did, PublicURL: publicURL, @@ -98,7 +101,6 @@ func NewHoldPDS(ctx context.Context, did, publicURL, dbPath, keyPath string) (*H dbPath: dbPath, uid: uid, signingKey: signingKey, - authDB: authDB, }, nil } @@ -184,54 +186,6 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri } else { fmt.Printf("✅ Bluesky profile record already exists, skipping\n") } - - // Create Tangled profile record (idempotent - check if exists first) - _, _, err = p.GetTangledProfileRecord(ctx) - if err != nil { - // Tangled profile doesn't exist, create it - description := "ahoy from the cargo hold" - links := []string{"https://atcr.io"} - - _, err = p.CreateTangledProfileRecord(ctx, links, description) - if err != nil { - return fmt.Errorf("failed to create tangled profile record: %w", err) - } - fmt.Printf("✅ Created Tangled profile record\n") - } else { - fmt.Printf("✅ Tangled profile record already exists, skipping\n") - } - } - - // Create bootstrap app password if none exist (one-time setup) - passwords, err := p.authDB.ListAppPasswords() - if err != nil { - return fmt.Errorf("failed to list app passwords: %w", err) - } - - if len(passwords) == 0 { - // No app passwords exist, create one - password, err := p.CreateAppPassword("bootstrap") - if err != nil { - return fmt.Errorf("failed to create bootstrap app password: %w", err) - } - - fmt.Printf("\n") - fmt.Printf("╔════════════════════════════════════════════════════════════════╗\n") - fmt.Printf("║ 🔑 APP PASSWORD CREATED ║\n") - fmt.Printf("╠════════════════════════════════════════════════════════════════╣\n") - fmt.Printf("║ ║\n") - fmt.Printf("║ Password: %-51s ║\n", password) - fmt.Printf("║ ║\n") - fmt.Printf("║ ⚠️ SAVE THIS PASSWORD - it will not be shown again ║\n") - fmt.Printf("║ ║\n") - fmt.Printf("║ Use this password to log into Bluesky app or CLI tools ║\n") - fmt.Printf("║ PDS URL: %-51s ║\n", p.PublicURL) - fmt.Printf("║ Username: %-50s ║\n", p.did) - fmt.Printf("║ ║\n") - fmt.Printf("╚════════════════════════════════════════════════════════════════╝\n") - fmt.Printf("\n") - } else { - fmt.Printf("✅ App passwords already exist (count: %d), skipping auto-generation\n", len(passwords)) } return nil diff --git a/pkg/hold/pds/session.go b/pkg/hold/pds/session.go deleted file mode 100644 index abf21de..0000000 --- a/pkg/hold/pds/session.go +++ /dev/null @@ -1,381 +0,0 @@ -package pds - -import ( - "encoding/json" - "fmt" - "io" - "net/http" - "strings" - - cbg "github.com/whyrusleeping/cbor-gen" - "github.com/ipfs/go-cid" -) - -// CreateSessionRequest represents a session creation request -type CreateSessionRequest struct { - Identifier string `json:"identifier"` // DID or handle - Password string `json:"password"` // App password -} - -// CreateSessionResponse represents a successful session creation -type CreateSessionResponse struct { - AccessJwt string `json:"accessJwt"` - RefreshJwt string `json:"refreshJwt"` - Handle string `json:"handle"` - DID string `json:"did"` - DIDDoc map[string]interface{} `json:"didDoc,omitempty"` // Optional DID document - Email string `json:"email,omitempty"` // Optional, not used for holds - Active *bool `json:"active,omitempty"` // Optional account status - Status string `json:"status,omitempty"` // Optional account status -} - -// SessionInfo represents session information -type SessionInfo struct { - Handle string `json:"handle"` - DID string `json:"did"` - Email string `json:"email,omitempty"` -} - -// HandleCreateSession handles com.atproto.server.createSession -func (h *XRPCHandler) HandleCreateSession(w http.ResponseWriter, r *http.Request) { - // Parse request - var req CreateSessionRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest) - return - } - - // Validate required fields - if req.Identifier == "" || req.Password == "" { - http.Error(w, "identifier and password are required", http.StatusBadRequest) - return - } - - // Validate identifier matches this hold's DID or handle - holdDID := h.pds.DID() - // For did:web, handle is the domain part without "did:web:" prefix - // e.g., "did:web:hold01.atcr.io" -> "hold01.atcr.io" - holdHandle := strings.TrimPrefix(holdDID, "did:web:") - - // Normalize the identifier (strip "at://" prefix if present) - identifier := strings.TrimPrefix(req.Identifier, "at://") - - // Accept any of: - // 1. Full DID: "did:web:hold01.atcr.io" - // 2. Handle (domain): "hold01.atcr.io" - // 3. Either with "at://" prefix - isValidIdentifier := identifier == holdDID || identifier == holdHandle - - if !isValidIdentifier { - fmt.Printf("Invalid identifier: got %q, expected DID %q or handle %q\n", req.Identifier, holdDID, holdHandle) - http.Error(w, "invalid identifier", http.StatusUnauthorized) - return - } - - // Validate app password - _, err := h.pds.ValidateAnyAppPassword(req.Password) - if err != nil { - http.Error(w, "invalid password", http.StatusUnauthorized) - return - } - - // Issue access and refresh tokens - accessToken, err := h.pds.IssueAccessToken(holdDID, holdHandle) - if err != nil { - http.Error(w, fmt.Sprintf("failed to issue access token: %v", err), http.StatusInternalServerError) - return - } - - refreshToken, err := h.pds.IssueRefreshToken(holdDID, holdHandle) - if err != nil { - http.Error(w, fmt.Sprintf("failed to issue refresh token: %v", err), http.StatusInternalServerError) - return - } - - // Return session response - active := true - response := CreateSessionResponse{ - AccessJwt: accessToken, - RefreshJwt: refreshToken, - Handle: holdHandle, - DID: holdDID, - Active: &active, // Account is active - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(response) -} - -// HandleRefreshSession handles com.atproto.server.refreshSession -func (h *XRPCHandler) HandleRefreshSession(w http.ResponseWriter, r *http.Request) { - // Extract refresh token from Authorization header - authHeader := r.Header.Get("Authorization") - if authHeader == "" { - http.Error(w, "authorization header required", http.StatusUnauthorized) - return - } - - // Remove "Bearer " prefix - refreshToken := strings.TrimPrefix(authHeader, "Bearer ") - if refreshToken == authHeader { - http.Error(w, "invalid authorization header format", http.StatusUnauthorized) - return - } - - // Validate refresh token - claims, err := h.pds.ValidateRefreshToken(refreshToken) - if err != nil { - http.Error(w, fmt.Sprintf("invalid refresh token: %v", err), http.StatusUnauthorized) - return - } - - // Issue new access token (and optionally new refresh token) - accessToken, err := h.pds.IssueAccessToken(claims.DID, claims.Handle) - if err != nil { - http.Error(w, fmt.Sprintf("failed to issue access token: %v", err), http.StatusInternalServerError) - return - } - - // Issue new refresh token (rotate refresh tokens for security) - newRefreshToken, err := h.pds.IssueRefreshToken(claims.DID, claims.Handle) - if err != nil { - http.Error(w, fmt.Sprintf("failed to issue refresh token: %v", err), http.StatusInternalServerError) - return - } - - // Revoke old refresh token - if err := h.pds.RevokeRefreshToken(refreshToken); err != nil { - // Log but don't fail - new tokens are already issued - fmt.Printf("Warning: failed to revoke old refresh token: %v\n", err) - } - - // Return new tokens - active := true - response := CreateSessionResponse{ - AccessJwt: accessToken, - RefreshJwt: newRefreshToken, - Handle: claims.Handle, - DID: claims.DID, - Active: &active, // Account is active - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(response) -} - -// HandleGetSession handles com.atproto.server.getSession -func (h *XRPCHandler) HandleGetSession(w http.ResponseWriter, r *http.Request) { - // Extract access token from Authorization header - authHeader := r.Header.Get("Authorization") - if authHeader == "" { - http.Error(w, "authorization header required", http.StatusUnauthorized) - return - } - - // Remove "Bearer " prefix - accessToken := strings.TrimPrefix(authHeader, "Bearer ") - if accessToken == authHeader { - http.Error(w, "invalid authorization header format", http.StatusUnauthorized) - return - } - - // Validate access token - claims, err := h.pds.ValidateAccessToken(accessToken) - if err != nil { - http.Error(w, fmt.Sprintf("invalid access token: %v", err), http.StatusUnauthorized) - return - } - - // Return session info - response := SessionInfo{ - Handle: claims.Handle, - DID: claims.DID, - } - - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(response) -} - -// CreateRecordRequest represents a record creation request -type CreateRecordRequest struct { - Repo string `json:"repo"` // DID of the repository - Collection string `json:"collection"` // Collection name (e.g., "app.bsky.feed.post") - Rkey string `json:"rkey,omitempty"` // Optional record key (TID generated if not provided) - Validate *bool `json:"validate,omitempty"` // Optional validation flag - Record interface{} `json:"record"` // The record value (JSON object) -} - -// CreateRecordResponse represents a successful record creation -type CreateRecordResponse struct { - URI string `json:"uri"` // at://did/collection/rkey - CID string `json:"cid"` // Record CID -} - -// RawRecord wraps a record value and implements CBORMarshaler -// This allows us to accept any JSON record and marshal it to CBOR -type RawRecord struct { - Value map[string]interface{} -} - -// MarshalCBOR implements CBORMarshaler for RawRecord -func (r *RawRecord) MarshalCBOR(w io.Writer) error { - // Write CBOR map header - if err := cbg.WriteMajorTypeHeader(w, cbg.MajMap, uint64(len(r.Value))); err != nil { - return err - } - - // Write each key-value pair - for key, val := range r.Value { - // Write key as text string - if err := cbg.WriteMajorTypeHeader(w, cbg.MajTextString, uint64(len(key))); err != nil { - return err - } - if _, err := w.Write([]byte(key)); err != nil { - return err - } - - // Write value (simplified - handles common types) - if err := writeValue(w, val); err != nil { - return err - } - } - - return nil -} - -// writeValue writes a value to CBOR (helper for RawRecord) -func writeValue(w io.Writer, val interface{}) error { - switch v := val.(type) { - case string: - if err := cbg.WriteMajorTypeHeader(w, cbg.MajTextString, uint64(len(v))); err != nil { - return err - } - _, err := w.Write([]byte(v)) - return err - case int64: - return cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)) - case float64: - // Write as unsigned int for now (simplified) - return cbg.CborWriteHeader(w, cbg.MajUnsignedInt, uint64(v)) - case bool: - if v { - return cbg.WriteBool(w, true) - } - return cbg.WriteBool(w, false) - case map[string]interface{}: - rec := &RawRecord{Value: v} - return rec.MarshalCBOR(w) - case []interface{}: - if err := cbg.WriteMajorTypeHeader(w, cbg.MajArray, uint64(len(v))); err != nil { - return err - } - for _, item := range v { - if err := writeValue(w, item); err != nil { - return err - } - } - return nil - default: - // For unknown types, convert to JSON then write as string - jsonBytes, err := json.Marshal(v) - if err != nil { - return err - } - if err := cbg.WriteMajorTypeHeader(w, cbg.MajTextString, uint64(len(jsonBytes))); err != nil { - return err - } - _, err = w.Write(jsonBytes) - return err - } -} - -// HandleCreateRecord handles com.atproto.repo.createRecord -func (h *XRPCHandler) HandleCreateRecord(w http.ResponseWriter, r *http.Request) { - // Validate JWT authentication - user, err := ValidateJWTAuth(r, h.pds) - if err != nil { - http.Error(w, fmt.Sprintf("authentication required: %v", err), http.StatusUnauthorized) - return - } - - // Parse request - var req CreateRecordRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest) - return - } - - // Validate required fields - if req.Repo == "" || req.Collection == "" || req.Record == nil { - http.Error(w, "repo, collection, and record are required", http.StatusBadRequest) - return - } - - // Verify repo matches authenticated user - if req.Repo != user.DID { - http.Error(w, "repo must match authenticated user DID", http.StatusForbidden) - return - } - - // Verify repo matches this hold's DID - if req.Repo != h.pds.DID() { - http.Error(w, "invalid repo (must be this hold's DID)", http.StatusBadRequest) - return - } - - // Convert record from JSON to CBOR-marshalable format - recordMap, ok := req.Record.(map[string]interface{}) - if !ok { - http.Error(w, "record must be a JSON object", http.StatusBadRequest) - return - } - - // Wrap in RawRecord which implements CBORMarshaler - recordValue := &RawRecord{Value: recordMap} - - // Create record using repomgr - var recordPath string - var recordCID cid.Cid - - if req.Rkey != "" { - // Use PutRecord if rkey is specified - recordPath, recordCID, err = h.pds.repomgr.PutRecord( - r.Context(), - h.pds.uid, - req.Collection, - req.Rkey, - recordValue, - ) - } else { - // Use CreateRecord if no rkey (auto-generates TID) - recordPath, recordCID, err = h.pds.repomgr.CreateRecord( - r.Context(), - h.pds.uid, - req.Collection, - recordValue, - ) - } - - if err != nil { - http.Error(w, fmt.Sprintf("failed to create record: %v", err), http.StatusInternalServerError) - return - } - - // Extract rkey from path (format: "collection/rkey") - parts := strings.Split(recordPath, "/") - if len(parts) < 2 { - http.Error(w, "invalid record path returned", http.StatusInternalServerError) - return - } - actualRkey := parts[len(parts)-1] - - // Return success response - response := CreateRecordResponse{ - URI: fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), req.Collection, actualRkey), - CID: recordCID.String(), - } - - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusCreated) - json.NewEncoder(w).Encode(response) -} diff --git a/pkg/hold/pds/status.go b/pkg/hold/pds/status.go index 8ce48b6..c745a76 100644 --- a/pkg/hold/pds/status.go +++ b/pkg/hold/pds/status.go @@ -6,19 +6,16 @@ import ( "time" bsky "github.com/bluesky-social/indigo/api/bsky" - "github.com/ipfs/go-cid" ) const ( - // StatusPostRkey is the fixed rkey for the status post (singleton) - StatusPostRkey = "status" - // StatusPostCollection is the collection name for Bluesky posts StatusPostCollection = "app.bsky.feed.post" ) -// SetStatus creates or updates the hold's status post on Bluesky +// SetStatus creates a new status post on Bluesky // status should be "online" or "offline" +// Each call creates a unique post with a TID-based rkey func (p *HoldPDS) SetStatus(ctx context.Context, status string) error { // Format the post text with emoji indicator emoji := "🟢" @@ -27,70 +24,27 @@ func (p *HoldPDS) SetStatus(ctx context.Context, status string) error { } text := fmt.Sprintf("%s Current status: %s", emoji, status) - // Check if status post already exists - _, existingPost, err := p.GetStatusPost(ctx) - if err != nil { - // Post doesn't exist, create it - return p.createStatusPost(ctx, text) - } - - // Post exists, update it - // We need to preserve the original CreatedAt timestamp - return p.updateStatusPost(ctx, text, existingPost.CreatedAt) + // Create the post with a unique TID + return p.createStatusPost(ctx, text) } -// GetStatusPost retrieves the status post if it exists -func (p *HoldPDS) GetStatusPost(ctx context.Context) (cid.Cid, *bsky.FeedPost, error) { - // Use repomgr.GetRecord - recordCID, val, err := p.repomgr.GetRecord(ctx, p.uid, StatusPostCollection, StatusPostRkey, cid.Undef) - if err != nil { - return cid.Undef, nil, fmt.Errorf("failed to get status post: %w", err) - } - - // Type assert to bsky.FeedPost - post, ok := val.(*bsky.FeedPost) - if !ok { - return cid.Undef, nil, fmt.Errorf("unexpected type for status post: %T", val) - } - - return recordCID, post, nil -} - -// createStatusPost creates a new status post (first time) +// createStatusPost creates a new status post with a TID-based rkey func (p *HoldPDS) createStatusPost(ctx context.Context, text string) error { // Create post struct - now := time.Now().Format(time.RFC3339) + now := time.Now() post := &bsky.FeedPost{ LexiconTypeID: "app.bsky.feed.post", Text: text, - CreatedAt: now, + CreatedAt: now.Format(time.RFC3339), } - // Use repomgr.PutRecord - creates with explicit rkey, fails if already exists - recordPath, recordCID, err := p.repomgr.PutRecord(ctx, p.uid, StatusPostCollection, StatusPostRkey, post) + // Use repomgr.CreateRecord to create the post with auto-generated TID + // CreateRecord automatically generates a unique TID using the repo's clock + rkey, recordCID, err := p.repomgr.CreateRecord(ctx, p.uid, StatusPostCollection, post) if err != nil { return fmt.Errorf("failed to create status post: %w", err) } - fmt.Printf("Created status post at %s, cid: %s, text: %s\n", recordPath, recordCID, text) - return nil -} - -// updateStatusPost updates an existing status post -func (p *HoldPDS) updateStatusPost(ctx context.Context, text string, createdAt string) error { - // Create updated post struct with original CreatedAt - post := &bsky.FeedPost{ - LexiconTypeID: "app.bsky.feed.post", - Text: text, - CreatedAt: createdAt, // Preserve original creation time - } - - // Use repomgr.UpdateRecord - recordCID, err := p.repomgr.UpdateRecord(ctx, p.uid, StatusPostCollection, StatusPostRkey, post) - if err != nil { - return fmt.Errorf("failed to update status post: %w", err) - } - - fmt.Printf("Updated status post, cid: %s, text: %s\n", recordCID, text) + fmt.Printf("Created status post at %s/%s (rkey: %s), cid: %s, text: %s\n", StatusPostCollection, rkey, rkey, recordCID, text) return nil } diff --git a/pkg/hold/pds/status_test.go b/pkg/hold/pds/status_test.go index a954c17..22e754e 100644 --- a/pkg/hold/pds/status_test.go +++ b/pkg/hold/pds/status_test.go @@ -2,19 +2,41 @@ package pds import ( "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" "os" "path/filepath" "testing" + "time" + "atcr.io/pkg/atproto" + "atcr.io/pkg/s3" bsky "github.com/bluesky-social/indigo/api/bsky" ) +// Shared test resources (used across all test files in package) +var ( + sharedTestKeyPath string + sharedTestKey []byte + sharedPDS *HoldPDS // Shared bootstrapped PDS for read-only tests + sharedHandler *XRPCHandler // Shared handler for read-only tests + sharedCtx context.Context // Shared context +) + func TestStatusPost(t *testing.T) { - // Create temporary directory for test database + // Create temporary directory for test tmpDir := t.TempDir() - dbPath := filepath.Join(tmpDir, "test.db") + // Use in-memory database for speed + dbPath := ":memory:" keyPath := filepath.Join(tmpDir, "test.key") + // Copy shared signing key + if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil { + t.Fatalf("Failed to copy shared signing key: %v", err) + } + // Create test PDS ctx := context.Background() did := "did:web:test.example.com" @@ -31,6 +53,45 @@ func TestStatusPost(t *testing.T) { t.Fatalf("Failed to initialize repo: %v", err) } + // Create handler for XRPC endpoints + handler := NewXRPCHandler(holdPDS, s3.S3Service{}, nil, nil, &mockPDSClient{}) + + // Helper function to list posts via XRPC + listPosts := func() ([]map[string]any, error) { + req := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ + "repo": did, + "collection": StatusPostCollection, + "limit": "100", + "reverse": "true", // Most recent first + }) + w := httptest.NewRecorder() + handler.HandleListRecords(w, req) + + if w.Code != http.StatusOK { + return nil, fmt.Errorf("unexpected status code: %d, body: %s", w.Code, w.Body.String()) + } + + var result map[string]any + if err := json.NewDecoder(w.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + records, ok := result["records"].([]any) + if !ok { + return nil, fmt.Errorf("expected records array, got %T", result["records"]) + } + + posts := make([]map[string]any, len(records)) + for i, rec := range records { + post, ok := rec.(map[string]any) + if !ok { + return nil, fmt.Errorf("expected record map, got %T", rec) + } + posts[i] = post + } + return posts, nil + } + t.Run("CreateStatusPost", func(t *testing.T) { // Set status to online (creates new post) err := holdPDS.SetStatus(ctx, "online") @@ -38,110 +99,151 @@ func TestStatusPost(t *testing.T) { t.Fatalf("Failed to set status to online: %v", err) } - // Verify post was created - _, post, err := holdPDS.GetStatusPost(ctx) + // List posts + posts, err := listPosts() if err != nil { - t.Fatalf("Failed to get status post: %v", err) + t.Fatalf("Failed to list posts: %v", err) } - if post.Text != "🟢 Current status: online" { - t.Errorf("Expected text '🟢 Current status: online', got '%s'", post.Text) + if len(posts) == 0 { + t.Fatal("Expected at least one status post, got 0") } - if post.LexiconTypeID != "app.bsky.feed.post" { - t.Errorf("Expected LexiconTypeID 'app.bsky.feed.post', got '%s'", post.LexiconTypeID) + // Get the latest post + post := posts[0] + + value, ok := post["value"].(map[string]any) + if !ok { + t.Fatalf("Expected value map, got %T", post["value"]) } - if post.CreatedAt == "" { - t.Error("CreatedAt should not be empty") + text, ok := value["text"].(string) + if !ok { + t.Fatalf("Expected text string, got %T", value["text"]) + } + + if text != "🟢 Current status: online" { + t.Errorf("Expected text '🟢 Current status: online', got '%s'", text) + } + + // Verify TID-based rkey (extract from URI) + uri, ok := post["uri"].(string) + if !ok { + t.Fatalf("Expected uri string, got %T", post["uri"]) + } + // URI format: at://did:web:test.example.com/app.bsky.feed.post/3m3c4... + // We just check that it contains the collection + if !contains(uri, StatusPostCollection) { + t.Errorf("Expected URI to contain collection %s, got %s", StatusPostCollection, uri) } }) - t.Run("UpdateStatusPost", func(t *testing.T) { - // Get the original post to check CreatedAt preservation - _, originalPost, err := holdPDS.GetStatusPost(ctx) - if err != nil { - t.Fatalf("Failed to get original status post: %v", err) - } - - // Set status to offline (updates existing post) - err = holdPDS.SetStatus(ctx, "offline") + t.Run("CreateMultiplePosts", func(t *testing.T) { + // Create multiple status posts + err := holdPDS.SetStatus(ctx, "offline") if err != nil { t.Fatalf("Failed to set status to offline: %v", err) } - // Verify post was updated - _, post, err := holdPDS.GetStatusPost(ctx) + // Wait a moment to ensure different timestamp + time.Sleep(10 * time.Millisecond) + + err = holdPDS.SetStatus(ctx, "online") if err != nil { - t.Fatalf("Failed to get updated status post: %v", err) + t.Fatalf("Failed to set status to online again: %v", err) } - if post.Text != "🔴 Current status: offline" { - t.Errorf("Expected text '🔴 Current status: offline', got '%s'", post.Text) + // List all posts - should have at least 3 now (1 from previous test + 2 from this test) + posts, err := listPosts() + if err != nil { + t.Fatalf("Failed to list posts: %v", err) } - // Verify CreatedAt was preserved - if post.CreatedAt != originalPost.CreatedAt { - t.Errorf("CreatedAt should be preserved. Expected '%s', got '%s'", originalPost.CreatedAt, post.CreatedAt) + if len(posts) < 3 { + t.Errorf("Expected at least 3 status posts, got %d", len(posts)) + } + + // Verify each post has a unique URI + uris := make(map[string]bool) + for _, post := range posts { + uri, ok := post["uri"].(string) + if !ok { + t.Errorf("Expected uri string, got %T", post["uri"]) + continue + } + if uris[uri] { + t.Errorf("Duplicate URI found: %s", uri) + } + uris[uri] = true + } + + // Verify the latest post is online + latestPost := posts[0] + value, ok := latestPost["value"].(map[string]any) + if !ok { + t.Fatalf("Expected value map, got %T", latestPost["value"]) + } + text, ok := value["text"].(string) + if !ok { + t.Fatalf("Expected text string, got %T", value["text"]) + } + if text != "🟢 Current status: online" { + t.Errorf("Expected latest post text '🟢 Current status: online', got '%s'", text) } }) - t.Run("ToggleStatus", func(t *testing.T) { - // Toggle back to online - err := holdPDS.SetStatus(ctx, "online") + t.Run("OfflineStatus", func(t *testing.T) { + // Create offline status post + err := holdPDS.SetStatus(ctx, "offline") if err != nil { - t.Fatalf("Failed to set status to online: %v", err) + t.Fatalf("Failed to set status to offline: %v", err) } - _, post, err := holdPDS.GetStatusPost(ctx) + // Get the latest post + posts, err := listPosts() if err != nil { - t.Fatalf("Failed to get status post: %v", err) + t.Fatalf("Failed to list posts: %v", err) } - if post.Text != "🟢 Current status: online" { - t.Errorf("Expected text '🟢 Current status: online', got '%s'", post.Text) + if len(posts) == 0 { + t.Fatal("Expected at least one status post, got 0") + } + + latestPost := posts[0] + value, ok := latestPost["value"].(map[string]any) + if !ok { + t.Fatalf("Expected value map, got %T", latestPost["value"]) + } + text, ok := value["text"].(string) + if !ok { + t.Fatalf("Expected text string, got %T", value["text"]) + } + + if text != "🔴 Current status: offline" { + t.Errorf("Expected text '🔴 Current status: offline', got '%s'", text) } }) } func TestStatusPostCollection(t *testing.T) { - // Verify constants + // Verify constant if StatusPostCollection != "app.bsky.feed.post" { t.Errorf("Expected StatusPostCollection 'app.bsky.feed.post', got '%s'", StatusPostCollection) } - - if StatusPostRkey != "status" { - t.Errorf("Expected StatusPostRkey 'status', got '%s'", StatusPostRkey) - } } -func TestGetStatusPostNotExists(t *testing.T) { - // Create temporary directory for test database - tmpDir := t.TempDir() - dbPath := filepath.Join(tmpDir, "test.db") - keyPath := filepath.Join(tmpDir, "test.key") +// Helper function to check if a string contains a substring +func contains(s, substr string) bool { + return len(s) >= len(substr) && (s == substr || len(s) > len(substr) && findSubstring(s, substr)) +} - // Create test PDS - ctx := context.Background() - did := "did:web:test2.example.com" - publicURL := "https://test2.example.com" - - holdPDS, err := NewHoldPDS(ctx, did, publicURL, dbPath, keyPath) - if err != nil { - t.Fatalf("Failed to create test PDS: %v", err) - } - - // Initialize empty repo - err = holdPDS.repomgr.InitNewActor(ctx, holdPDS.uid, "", did, "", "", "") - if err != nil { - t.Fatalf("Failed to initialize repo: %v", err) - } - - // Try to get status post that doesn't exist - _, _, err = holdPDS.GetStatusPost(ctx) - if err == nil { - t.Error("Expected error when getting non-existent status post, got nil") +func findSubstring(s, substr string) bool { + for i := 0; i <= len(s)-len(substr); i++ { + if s[i:i+len(substr)] == substr { + return true + } } + return false } func init() { @@ -154,6 +256,41 @@ func init() { // Cleanup function to remove test files func TestMain(m *testing.M) { + // Create a temporary directory for shared test key + tmpDir, err := os.MkdirTemp("", "pds-test-shared-*") + if err != nil { + panic(fmt.Sprintf("Failed to create temp dir: %v", err)) + } + defer os.RemoveAll(tmpDir) + + // Generate one signing key to be reused across all tests in the package + sharedTestKeyPath = filepath.Join(tmpDir, "shared-signing-key") + privateKey, err := GenerateOrLoadKey(sharedTestKeyPath) + if err != nil { + panic(fmt.Sprintf("Failed to generate shared signing key: %v", err)) + } + + // Store the key bytes so tests can copy them + sharedTestKey = privateKey.Bytes() + + // Create one shared, bootstrapped PDS for read-only tests + // Use in-memory database for speed + sharedCtx = context.Background() + sharedPDS, err = NewHoldPDS(sharedCtx, "did:web:hold.example.com", "https://hold.example.com", ":memory:", sharedTestKeyPath) + if err != nil { + panic(fmt.Sprintf("Failed to create shared PDS: %v", err)) + } + + // Bootstrap once + ownerDID := "did:plc:testowner123" + err = sharedPDS.Bootstrap(sharedCtx, nil, ownerDID, true, false, "") + if err != nil { + panic(fmt.Sprintf("Failed to bootstrap shared PDS: %v", err)) + } + + // Create shared handler + sharedHandler = NewXRPCHandler(sharedPDS, s3.S3Service{}, nil, nil, &mockPDSClient{}) + // Run tests code := m.Run() diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index a935958..5ed55a8 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -92,7 +92,7 @@ func (h *XRPCHandler) CORSMiddleware() func(http.Handler) http.Handler { // Handle OPTIONS preflight if r.Method == "OPTIONS" { - w.WriteHeader(http.StatusNoContent) + w.WriteHeader(http.StatusOK) return } @@ -150,11 +150,6 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { r.Get("/xrpc/_health", h.HandleHealth) r.Get(atproto.ServerDescribeServer, h.HandleDescribeServer) - // Session management (public - creates sessions) - r.Post(atproto.ServerCreateSession, h.HandleCreateSession) - r.Post(atproto.ServerRefreshSession, h.HandleRefreshSession) - r.Get(atproto.ServerGetSession, h.HandleGetSession) - // Repository metadata r.Get(atproto.RepoDescribeRepo, h.HandleDescribeRepo) r.Get(atproto.RepoGetRecord, h.HandleGetRecord) @@ -186,7 +181,6 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { // Write endpoints (owner/crew admin auth) r.Group(func(r chi.Router) { r.Use(h.requireOwnerOrCrewAdmin) - r.Post(atproto.RepoDeleteRecord, h.HandleDeleteRecord) r.Post(atproto.RepoUploadBlob, h.HandleUploadBlob) }) @@ -194,15 +188,8 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { // Auth-only endpoints (DPoP auth) r.Group(func(r chi.Router) { r.Use(h.requireAuth) - r.Post(atproto.HoldRequestCrew, h.HandleRequestCrew) }) - - // JWT-authenticated endpoints (JWT auth from createSession) - // Note: JWT auth is validated inside each handler - r.Group(func(r chi.Router) { - r.Post("/xrpc/com.atproto.repo.createRecord", h.HandleCreateRecord) - }) } // HandleHealth returns health check information @@ -861,7 +848,9 @@ func (h *XRPCHandler) HandleSubscribeRepos(w http.ResponseWriter, r *http.Reques } // Get optional cursor parameter for backfill - var cursor int64 = 0 + // Default to -1 (no backfill, only stream new events) + // cursor=0 means "replay all events from the beginning" + var cursor int64 = -1 if cursorStr := r.URL.Query().Get("cursor"); cursorStr != "" { var err error cursor, err = strconv.ParseInt(cursorStr, 10, 64) @@ -879,21 +868,9 @@ func (h *XRPCHandler) HandleSubscribeRepos(w http.ResponseWriter, r *http.Reques } // Subscribe to events - sub := h.broadcaster.Subscribe(conn, cursor) - // The broadcaster's handleSubscriber goroutine will manage this connection - // We just need to keep reading to detect client disconnects - go func() { - defer h.broadcaster.Unsubscribe(sub) - for { - // Read messages from client (mostly just to detect disconnect) - _, _, err := conn.ReadMessage() - if err != nil { - // Client disconnected - break - } - } - }() + // and handle cleanup when the client disconnects + h.broadcaster.Subscribe(conn, cursor) } // HandleUploadBlob handles blob uploads with support for multipart operations diff --git a/pkg/hold/pds/xrpc_test.go b/pkg/hold/pds/xrpc_test.go index c6ab6c0..84e423d 100644 --- a/pkg/hold/pds/xrpc_test.go +++ b/pkg/hold/pds/xrpc_test.go @@ -12,12 +12,17 @@ import ( "path/filepath" "strings" "testing" + "time" "atcr.io/pkg/atproto" "atcr.io/pkg/s3" + indigoAtproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/events" "github.com/distribution/distribution/v3/registry/storage/driver/factory" _ "github.com/distribution/distribution/v3/registry/storage/driver/filesystem" "github.com/go-chi/chi/v5" + "github.com/gorilla/websocket" + "github.com/ipfs/go-cid" ) // Test helpers @@ -30,9 +35,15 @@ func setupTestXRPCHandler(t *testing.T) (*XRPCHandler, context.Context) { ctx := context.Background() tmpDir := t.TempDir() - dbPath := filepath.Join(tmpDir, "pds.db") + // Use in-memory database for speed + dbPath := ":memory:" keyPath := filepath.Join(tmpDir, "signing-key") + // Copy shared signing key instead of generating a new one + if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil { + t.Fatalf("Failed to copy shared signing key: %v", err) + } + pds, err := NewHoldPDS(ctx, "did:web:hold.example.com", "https://hold.example.com", dbPath, keyPath) if err != nil { t.Fatalf("Failed to create test PDS: %v", err) @@ -117,6 +128,32 @@ func assertJSONResponse(t *testing.T, w *httptest.ResponseRecorder, expectedCode return result } +// decodeFirehoseMessage decodes an ATProto firehose message (header + CBOR body) +func decodeFirehoseMessage(t *testing.T, message []byte) (*events.EventHeader, *indigoAtproto.SyncSubscribeRepos_Commit) { + t.Helper() + + reader := bytes.NewReader(message) + + // Decode header + var header events.EventHeader + if err := header.UnmarshalCBOR(reader); err != nil { + t.Fatalf("Failed to decode event header: %v", err) + } + + // Verify it's a commit event + if header.MsgType != "#commit" { + t.Fatalf("Expected #commit event, got %s", header.MsgType) + } + + // Decode commit event + var commit indigoAtproto.SyncSubscribeRepos_Commit + if err := commit.UnmarshalCBOR(reader); err != nil { + t.Fatalf("Failed to decode commit event: %v", err) + } + + return &header, &commit +} + // assertCARResponse validates CAR file response func assertCARResponse(t *testing.T, w *httptest.ResponseRecorder, expectedCode int) []byte { t.Helper() @@ -1331,9 +1368,15 @@ func setupTestXRPCHandlerWithBlobs(t *testing.T) (*XRPCHandler, *mockS3Service, ctx := context.Background() tmpDir := t.TempDir() - dbPath := filepath.Join(tmpDir, "pds.db") + // Use in-memory database for speed + dbPath := ":memory:" keyPath := filepath.Join(tmpDir, "signing-key") + // Copy shared signing key instead of generating a new one + if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil { + t.Fatalf("Failed to copy shared signing key: %v", err) + } + pds, err := NewHoldPDS(ctx, "did:web:hold.example.com", "https://hold.example.com", dbPath, keyPath) if err != nil { t.Fatalf("Failed to create test PDS: %v", err) @@ -1686,7 +1729,7 @@ func TestHandleGetBlob_CORSHeaders(t *testing.T) { w := httptest.NewRecorder() // Wrap with CORS middleware (chi-style) - corsHandler := handler.corsMiddleware(http.HandlerFunc(handler.HandleGetBlob)) + corsHandler := handler.CORSMiddleware()(http.HandlerFunc(handler.HandleGetBlob)) corsHandler.ServeHTTP(w, req) // Verify CORS headers are present @@ -1719,6 +1762,7 @@ func TestCORSMiddleware(t *testing.T) { // Create chi router and register handlers r := chi.NewRouter() + r.Use(handler.CORSMiddleware()) // Apply CORS middleware handler.RegisterHandlers(r) tests := []struct { @@ -2009,3 +2053,213 @@ func TestRouteMethodEnforcement_POST(t *testing.T) { }) } } + +// TestHandleSubscribeRepos tests the WebSocket firehose endpoint +func TestHandleSubscribeRepos(t *testing.T) { + handler, ctx := setupTestXRPCHandler(t) + + // Create EventBroadcaster + broadcaster := NewEventBroadcaster(handler.pds.DID(), 100) + handler.broadcaster = broadcaster + + // Set up test HTTP server + r := chi.NewRouter() + handler.RegisterHandlers(r) + server := httptest.NewServer(r) + defer server.Close() + + // Broadcast some events before connecting + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + for i := 1; i <= 3; i++ { + event := &RepoEvent{ + NewRoot: testCID, + Rev: fmt.Sprintf("rev-%d", i), + RepoSlice: []byte(fmt.Sprintf("CAR data %d", i)), + Ops: []RepoOp{}, + } + broadcaster.Broadcast(ctx, event) + } + + // Verify events were stored + if broadcaster.eventSeq != 3 { + t.Fatalf("Expected eventSeq=3, got %d", broadcaster.eventSeq) + } + + t.Run("cursor=0 replays all events", func(t *testing.T) { + // Connect to WebSocket with cursor=0 + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/xrpc/com.atproto.sync.subscribeRepos?cursor=0" + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("Failed to connect to WebSocket: %v", err) + } + defer conn.Close() + + // Should receive the 3 historical events + for i := 0; i < 3; i++ { + messageType, message, err := conn.ReadMessage() + if err != nil { + t.Fatalf("Failed to read message: %v", err) + } + + if messageType != websocket.BinaryMessage { + t.Errorf("Expected binary message, got type %d", messageType) + } + + // Decode CBOR message (header + commit) + header, commit := decodeFirehoseMessage(t, message) + + // Verify header + if header.MsgType != "#commit" { + t.Errorf("Expected MsgType=#commit, got %s", header.MsgType) + } + + // Verify commit fields + expectedSeq := int64(i + 1) + if commit.Seq != expectedSeq { + t.Errorf("Expected seq=%d, got %d", expectedSeq, commit.Seq) + } + if commit.Repo != handler.pds.DID() { + t.Errorf("Expected repo=%s, got %s", handler.pds.DID(), commit.Repo) + } + } + }) + + t.Run("cursor=2 only replays events after 2", func(t *testing.T) { + // Connect with cursor=2 + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/xrpc/com.atproto.sync.subscribeRepos?cursor=2" + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("Failed to connect to WebSocket: %v", err) + } + defer conn.Close() + + // Should only receive event 3 (after cursor=2) + messageType, message, err := conn.ReadMessage() + if err != nil { + t.Fatalf("Failed to read message: %v", err) + } + + if messageType != websocket.BinaryMessage { + t.Errorf("Expected binary message, got type %d", messageType) + } + + header, commit := decodeFirehoseMessage(t, message) + if header.MsgType != "#commit" { + t.Errorf("Expected MsgType=#commit, got %s", header.MsgType) + } + + if commit.Seq != 3 { + t.Errorf("Expected seq=3, got %d", commit.Seq) + } + + // Verify no more events (use timeout) + conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond)) + _, _, err = conn.ReadMessage() + if err == nil { + t.Error("Expected no more events, but received another message") + } + }) + + t.Run("no cursor streams only new events", func(t *testing.T) { + // Connect without cursor (should not get backfill) + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/xrpc/com.atproto.sync.subscribeRepos" + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("Failed to connect to WebSocket: %v", err) + } + defer conn.Close() + + // Verify no historical events by broadcasting immediately and checking + // that we only receive the new event (not historical ones) + // Give subscriber time to register first + time.Sleep(100 * time.Millisecond) + + // Broadcast a new event (seq 4) + newEvent := &RepoEvent{ + NewRoot: testCID, + Rev: "rev-4", + RepoSlice: []byte("CAR data 4"), + Ops: []RepoOp{}, + } + broadcaster.Broadcast(ctx, newEvent) + + // Should receive ONLY the new event (seq 4), not historical events 1-3 + conn.SetReadDeadline(time.Now().Add(1 * time.Second)) + messageType, message, err := conn.ReadMessage() + if err != nil { + t.Fatalf("Failed to read new event: %v", err) + } + + if messageType != websocket.BinaryMessage { + t.Errorf("Expected binary message, got type %d", messageType) + } + + header, commit := decodeFirehoseMessage(t, message) + if header.MsgType != "#commit" { + t.Errorf("Expected MsgType=#commit, got %s", header.MsgType) + } + + // Key assertion: should be seq 4 (new event), not seq 1 (historical backfill) + if commit.Seq != 4 { + t.Errorf("Expected seq=4 for new event (no backfill), got %d", commit.Seq) + } + + // Verify no more messages (no historical backfill) + conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond)) + _, _, err = conn.ReadMessage() + if err == nil { + t.Error("Expected no more events, but received another message (possible backfill leak)") + } + }) + + t.Run("real-time event delivery", func(t *testing.T) { + // Connect with cursor=0 to get all events first + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/xrpc/com.atproto.sync.subscribeRepos?cursor=0" + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("Failed to connect to WebSocket: %v", err) + } + defer conn.Close() + + // Read and discard the 4 historical events (seq 1-4) + for i := 0; i < 4; i++ { + _, _, err := conn.ReadMessage() + if err != nil { + t.Fatalf("Failed to read historical event %d: %v", i+1, err) + } + } + + // Broadcast 2 new events + for i := 5; i <= 6; i++ { + newEvent := &RepoEvent{ + NewRoot: testCID, + Rev: fmt.Sprintf("rev-%d", i), + RepoSlice: []byte(fmt.Sprintf("CAR data %d", i)), + Ops: []RepoOp{}, + } + broadcaster.Broadcast(ctx, newEvent) + } + + // Should receive both new events + for expectedSeq := 5; expectedSeq <= 6; expectedSeq++ { + conn.SetReadDeadline(time.Now().Add(1 * time.Second)) + messageType, message, err := conn.ReadMessage() + if err != nil { + t.Fatalf("Failed to read event seq=%d: %v", expectedSeq, err) + } + + if messageType != websocket.BinaryMessage { + t.Errorf("Expected binary message, got type %d", messageType) + } + + header, commit := decodeFirehoseMessage(t, message) + if header.MsgType != "#commit" { + t.Errorf("Expected MsgType=#commit, got %s", header.MsgType) + } + + if commit.Seq != int64(expectedSeq) { + t.Errorf("Expected seq=%d, got %d", expectedSeq, commit.Seq) + } + } + }) +}