A community based topic aggregation platform built on atproto
bretton.dev
1package integration
2
3import (
4 "Coves/internal/api/middleware"
5 "Coves/internal/api/routes"
6 "Coves/internal/atproto/identity"
7 "Coves/internal/atproto/jetstream"
8 "Coves/internal/atproto/utils"
9 "Coves/internal/core/communities"
10 "Coves/internal/core/users"
11 "Coves/internal/db/postgres"
12 "bytes"
13 "context"
14 "database/sql"
15 "encoding/json"
16 "fmt"
17 "io"
18 "net"
19 "net/http"
20 "net/http/httptest"
21 "os"
22 "strings"
23 "testing"
24 "time"
25
26 "github.com/go-chi/chi/v5"
27 "github.com/gorilla/websocket"
28 _ "github.com/lib/pq"
29 "github.com/pressly/goose/v3"
30)
31
32// TestCommunity_E2E is a TRUE end-to-end test covering the complete flow:
33// 1. HTTP Endpoint → Service Layer → PDS Account Creation → PDS Record Write
34// 2. PDS → REAL Jetstream Firehose → Consumer → AppView DB (TRUE E2E!)
35// 3. AppView DB → XRPC HTTP Endpoints → Client
36//
37// This test verifies:
38// - V2: Community owns its own PDS account and repository
39// - V2: Record URI points to community's repo (at://community_did/...)
40// - Real Jetstream firehose subscription and event consumption
41// - Complete data flow from HTTP write to HTTP read via real infrastructure
42func TestCommunity_E2E(t *testing.T) {
43 // Skip in short mode since this requires real PDS
44 if testing.Short() {
45 t.Skip("Skipping E2E test in short mode")
46 }
47
48 // Setup test database
49 dbURL := os.Getenv("TEST_DATABASE_URL")
50 if dbURL == "" {
51 dbURL = "postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable"
52 }
53
54 db, err := sql.Open("postgres", dbURL)
55 if err != nil {
56 t.Fatalf("Failed to connect to test database: %v", err)
57 }
58 defer func() {
59 if closeErr := db.Close(); closeErr != nil {
60 t.Logf("Failed to close database: %v", closeErr)
61 }
62 }()
63
64 // Run migrations
65 if dialectErr := goose.SetDialect("postgres"); dialectErr != nil {
66 t.Fatalf("Failed to set goose dialect: %v", dialectErr)
67 }
68 if migrateErr := goose.Up(db, "../../internal/db/migrations"); migrateErr != nil {
69 t.Fatalf("Failed to run migrations: %v", migrateErr)
70 }
71
72 // Check if PDS is running
73 pdsURL := os.Getenv("PDS_URL")
74 if pdsURL == "" {
75 pdsURL = "http://localhost:3001"
76 }
77
78 healthResp, err := http.Get(pdsURL + "/xrpc/_health")
79 if err != nil {
80 t.Skipf("PDS not running at %s: %v", pdsURL, err)
81 }
82 func() {
83 if closeErr := healthResp.Body.Close(); closeErr != nil {
84 t.Logf("Failed to close health response: %v", closeErr)
85 }
86 }()
87
88 // Setup dependencies
89 communityRepo := postgres.NewCommunityRepository(db)
90
91 // Get instance credentials
92 instanceHandle := os.Getenv("PDS_INSTANCE_HANDLE")
93 instancePassword := os.Getenv("PDS_INSTANCE_PASSWORD")
94 if instanceHandle == "" {
95 instanceHandle = "testuser123.local.coves.dev"
96 }
97 if instancePassword == "" {
98 instancePassword = "test-password-123"
99 }
100
101 t.Logf("🔐 Authenticating with PDS as: %s", instanceHandle)
102
103 // Authenticate to get instance DID
104 accessToken, instanceDID, err := authenticateWithPDS(pdsURL, instanceHandle, instancePassword)
105 if err != nil {
106 t.Fatalf("Failed to authenticate with PDS: %v", err)
107 }
108
109 t.Logf("✅ Authenticated - Instance DID: %s", instanceDID)
110
111 // Initialize auth middleware (skipVerify=true for E2E tests)
112 authMiddleware := middleware.NewAtProtoAuthMiddleware(nil, true)
113
114 // V2.0: Extract instance domain for community provisioning
115 var instanceDomain string
116 if strings.HasPrefix(instanceDID, "did:web:") {
117 instanceDomain = strings.TrimPrefix(instanceDID, "did:web:")
118 } else {
119 // Use .social for testing (not .local - that TLD is disallowed by atProto)
120 instanceDomain = "coves.social"
121 }
122
123 // V2.0: Create user service with REAL identity resolution using local PLC
124 plcURL := os.Getenv("PLC_DIRECTORY_URL")
125 if plcURL == "" {
126 plcURL = "http://localhost:3002" // Local PLC directory
127 }
128 userRepo := postgres.NewUserRepository(db)
129 identityConfig := identity.DefaultConfig()
130 identityConfig.PLCURL = plcURL // Use local PLC for identity resolution
131 identityResolver := identity.NewResolver(db, identityConfig)
132 _ = users.NewUserService(userRepo, identityResolver, pdsURL) // Keep for potential future use
133 t.Logf("✅ Identity resolver configured with local PLC: %s", plcURL)
134
135 // V2.0: Initialize PDS account provisioner (simplified - no DID generator needed!)
136 // PDS handles all DID generation and registration automatically
137 provisioner := communities.NewPDSAccountProvisioner(instanceDomain, pdsURL)
138
139 // Create service (no longer needs didGen directly - provisioner owns it)
140 communityService := communities.NewCommunityService(communityRepo, pdsURL, instanceDID, instanceDomain, provisioner)
141 if svc, ok := communityService.(interface{ SetPDSAccessToken(string) }); ok {
142 svc.SetPDSAccessToken(accessToken)
143 }
144
145 // Use real identity resolver with local PLC for production-like testing
146 consumer := jetstream.NewCommunityEventConsumer(communityRepo, "did:web:coves.local", true, identityResolver)
147
148 // Setup HTTP server with XRPC routes
149 r := chi.NewRouter()
150 routes.RegisterCommunityRoutes(r, communityService, authMiddleware)
151 httpServer := httptest.NewServer(r)
152 defer httpServer.Close()
153
154 ctx := context.Background()
155
156 // ====================================================================================
157 // Part 1: Write-Forward to PDS (Service Layer)
158 // ====================================================================================
159 t.Run("1. Write-Forward to PDS", func(t *testing.T) {
160 // Use shorter names to avoid "Handle too long" errors
161 // atProto handles max: 63 chars, format: name.community.coves.social
162 communityName := fmt.Sprintf("e2e-%d", time.Now().Unix())
163
164 createReq := communities.CreateCommunityRequest{
165 Name: communityName,
166 DisplayName: "E2E Test Community",
167 Description: "Testing full E2E flow",
168 Visibility: "public",
169 CreatedByDID: instanceDID,
170 HostedByDID: instanceDID,
171 AllowExternalDiscovery: true,
172 }
173
174 t.Logf("\n📝 Creating community via service: %s", communityName)
175 community, err := communityService.CreateCommunity(ctx, createReq)
176 if err != nil {
177 t.Fatalf("Failed to create community: %v", err)
178 }
179
180 t.Logf("✅ Service returned:")
181 t.Logf(" DID: %s", community.DID)
182 t.Logf(" Handle: %s", community.Handle)
183 t.Logf(" RecordURI: %s", community.RecordURI)
184 t.Logf(" RecordCID: %s", community.RecordCID)
185
186 // Verify DID format
187 if community.DID[:8] != "did:plc:" {
188 t.Errorf("Expected did:plc DID, got: %s", community.DID)
189 }
190
191 // V2: Verify PDS account was created for the community
192 t.Logf("\n🔍 V2: Verifying community PDS account exists...")
193 expectedHandle := fmt.Sprintf("%s.community.%s", communityName, instanceDomain)
194 t.Logf(" Expected handle: %s", expectedHandle)
195 t.Logf(" (Using subdomain: *.community.%s)", instanceDomain)
196
197 accountDID, accountHandle, err := queryPDSAccount(pdsURL, expectedHandle)
198 if err != nil {
199 t.Fatalf("❌ V2: Community PDS account not found: %v", err)
200 }
201
202 t.Logf("✅ V2: Community PDS account exists!")
203 t.Logf(" Account DID: %s", accountDID)
204 t.Logf(" Account Handle: %s", accountHandle)
205
206 // Verify the account DID matches the community DID
207 if accountDID != community.DID {
208 t.Errorf("❌ V2: Account DID mismatch! Community DID: %s, PDS Account DID: %s",
209 community.DID, accountDID)
210 } else {
211 t.Logf("✅ V2: Community DID matches PDS account DID (self-owned repository)")
212 }
213
214 // V2: Verify record exists in PDS (in community's own repository)
215 t.Logf("\n📡 V2: Querying PDS for record in community's repository...")
216
217 collection := "social.coves.community.profile"
218 rkey := utils.ExtractRKeyFromURI(community.RecordURI)
219
220 // V2: Query community's repository (not instance repository!)
221 getRecordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
222 pdsURL, community.DID, collection, rkey)
223
224 t.Logf(" Querying: at://%s/%s/%s", community.DID, collection, rkey)
225
226 pdsResp, err := http.Get(getRecordURL)
227 if err != nil {
228 t.Fatalf("Failed to query PDS: %v", err)
229 }
230 defer func() { _ = pdsResp.Body.Close() }()
231
232 if pdsResp.StatusCode != http.StatusOK {
233 body, readErr := io.ReadAll(pdsResp.Body)
234 if readErr != nil {
235 t.Fatalf("PDS returned status %d (failed to read body: %v)", pdsResp.StatusCode, readErr)
236 }
237 t.Fatalf("PDS returned status %d: %s", pdsResp.StatusCode, string(body))
238 }
239
240 var pdsRecord struct {
241 Value map[string]interface{} `json:"value"`
242 URI string `json:"uri"`
243 CID string `json:"cid"`
244 }
245
246 if err := json.NewDecoder(pdsResp.Body).Decode(&pdsRecord); err != nil {
247 t.Fatalf("Failed to decode PDS response: %v", err)
248 }
249
250 t.Logf("✅ Record found in PDS!")
251 t.Logf(" URI: %s", pdsRecord.URI)
252 t.Logf(" CID: %s", pdsRecord.CID)
253
254 // Print full record for inspection
255 recordJSON, marshalErr := json.MarshalIndent(pdsRecord.Value, " ", " ")
256 if marshalErr != nil {
257 t.Logf(" Failed to marshal record: %v", marshalErr)
258 } else {
259 t.Logf(" Record value:\n %s", string(recordJSON))
260 }
261
262 // V2: DID and Handle are NOT in the record - they're resolved from the repository URI
263 // The record should have name, hostedBy, createdBy, etc. but no 'did' or 'handle' fields
264 // This matches Bluesky's app.bsky.actor.profile pattern (no handle in record)
265 // Handles are mutable and resolved from DIDs via PLC, so they shouldn't be stored in immutable records
266
267 // ====================================================================================
268 // Part 2: TRUE E2E - Real Jetstream Firehose Consumer
269 // ====================================================================================
270 t.Run("2. Real Jetstream Firehose Consumption", func(t *testing.T) {
271 t.Logf("\n🔄 TRUE E2E: Subscribing to real Jetstream firehose...")
272
273 // Get PDS hostname for Jetstream filtering
274 pdsHostname := strings.TrimPrefix(pdsURL, "http://")
275 pdsHostname = strings.TrimPrefix(pdsHostname, "https://")
276 pdsHostname = strings.Split(pdsHostname, ":")[0] // Remove port
277
278 // Build Jetstream URL with filters
279 // Filter to our PDS and social.coves.community.profile collection
280 jetstreamURL := fmt.Sprintf("ws://%s:6008/subscribe?wantedCollections=social.coves.community.profile",
281 pdsHostname)
282
283 t.Logf(" Jetstream URL: %s", jetstreamURL)
284 t.Logf(" Looking for community DID: %s", community.DID)
285
286 // Channel to receive the event
287 eventChan := make(chan *jetstream.JetstreamEvent, 10)
288 errorChan := make(chan error, 1)
289 done := make(chan bool)
290
291 // Start Jetstream consumer in background
292 go func() {
293 err := subscribeToJetstream(ctx, jetstreamURL, community.DID, consumer, eventChan, errorChan, done)
294 if err != nil {
295 errorChan <- err
296 }
297 }()
298
299 // Wait for event or timeout
300 t.Logf("⏳ Waiting for Jetstream event (max 30 seconds)...")
301
302 select {
303 case event := <-eventChan:
304 t.Logf("✅ Received real Jetstream event!")
305 t.Logf(" Event DID: %s", event.Did)
306 t.Logf(" Collection: %s", event.Commit.Collection)
307 t.Logf(" Operation: %s", event.Commit.Operation)
308 t.Logf(" RKey: %s", event.Commit.RKey)
309
310 // Verify it's our community
311 if event.Did != community.DID {
312 t.Errorf("❌ Expected DID %s, got %s", community.DID, event.Did)
313 }
314
315 // Verify indexed in AppView database
316 t.Logf("\n🔍 Querying AppView database...")
317
318 indexed, err := communityRepo.GetByDID(ctx, community.DID)
319 if err != nil {
320 t.Fatalf("Community not indexed in AppView: %v", err)
321 }
322
323 t.Logf("✅ Community indexed in AppView:")
324 t.Logf(" DID: %s", indexed.DID)
325 t.Logf(" Handle: %s", indexed.Handle)
326 t.Logf(" DisplayName: %s", indexed.DisplayName)
327 t.Logf(" RecordURI: %s", indexed.RecordURI)
328
329 // V2: Verify record_uri points to COMMUNITY's own repo
330 expectedURIPrefix := "at://" + community.DID
331 if !strings.HasPrefix(indexed.RecordURI, expectedURIPrefix) {
332 t.Errorf("❌ V2: record_uri should point to community's repo\n Expected prefix: %s\n Got: %s",
333 expectedURIPrefix, indexed.RecordURI)
334 } else {
335 t.Logf("✅ V2: Record URI correctly points to community's own repository")
336 }
337
338 // Signal to stop Jetstream consumer
339 close(done)
340
341 case err := <-errorChan:
342 t.Fatalf("❌ Jetstream error: %v", err)
343
344 case <-time.After(30 * time.Second):
345 t.Fatalf("❌ Timeout: No Jetstream event received within 30 seconds")
346 }
347
348 t.Logf("\n✅ Part 2 Complete: TRUE E2E - PDS → Jetstream → Consumer → AppView ✓")
349 })
350 })
351
352 // ====================================================================================
353 // Part 3: XRPC HTTP Endpoints
354 // ====================================================================================
355 t.Run("3. XRPC HTTP Endpoints", func(t *testing.T) {
356 t.Run("Create via XRPC endpoint", func(t *testing.T) {
357 // Use Unix timestamp (seconds) instead of UnixNano to keep handle short
358 // NOTE: Both createdByDid and hostedByDid are derived server-side:
359 // - createdByDid: from JWT token (authenticated user)
360 // - hostedByDid: from instance configuration (security: prevents spoofing)
361 createReq := map[string]interface{}{
362 "name": fmt.Sprintf("xrpc-%d", time.Now().Unix()),
363 "displayName": "XRPC E2E Test",
364 "description": "Testing true end-to-end flow",
365 "visibility": "public",
366 "allowExternalDiscovery": true,
367 }
368
369 reqBody, marshalErr := json.Marshal(createReq)
370 if marshalErr != nil {
371 t.Fatalf("Failed to marshal request: %v", marshalErr)
372 }
373
374 // Step 1: Client POSTs to XRPC endpoint with JWT authentication
375 t.Logf("📡 Client → POST /xrpc/social.coves.community.create")
376 t.Logf(" Request: %s", string(reqBody))
377
378 req, err := http.NewRequest(http.MethodPost,
379 httpServer.URL+"/xrpc/social.coves.community.create",
380 bytes.NewBuffer(reqBody))
381 if err != nil {
382 t.Fatalf("Failed to create request: %v", err)
383 }
384 req.Header.Set("Content-Type", "application/json")
385 // Use real PDS access token for E2E authentication
386 req.Header.Set("Authorization", "Bearer "+accessToken)
387
388 resp, err := http.DefaultClient.Do(req)
389 if err != nil {
390 t.Fatalf("Failed to POST: %v", err)
391 }
392 defer func() { _ = resp.Body.Close() }()
393
394 if resp.StatusCode != http.StatusOK {
395 body, readErr := io.ReadAll(resp.Body)
396 if readErr != nil {
397 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
398 }
399 t.Logf("❌ XRPC Create Failed")
400 t.Logf(" Status: %d", resp.StatusCode)
401 t.Logf(" Response: %s", string(body))
402 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
403 }
404
405 var createResp struct {
406 URI string `json:"uri"`
407 CID string `json:"cid"`
408 DID string `json:"did"`
409 Handle string `json:"handle"`
410 }
411
412 if err := json.NewDecoder(resp.Body).Decode(&createResp); err != nil {
413 t.Fatalf("Failed to decode create response: %v", err)
414 }
415
416 t.Logf("✅ XRPC response received:")
417 t.Logf(" DID: %s", createResp.DID)
418 t.Logf(" Handle: %s", createResp.Handle)
419 t.Logf(" URI: %s", createResp.URI)
420
421 // Step 2: Simulate firehose consumer picking up the event
422 // NOTE: Using synthetic event for speed. Real Jetstream WebSocket testing
423 // happens in "Part 2: Real Jetstream Firehose Consumption" above.
424 t.Logf("🔄 Simulating Jetstream consumer indexing...")
425 rkey := utils.ExtractRKeyFromURI(createResp.URI)
426 // V2: Event comes from community's DID (community owns the repo)
427 event := jetstream.JetstreamEvent{
428 Did: createResp.DID,
429 TimeUS: time.Now().UnixMicro(),
430 Kind: "commit",
431 Commit: &jetstream.CommitEvent{
432 Rev: "test-rev",
433 Operation: "create",
434 Collection: "social.coves.community.profile",
435 RKey: rkey,
436 Record: map[string]interface{}{
437 // Note: No 'did' or 'handle' in record (atProto best practice)
438 // These are mutable and resolved from DIDs, not stored in immutable records
439 "name": createReq["name"],
440 "displayName": createReq["displayName"],
441 "description": createReq["description"],
442 "visibility": createReq["visibility"],
443 // Server-side derives these from JWT auth (instanceDID is the authenticated user)
444 "owner": instanceDID,
445 "createdBy": instanceDID,
446 "hostedBy": instanceDID,
447 "federation": map[string]interface{}{
448 "allowExternalDiscovery": createReq["allowExternalDiscovery"],
449 },
450 "createdAt": time.Now().Format(time.RFC3339),
451 },
452 CID: createResp.CID,
453 },
454 }
455 if handleErr := consumer.HandleEvent(context.Background(), &event); handleErr != nil {
456 t.Logf("Warning: failed to handle event: %v", handleErr)
457 }
458
459 // Step 3: Verify it's indexed in AppView
460 t.Logf("🔍 Querying AppView to verify indexing...")
461 var indexedCommunity communities.Community
462 err = db.QueryRow(`
463 SELECT did, handle, display_name, description
464 FROM communities
465 WHERE did = $1
466 `, createResp.DID).Scan(
467 &indexedCommunity.DID,
468 &indexedCommunity.Handle,
469 &indexedCommunity.DisplayName,
470 &indexedCommunity.Description,
471 )
472 if err != nil {
473 t.Fatalf("Community not indexed in AppView: %v", err)
474 }
475
476 t.Logf("✅ TRUE E2E FLOW COMPLETE:")
477 t.Logf(" Client → XRPC → PDS → Firehose → AppView ✓")
478 t.Logf(" Indexed community: %s (%s)", indexedCommunity.Handle, indexedCommunity.DisplayName)
479 })
480
481 t.Run("Get via XRPC endpoint", func(t *testing.T) {
482 // Create a community first (via service, so it's indexed)
483 community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
484
485 // GET via HTTP endpoint
486 resp, err := http.Get(fmt.Sprintf("%s/xrpc/social.coves.community.get?community=%s",
487 httpServer.URL, community.DID))
488 if err != nil {
489 t.Fatalf("Failed to GET: %v", err)
490 }
491 defer func() { _ = resp.Body.Close() }()
492
493 if resp.StatusCode != http.StatusOK {
494 body, readErr := io.ReadAll(resp.Body)
495 if readErr != nil {
496 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
497 }
498 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
499 }
500
501 var getCommunity communities.Community
502 if err := json.NewDecoder(resp.Body).Decode(&getCommunity); err != nil {
503 t.Fatalf("Failed to decode get response: %v", err)
504 }
505
506 t.Logf("Retrieved via XRPC HTTP endpoint:")
507 t.Logf(" DID: %s", getCommunity.DID)
508 t.Logf(" DisplayName: %s", getCommunity.DisplayName)
509
510 if getCommunity.DID != community.DID {
511 t.Errorf("DID mismatch: expected %s, got %s", community.DID, getCommunity.DID)
512 }
513 })
514
515 t.Run("List via XRPC endpoint", func(t *testing.T) {
516 // Create and index multiple communities
517 for i := 0; i < 3; i++ {
518 createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
519 }
520
521 resp, err := http.Get(fmt.Sprintf("%s/xrpc/social.coves.community.list?limit=10",
522 httpServer.URL))
523 if err != nil {
524 t.Fatalf("Failed to GET list: %v", err)
525 }
526 defer func() { _ = resp.Body.Close() }()
527
528 if resp.StatusCode != http.StatusOK {
529 body, readErr := io.ReadAll(resp.Body)
530 if readErr != nil {
531 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
532 }
533 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
534 }
535
536 var listResp struct {
537 Communities []communities.Community `json:"communities"`
538 Total int `json:"total"`
539 }
540
541 if err := json.NewDecoder(resp.Body).Decode(&listResp); err != nil {
542 t.Fatalf("Failed to decode list response: %v", err)
543 }
544
545 t.Logf("✅ Listed %d communities via XRPC", len(listResp.Communities))
546
547 if len(listResp.Communities) < 3 {
548 t.Errorf("Expected at least 3 communities, got %d", len(listResp.Communities))
549 }
550 })
551
552 t.Run("Subscribe via XRPC endpoint", func(t *testing.T) {
553 // Create a community to subscribe to
554 community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
555
556 // Get initial subscriber count
557 initialCommunity, err := communityRepo.GetByDID(ctx, community.DID)
558 if err != nil {
559 t.Fatalf("Failed to get initial community state: %v", err)
560 }
561 initialSubscriberCount := initialCommunity.SubscriberCount
562 t.Logf("Initial subscriber count: %d", initialSubscriberCount)
563
564 // Subscribe to the community with contentVisibility=5 (test max visibility)
565 // NOTE: HTTP API uses "community" field, but atProto record uses "subject" internally
566 subscribeReq := map[string]interface{}{
567 "community": community.DID,
568 "contentVisibility": 5, // Test with max visibility
569 }
570
571 reqBody, marshalErr := json.Marshal(subscribeReq)
572 if marshalErr != nil {
573 t.Fatalf("Failed to marshal subscribe request: %v", marshalErr)
574 }
575
576 // POST subscribe request
577 t.Logf("📡 Client → POST /xrpc/social.coves.community.subscribe")
578 t.Logf(" Subscribing to community: %s", community.DID)
579
580 req, err := http.NewRequest(http.MethodPost,
581 httpServer.URL+"/xrpc/social.coves.community.subscribe",
582 bytes.NewBuffer(reqBody))
583 if err != nil {
584 t.Fatalf("Failed to create request: %v", err)
585 }
586 req.Header.Set("Content-Type", "application/json")
587 // Use real PDS access token for E2E authentication
588 req.Header.Set("Authorization", "Bearer "+accessToken)
589
590 resp, err := http.DefaultClient.Do(req)
591 if err != nil {
592 t.Fatalf("Failed to POST subscribe: %v", err)
593 }
594 defer func() { _ = resp.Body.Close() }()
595
596 if resp.StatusCode != http.StatusOK {
597 body, readErr := io.ReadAll(resp.Body)
598 if readErr != nil {
599 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
600 }
601 t.Logf("❌ XRPC Subscribe Failed")
602 t.Logf(" Status: %d", resp.StatusCode)
603 t.Logf(" Response: %s", string(body))
604 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
605 }
606
607 var subscribeResp struct {
608 URI string `json:"uri"`
609 CID string `json:"cid"`
610 Existing bool `json:"existing"`
611 }
612
613 if err := json.NewDecoder(resp.Body).Decode(&subscribeResp); err != nil {
614 t.Fatalf("Failed to decode subscribe response: %v", err)
615 }
616
617 t.Logf("✅ XRPC subscribe response received:")
618 t.Logf(" URI: %s", subscribeResp.URI)
619 t.Logf(" CID: %s", subscribeResp.CID)
620 t.Logf(" Existing: %v", subscribeResp.Existing)
621
622 // Verify the subscription was written to PDS (in user's repository)
623 t.Logf("🔍 Verifying subscription record on PDS...")
624 pdsURL := os.Getenv("PDS_URL")
625 if pdsURL == "" {
626 pdsURL = "http://localhost:3001"
627 }
628
629 rkey := utils.ExtractRKeyFromURI(subscribeResp.URI)
630 // CRITICAL: Use correct collection name (record type, not XRPC endpoint)
631 collection := "social.coves.community.subscription"
632
633 pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
634 pdsURL, instanceDID, collection, rkey))
635 if pdsErr != nil {
636 t.Fatalf("Failed to fetch subscription record from PDS: %v", pdsErr)
637 }
638 defer func() {
639 if closeErr := pdsResp.Body.Close(); closeErr != nil {
640 t.Logf("Failed to close PDS response: %v", closeErr)
641 }
642 }()
643
644 if pdsResp.StatusCode != http.StatusOK {
645 body, _ := io.ReadAll(pdsResp.Body)
646 t.Fatalf("Subscription record not found on PDS: status %d, body: %s", pdsResp.StatusCode, string(body))
647 }
648
649 var pdsRecord struct {
650 Value map[string]interface{} `json:"value"`
651 }
652 if decodeErr := json.NewDecoder(pdsResp.Body).Decode(&pdsRecord); decodeErr != nil {
653 t.Fatalf("Failed to decode PDS record: %v", decodeErr)
654 }
655
656 t.Logf("✅ Subscription record found on PDS:")
657 t.Logf(" Subject (community): %v", pdsRecord.Value["subject"])
658 t.Logf(" ContentVisibility: %v", pdsRecord.Value["contentVisibility"])
659
660 // Verify the subject (community) DID matches
661 if pdsRecord.Value["subject"] != community.DID {
662 t.Errorf("Community DID mismatch: expected %s, got %v", community.DID, pdsRecord.Value["subject"])
663 }
664
665 // Verify contentVisibility was stored correctly
666 if cv, ok := pdsRecord.Value["contentVisibility"].(float64); ok {
667 if int(cv) != 5 {
668 t.Errorf("ContentVisibility mismatch: expected 5, got %v", cv)
669 }
670 } else {
671 t.Errorf("ContentVisibility not found or wrong type in PDS record")
672 }
673
674 // CRITICAL: Simulate Jetstream consumer indexing the subscription
675 // This is the MISSING PIECE - we need to verify the firehose event gets indexed
676 t.Logf("🔄 Simulating Jetstream consumer indexing subscription...")
677 subEvent := jetstream.JetstreamEvent{
678 Did: instanceDID,
679 TimeUS: time.Now().UnixMicro(),
680 Kind: "commit",
681 Commit: &jetstream.CommitEvent{
682 Rev: "test-sub-rev",
683 Operation: "create",
684 Collection: "social.coves.community.subscription", // CORRECT collection
685 RKey: rkey,
686 CID: subscribeResp.CID,
687 Record: map[string]interface{}{
688 "$type": "social.coves.community.subscription",
689 "subject": community.DID,
690 "contentVisibility": float64(5), // JSON numbers are float64
691 "createdAt": time.Now().Format(time.RFC3339),
692 },
693 },
694 }
695 if handleErr := consumer.HandleEvent(context.Background(), &subEvent); handleErr != nil {
696 t.Fatalf("Failed to handle subscription event: %v", handleErr)
697 }
698
699 // Verify subscription was indexed in AppView
700 t.Logf("🔍 Verifying subscription indexed in AppView...")
701 indexedSub, err := communityRepo.GetSubscription(ctx, instanceDID, community.DID)
702 if err != nil {
703 t.Fatalf("Subscription not indexed in AppView: %v", err)
704 }
705
706 t.Logf("✅ Subscription indexed in AppView:")
707 t.Logf(" User: %s", indexedSub.UserDID)
708 t.Logf(" Community: %s", indexedSub.CommunityDID)
709 t.Logf(" ContentVisibility: %d", indexedSub.ContentVisibility)
710 t.Logf(" RecordURI: %s", indexedSub.RecordURI)
711
712 // Verify contentVisibility was indexed correctly
713 if indexedSub.ContentVisibility != 5 {
714 t.Errorf("ContentVisibility not indexed correctly: expected 5, got %d", indexedSub.ContentVisibility)
715 }
716
717 // Verify subscriber count was incremented
718 t.Logf("🔍 Verifying subscriber count incremented...")
719 updatedCommunity, err := communityRepo.GetByDID(ctx, community.DID)
720 if err != nil {
721 t.Fatalf("Failed to get updated community: %v", err)
722 }
723
724 expectedCount := initialSubscriberCount + 1
725 if updatedCommunity.SubscriberCount != expectedCount {
726 t.Errorf("Subscriber count not incremented: expected %d, got %d",
727 expectedCount, updatedCommunity.SubscriberCount)
728 } else {
729 t.Logf("✅ Subscriber count incremented: %d → %d",
730 initialSubscriberCount, updatedCommunity.SubscriberCount)
731 }
732
733 t.Logf("✅ TRUE E2E SUBSCRIBE FLOW COMPLETE:")
734 t.Logf(" Client → XRPC Subscribe → PDS (user repo) → Firehose → Consumer → AppView ✓")
735 t.Logf(" ✓ Subscription written to PDS")
736 t.Logf(" ✓ Subscription indexed in AppView")
737 t.Logf(" ✓ ContentVisibility stored and indexed correctly (5)")
738 t.Logf(" ✓ Subscriber count incremented")
739 })
740
741 t.Run("Unsubscribe via XRPC endpoint", func(t *testing.T) {
742 // Create a community and subscribe to it first
743 community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
744
745 // Get initial subscriber count
746 initialCommunity, err := communityRepo.GetByDID(ctx, community.DID)
747 if err != nil {
748 t.Fatalf("Failed to get initial community state: %v", err)
749 }
750 initialSubscriberCount := initialCommunity.SubscriberCount
751 t.Logf("Initial subscriber count: %d", initialSubscriberCount)
752
753 // Subscribe first (using instance access token for instance user, with contentVisibility=3)
754 subscription, err := communityService.SubscribeToCommunity(ctx, instanceDID, accessToken, community.DID, 3)
755 if err != nil {
756 t.Fatalf("Failed to subscribe: %v", err)
757 }
758
759 // Index the subscription in AppView (simulate firehose event)
760 rkey := utils.ExtractRKeyFromURI(subscription.RecordURI)
761 subEvent := jetstream.JetstreamEvent{
762 Did: instanceDID,
763 TimeUS: time.Now().UnixMicro(),
764 Kind: "commit",
765 Commit: &jetstream.CommitEvent{
766 Rev: "test-sub-rev",
767 Operation: "create",
768 Collection: "social.coves.community.subscription", // CORRECT collection
769 RKey: rkey,
770 CID: subscription.RecordCID,
771 Record: map[string]interface{}{
772 "$type": "social.coves.community.subscription",
773 "subject": community.DID,
774 "contentVisibility": float64(3),
775 "createdAt": time.Now().Format(time.RFC3339),
776 },
777 },
778 }
779 if handleErr := consumer.HandleEvent(context.Background(), &subEvent); handleErr != nil {
780 t.Fatalf("Failed to handle subscription event: %v", handleErr)
781 }
782
783 // Verify subscription was indexed
784 _, err = communityRepo.GetSubscription(ctx, instanceDID, community.DID)
785 if err != nil {
786 t.Fatalf("Subscription not indexed: %v", err)
787 }
788
789 // Verify subscriber count incremented
790 midCommunity, err := communityRepo.GetByDID(ctx, community.DID)
791 if err != nil {
792 t.Fatalf("Failed to get community after subscribe: %v", err)
793 }
794 if midCommunity.SubscriberCount != initialSubscriberCount+1 {
795 t.Errorf("Subscriber count not incremented after subscribe: expected %d, got %d",
796 initialSubscriberCount+1, midCommunity.SubscriberCount)
797 }
798
799 t.Logf("📝 Subscription created and indexed: %s", subscription.RecordURI)
800
801 // Now unsubscribe via XRPC endpoint
802 unsubscribeReq := map[string]interface{}{
803 "community": community.DID,
804 }
805
806 reqBody, marshalErr := json.Marshal(unsubscribeReq)
807 if marshalErr != nil {
808 t.Fatalf("Failed to marshal unsubscribe request: %v", marshalErr)
809 }
810
811 // POST unsubscribe request
812 t.Logf("📡 Client → POST /xrpc/social.coves.community.unsubscribe")
813 t.Logf(" Unsubscribing from community: %s", community.DID)
814
815 req, err := http.NewRequest(http.MethodPost,
816 httpServer.URL+"/xrpc/social.coves.community.unsubscribe",
817 bytes.NewBuffer(reqBody))
818 if err != nil {
819 t.Fatalf("Failed to create request: %v", err)
820 }
821 req.Header.Set("Content-Type", "application/json")
822 // Use real PDS access token for E2E authentication
823 req.Header.Set("Authorization", "Bearer "+accessToken)
824
825 resp, err := http.DefaultClient.Do(req)
826 if err != nil {
827 t.Fatalf("Failed to POST unsubscribe: %v", err)
828 }
829 defer func() { _ = resp.Body.Close() }()
830
831 if resp.StatusCode != http.StatusOK {
832 body, readErr := io.ReadAll(resp.Body)
833 if readErr != nil {
834 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
835 }
836 t.Logf("❌ XRPC Unsubscribe Failed")
837 t.Logf(" Status: %d", resp.StatusCode)
838 t.Logf(" Response: %s", string(body))
839 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
840 }
841
842 var unsubscribeResp struct {
843 Success bool `json:"success"`
844 }
845
846 if err := json.NewDecoder(resp.Body).Decode(&unsubscribeResp); err != nil {
847 t.Fatalf("Failed to decode unsubscribe response: %v", err)
848 }
849
850 t.Logf("✅ XRPC unsubscribe response received:")
851 t.Logf(" Success: %v", unsubscribeResp.Success)
852
853 if !unsubscribeResp.Success {
854 t.Errorf("Expected success: true, got: %v", unsubscribeResp.Success)
855 }
856
857 // Verify the subscription record was deleted from PDS
858 t.Logf("🔍 Verifying subscription record deleted from PDS...")
859 pdsURL := os.Getenv("PDS_URL")
860 if pdsURL == "" {
861 pdsURL = "http://localhost:3001"
862 }
863
864 // CRITICAL: Use correct collection name (record type, not XRPC endpoint)
865 collection := "social.coves.community.subscription"
866 pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
867 pdsURL, instanceDID, collection, rkey))
868 if pdsErr != nil {
869 t.Fatalf("Failed to query PDS: %v", pdsErr)
870 }
871 defer func() {
872 if closeErr := pdsResp.Body.Close(); closeErr != nil {
873 t.Logf("Failed to close PDS response: %v", closeErr)
874 }
875 }()
876
877 // Should return 404 since record was deleted
878 if pdsResp.StatusCode == http.StatusOK {
879 t.Errorf("❌ Subscription record still exists on PDS (expected 404, got 200)")
880 } else {
881 t.Logf("✅ Subscription record successfully deleted from PDS (status: %d)", pdsResp.StatusCode)
882 }
883
884 // CRITICAL: Simulate Jetstream consumer indexing the DELETE event
885 t.Logf("🔄 Simulating Jetstream consumer indexing DELETE event...")
886 deleteEvent := jetstream.JetstreamEvent{
887 Did: instanceDID,
888 TimeUS: time.Now().UnixMicro(),
889 Kind: "commit",
890 Commit: &jetstream.CommitEvent{
891 Rev: "test-unsub-rev",
892 Operation: "delete",
893 Collection: "social.coves.community.subscription",
894 RKey: rkey,
895 CID: "", // No CID on deletes
896 Record: nil, // No record data on deletes
897 },
898 }
899 if handleErr := consumer.HandleEvent(context.Background(), &deleteEvent); handleErr != nil {
900 t.Fatalf("Failed to handle delete event: %v", handleErr)
901 }
902
903 // Verify subscription was removed from AppView
904 t.Logf("🔍 Verifying subscription removed from AppView...")
905 _, err = communityRepo.GetSubscription(ctx, instanceDID, community.DID)
906 if err == nil {
907 t.Errorf("❌ Subscription still exists in AppView (should be deleted)")
908 } else if !communities.IsNotFound(err) {
909 t.Fatalf("Unexpected error querying subscription: %v", err)
910 } else {
911 t.Logf("✅ Subscription removed from AppView")
912 }
913
914 // Verify subscriber count was decremented
915 t.Logf("🔍 Verifying subscriber count decremented...")
916 finalCommunity, err := communityRepo.GetByDID(ctx, community.DID)
917 if err != nil {
918 t.Fatalf("Failed to get final community state: %v", err)
919 }
920
921 if finalCommunity.SubscriberCount != initialSubscriberCount {
922 t.Errorf("Subscriber count not decremented: expected %d, got %d",
923 initialSubscriberCount, finalCommunity.SubscriberCount)
924 } else {
925 t.Logf("✅ Subscriber count decremented: %d → %d",
926 initialSubscriberCount+1, finalCommunity.SubscriberCount)
927 }
928
929 t.Logf("✅ TRUE E2E UNSUBSCRIBE FLOW COMPLETE:")
930 t.Logf(" Client → XRPC Unsubscribe → PDS Delete → Firehose → Consumer → AppView ✓")
931 t.Logf(" ✓ Subscription deleted from PDS")
932 t.Logf(" ✓ Subscription removed from AppView")
933 t.Logf(" ✓ Subscriber count decremented")
934 })
935
936 t.Run("Block via XRPC endpoint", func(t *testing.T) {
937 // Create a community to block
938 community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
939
940 t.Logf("🚫 Blocking community via XRPC endpoint...")
941 blockReq := map[string]interface{}{
942 "community": community.DID,
943 }
944
945 blockJSON, err := json.Marshal(blockReq)
946 if err != nil {
947 t.Fatalf("Failed to marshal block request: %v", err)
948 }
949
950 req, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.blockCommunity", bytes.NewBuffer(blockJSON))
951 if err != nil {
952 t.Fatalf("Failed to create block request: %v", err)
953 }
954 req.Header.Set("Content-Type", "application/json")
955 req.Header.Set("Authorization", "Bearer "+accessToken)
956
957 resp, err := http.DefaultClient.Do(req)
958 if err != nil {
959 t.Fatalf("Failed to POST block: %v", err)
960 }
961 defer func() { _ = resp.Body.Close() }()
962
963 if resp.StatusCode != http.StatusOK {
964 body, readErr := io.ReadAll(resp.Body)
965 if readErr != nil {
966 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
967 }
968 t.Logf("❌ XRPC Block Failed")
969 t.Logf(" Status: %d", resp.StatusCode)
970 t.Logf(" Response: %s", string(body))
971 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
972 }
973
974 var blockResp struct {
975 Block struct {
976 RecordURI string `json:"recordUri"`
977 RecordCID string `json:"recordCid"`
978 } `json:"block"`
979 }
980
981 if err := json.NewDecoder(resp.Body).Decode(&blockResp); err != nil {
982 t.Fatalf("Failed to decode block response: %v", err)
983 }
984
985 t.Logf("✅ XRPC block response received:")
986 t.Logf(" RecordURI: %s", blockResp.Block.RecordURI)
987 t.Logf(" RecordCID: %s", blockResp.Block.RecordCID)
988
989 // Extract rkey from URI for verification
990 rkey := ""
991 if uriParts := strings.Split(blockResp.Block.RecordURI, "/"); len(uriParts) >= 4 {
992 rkey = uriParts[len(uriParts)-1]
993 }
994
995 // Verify the block record exists on PDS
996 t.Logf("🔍 Verifying block record exists on PDS...")
997 collection := "social.coves.community.block"
998 pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
999 pdsURL, instanceDID, collection, rkey))
1000 if pdsErr != nil {
1001 t.Fatalf("Failed to query PDS: %v", pdsErr)
1002 }
1003 defer func() {
1004 if closeErr := pdsResp.Body.Close(); closeErr != nil {
1005 t.Logf("Failed to close PDS response: %v", closeErr)
1006 }
1007 }()
1008
1009 if pdsResp.StatusCode != http.StatusOK {
1010 body, readErr := io.ReadAll(pdsResp.Body)
1011 if readErr != nil {
1012 t.Fatalf("Block record not found on PDS (status: %d, failed to read body: %v)", pdsResp.StatusCode, readErr)
1013 }
1014 t.Fatalf("Block record not found on PDS (status: %d): %s", pdsResp.StatusCode, string(body))
1015 }
1016 t.Logf("✅ Block record exists on PDS")
1017
1018 // CRITICAL: Simulate Jetstream consumer indexing the block
1019 t.Logf("🔄 Simulating Jetstream consumer indexing block event...")
1020 blockEvent := jetstream.JetstreamEvent{
1021 Did: instanceDID,
1022 TimeUS: time.Now().UnixMicro(),
1023 Kind: "commit",
1024 Commit: &jetstream.CommitEvent{
1025 Rev: "test-block-rev",
1026 Operation: "create",
1027 Collection: "social.coves.community.block",
1028 RKey: rkey,
1029 CID: blockResp.Block.RecordCID,
1030 Record: map[string]interface{}{
1031 "subject": community.DID,
1032 "createdAt": time.Now().Format(time.RFC3339),
1033 },
1034 },
1035 }
1036 if handleErr := consumer.HandleEvent(context.Background(), &blockEvent); handleErr != nil {
1037 t.Fatalf("Failed to handle block event: %v", handleErr)
1038 }
1039
1040 // Verify block was indexed in AppView
1041 t.Logf("🔍 Verifying block indexed in AppView...")
1042 block, err := communityRepo.GetBlock(ctx, instanceDID, community.DID)
1043 if err != nil {
1044 t.Fatalf("Failed to get block from AppView: %v", err)
1045 }
1046 if block.RecordURI != blockResp.Block.RecordURI {
1047 t.Errorf("RecordURI mismatch: expected %s, got %s", blockResp.Block.RecordURI, block.RecordURI)
1048 }
1049
1050 t.Logf("✅ TRUE E2E BLOCK FLOW COMPLETE:")
1051 t.Logf(" Client → XRPC Block → PDS Create → Firehose → Consumer → AppView ✓")
1052 t.Logf(" ✓ Block record created on PDS")
1053 t.Logf(" ✓ Block indexed in AppView")
1054 })
1055
1056 t.Run("Unblock via XRPC endpoint", func(t *testing.T) {
1057 // Create a community and block it first
1058 community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
1059
1060 // Block the community
1061 t.Logf("🚫 Blocking community first...")
1062 blockReq := map[string]interface{}{
1063 "community": community.DID,
1064 }
1065 blockJSON, err := json.Marshal(blockReq)
1066 if err != nil {
1067 t.Fatalf("Failed to marshal block request: %v", err)
1068 }
1069
1070 blockHttpReq, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.blockCommunity", bytes.NewBuffer(blockJSON))
1071 if err != nil {
1072 t.Fatalf("Failed to create block request: %v", err)
1073 }
1074 blockHttpReq.Header.Set("Content-Type", "application/json")
1075 blockHttpReq.Header.Set("Authorization", "Bearer "+accessToken)
1076
1077 blockResp, err := http.DefaultClient.Do(blockHttpReq)
1078 if err != nil {
1079 t.Fatalf("Failed to POST block: %v", err)
1080 }
1081
1082 var blockRespData struct {
1083 Block struct {
1084 RecordURI string `json:"recordUri"`
1085 } `json:"block"`
1086 }
1087 if err := json.NewDecoder(blockResp.Body).Decode(&blockRespData); err != nil {
1088 func() { _ = blockResp.Body.Close() }()
1089 t.Fatalf("Failed to decode block response: %v", err)
1090 }
1091 func() { _ = blockResp.Body.Close() }()
1092
1093 rkey := ""
1094 if uriParts := strings.Split(blockRespData.Block.RecordURI, "/"); len(uriParts) >= 4 {
1095 rkey = uriParts[len(uriParts)-1]
1096 }
1097
1098 // Index the block via consumer
1099 blockEvent := jetstream.JetstreamEvent{
1100 Did: instanceDID,
1101 TimeUS: time.Now().UnixMicro(),
1102 Kind: "commit",
1103 Commit: &jetstream.CommitEvent{
1104 Rev: "test-block-rev",
1105 Operation: "create",
1106 Collection: "social.coves.community.block",
1107 RKey: rkey,
1108 CID: "test-block-cid",
1109 Record: map[string]interface{}{
1110 "subject": community.DID,
1111 "createdAt": time.Now().Format(time.RFC3339),
1112 },
1113 },
1114 }
1115 if handleErr := consumer.HandleEvent(context.Background(), &blockEvent); handleErr != nil {
1116 t.Fatalf("Failed to handle block event: %v", handleErr)
1117 }
1118
1119 // Now unblock the community
1120 t.Logf("✅ Unblocking community via XRPC endpoint...")
1121 unblockReq := map[string]interface{}{
1122 "community": community.DID,
1123 }
1124
1125 unblockJSON, err := json.Marshal(unblockReq)
1126 if err != nil {
1127 t.Fatalf("Failed to marshal unblock request: %v", err)
1128 }
1129
1130 req, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.unblockCommunity", bytes.NewBuffer(unblockJSON))
1131 if err != nil {
1132 t.Fatalf("Failed to create unblock request: %v", err)
1133 }
1134 req.Header.Set("Content-Type", "application/json")
1135 req.Header.Set("Authorization", "Bearer "+accessToken)
1136
1137 resp, err := http.DefaultClient.Do(req)
1138 if err != nil {
1139 t.Fatalf("Failed to POST unblock: %v", err)
1140 }
1141 defer func() { _ = resp.Body.Close() }()
1142
1143 if resp.StatusCode != http.StatusOK {
1144 body, readErr := io.ReadAll(resp.Body)
1145 if readErr != nil {
1146 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
1147 }
1148 t.Logf("❌ XRPC Unblock Failed")
1149 t.Logf(" Status: %d", resp.StatusCode)
1150 t.Logf(" Response: %s", string(body))
1151 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
1152 }
1153
1154 var unblockResp struct {
1155 Success bool `json:"success"`
1156 }
1157
1158 if err := json.NewDecoder(resp.Body).Decode(&unblockResp); err != nil {
1159 t.Fatalf("Failed to decode unblock response: %v", err)
1160 }
1161
1162 if !unblockResp.Success {
1163 t.Errorf("Expected success: true, got: %v", unblockResp.Success)
1164 }
1165
1166 // Verify the block record was deleted from PDS
1167 t.Logf("🔍 Verifying block record deleted from PDS...")
1168 collection := "social.coves.community.block"
1169 pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
1170 pdsURL, instanceDID, collection, rkey))
1171 if pdsErr != nil {
1172 t.Fatalf("Failed to query PDS: %v", pdsErr)
1173 }
1174 defer func() {
1175 if closeErr := pdsResp.Body.Close(); closeErr != nil {
1176 t.Logf("Failed to close PDS response: %v", closeErr)
1177 }
1178 }()
1179
1180 if pdsResp.StatusCode == http.StatusOK {
1181 t.Errorf("❌ Block record still exists on PDS (expected 404, got 200)")
1182 } else {
1183 t.Logf("✅ Block record successfully deleted from PDS (status: %d)", pdsResp.StatusCode)
1184 }
1185
1186 // CRITICAL: Simulate Jetstream consumer indexing the DELETE event
1187 t.Logf("🔄 Simulating Jetstream consumer indexing DELETE event...")
1188 deleteEvent := jetstream.JetstreamEvent{
1189 Did: instanceDID,
1190 TimeUS: time.Now().UnixMicro(),
1191 Kind: "commit",
1192 Commit: &jetstream.CommitEvent{
1193 Rev: "test-unblock-rev",
1194 Operation: "delete",
1195 Collection: "social.coves.community.block",
1196 RKey: rkey,
1197 CID: "",
1198 Record: nil,
1199 },
1200 }
1201 if handleErr := consumer.HandleEvent(context.Background(), &deleteEvent); handleErr != nil {
1202 t.Fatalf("Failed to handle delete event: %v", handleErr)
1203 }
1204
1205 // Verify block was removed from AppView
1206 t.Logf("🔍 Verifying block removed from AppView...")
1207 _, err = communityRepo.GetBlock(ctx, instanceDID, community.DID)
1208 if err == nil {
1209 t.Errorf("❌ Block still exists in AppView (should be deleted)")
1210 } else if !communities.IsNotFound(err) {
1211 t.Fatalf("Unexpected error querying block: %v", err)
1212 } else {
1213 t.Logf("✅ Block removed from AppView")
1214 }
1215
1216 t.Logf("✅ TRUE E2E UNBLOCK FLOW COMPLETE:")
1217 t.Logf(" Client → XRPC Unblock → PDS Delete → Firehose → Consumer → AppView ✓")
1218 t.Logf(" ✓ Block deleted from PDS")
1219 t.Logf(" ✓ Block removed from AppView")
1220 })
1221
1222 t.Run("Block fails without authentication", func(t *testing.T) {
1223 // Create a community to attempt blocking
1224 community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
1225
1226 t.Logf("🔒 Attempting to block community without auth token...")
1227 blockReq := map[string]interface{}{
1228 "community": community.DID,
1229 }
1230
1231 blockJSON, err := json.Marshal(blockReq)
1232 if err != nil {
1233 t.Fatalf("Failed to marshal block request: %v", err)
1234 }
1235
1236 req, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.blockCommunity", bytes.NewBuffer(blockJSON))
1237 if err != nil {
1238 t.Fatalf("Failed to create block request: %v", err)
1239 }
1240 req.Header.Set("Content-Type", "application/json")
1241 // NO Authorization header
1242
1243 resp, err := http.DefaultClient.Do(req)
1244 if err != nil {
1245 t.Fatalf("Failed to POST block: %v", err)
1246 }
1247 defer func() { _ = resp.Body.Close() }()
1248
1249 // Should fail with 401 Unauthorized
1250 if resp.StatusCode != http.StatusUnauthorized {
1251 body, _ := io.ReadAll(resp.Body)
1252 t.Errorf("Expected 401 Unauthorized, got %d: %s", resp.StatusCode, string(body))
1253 } else {
1254 t.Logf("✅ Block correctly rejected without authentication (401)")
1255 }
1256 })
1257
1258 t.Run("Update via XRPC endpoint", func(t *testing.T) {
1259 // Create a community first (via service, so it's indexed)
1260 community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
1261
1262 // Update the community
1263 newDisplayName := "Updated E2E Test Community"
1264 newDescription := "This community has been updated"
1265 newVisibility := "unlisted"
1266
1267 // NOTE: updatedByDid is derived from JWT token, not provided in request
1268 updateReq := map[string]interface{}{
1269 "communityDid": community.DID,
1270 "displayName": newDisplayName,
1271 "description": newDescription,
1272 "visibility": newVisibility,
1273 }
1274
1275 reqBody, marshalErr := json.Marshal(updateReq)
1276 if marshalErr != nil {
1277 t.Fatalf("Failed to marshal update request: %v", marshalErr)
1278 }
1279
1280 // POST update request with JWT authentication
1281 t.Logf("📡 Client → POST /xrpc/social.coves.community.update")
1282 t.Logf(" Updating community: %s", community.DID)
1283
1284 req, err := http.NewRequest(http.MethodPost,
1285 httpServer.URL+"/xrpc/social.coves.community.update",
1286 bytes.NewBuffer(reqBody))
1287 if err != nil {
1288 t.Fatalf("Failed to create request: %v", err)
1289 }
1290 req.Header.Set("Content-Type", "application/json")
1291 // Use real PDS access token for E2E authentication
1292 req.Header.Set("Authorization", "Bearer "+accessToken)
1293
1294 resp, err := http.DefaultClient.Do(req)
1295 if err != nil {
1296 t.Fatalf("Failed to POST update: %v", err)
1297 }
1298 defer func() { _ = resp.Body.Close() }()
1299
1300 if resp.StatusCode != http.StatusOK {
1301 body, readErr := io.ReadAll(resp.Body)
1302 if readErr != nil {
1303 t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
1304 }
1305 t.Logf("❌ XRPC Update Failed")
1306 t.Logf(" Status: %d", resp.StatusCode)
1307 t.Logf(" Response: %s", string(body))
1308 t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
1309 }
1310
1311 var updateResp struct {
1312 URI string `json:"uri"`
1313 CID string `json:"cid"`
1314 DID string `json:"did"`
1315 Handle string `json:"handle"`
1316 }
1317
1318 if err := json.NewDecoder(resp.Body).Decode(&updateResp); err != nil {
1319 t.Fatalf("Failed to decode update response: %v", err)
1320 }
1321
1322 t.Logf("✅ XRPC update response received:")
1323 t.Logf(" DID: %s", updateResp.DID)
1324 t.Logf(" URI: %s", updateResp.URI)
1325 t.Logf(" CID: %s (changed after update)", updateResp.CID)
1326
1327 // Verify the CID changed (update creates a new version)
1328 if updateResp.CID == community.RecordCID {
1329 t.Logf("⚠️ Warning: CID did not change after update (expected for a new version)")
1330 }
1331
1332 // Simulate Jetstream consumer picking up the update event
1333 t.Logf("🔄 Simulating Jetstream consumer indexing update...")
1334 rkey := utils.ExtractRKeyFromURI(updateResp.URI)
1335
1336 // Fetch updated record from PDS
1337 pdsURL := os.Getenv("PDS_URL")
1338 if pdsURL == "" {
1339 pdsURL = "http://localhost:3001"
1340 }
1341
1342 collection := "social.coves.community.profile"
1343 pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
1344 pdsURL, community.DID, collection, rkey))
1345 if pdsErr != nil {
1346 t.Fatalf("Failed to fetch updated PDS record: %v", pdsErr)
1347 }
1348 defer func() {
1349 if closeErr := pdsResp.Body.Close(); closeErr != nil {
1350 t.Logf("Failed to close PDS response: %v", closeErr)
1351 }
1352 }()
1353
1354 var pdsRecord struct {
1355 Value map[string]interface{} `json:"value"`
1356 CID string `json:"cid"`
1357 }
1358 if decodeErr := json.NewDecoder(pdsResp.Body).Decode(&pdsRecord); decodeErr != nil {
1359 t.Fatalf("Failed to decode PDS record: %v", decodeErr)
1360 }
1361
1362 // Create update event for consumer
1363 updateEvent := jetstream.JetstreamEvent{
1364 Did: community.DID,
1365 TimeUS: time.Now().UnixMicro(),
1366 Kind: "commit",
1367 Commit: &jetstream.CommitEvent{
1368 Rev: "test-update-rev",
1369 Operation: "update",
1370 Collection: collection,
1371 RKey: rkey,
1372 CID: pdsRecord.CID,
1373 Record: pdsRecord.Value,
1374 },
1375 }
1376
1377 if handleErr := consumer.HandleEvent(context.Background(), &updateEvent); handleErr != nil {
1378 t.Fatalf("Failed to handle update event: %v", handleErr)
1379 }
1380
1381 // Verify update was indexed in AppView
1382 t.Logf("🔍 Querying AppView to verify update was indexed...")
1383 updated, err := communityService.GetCommunity(ctx, community.DID)
1384 if err != nil {
1385 t.Fatalf("Failed to get updated community: %v", err)
1386 }
1387
1388 t.Logf("✅ Update indexed in AppView:")
1389 t.Logf(" DisplayName: %s (was: %s)", updated.DisplayName, community.DisplayName)
1390 t.Logf(" Description: %s", updated.Description)
1391 t.Logf(" Visibility: %s (was: %s)", updated.Visibility, community.Visibility)
1392
1393 // Verify the updates were applied
1394 if updated.DisplayName != newDisplayName {
1395 t.Errorf("DisplayName not updated: expected %s, got %s", newDisplayName, updated.DisplayName)
1396 }
1397 if updated.Description != newDescription {
1398 t.Errorf("Description not updated: expected %s, got %s", newDescription, updated.Description)
1399 }
1400 if updated.Visibility != newVisibility {
1401 t.Errorf("Visibility not updated: expected %s, got %s", newVisibility, updated.Visibility)
1402 }
1403
1404 t.Logf("✅ TRUE E2E UPDATE FLOW COMPLETE:")
1405 t.Logf(" Client → XRPC Update → PDS → Firehose → AppView ✓")
1406 })
1407
1408 t.Logf("\n✅ Part 3 Complete: All XRPC HTTP endpoints working ✓")
1409 })
1410
1411 divider := strings.Repeat("=", 80)
1412 t.Logf("\n%s", divider)
1413 t.Logf("✅ TRUE END-TO-END TEST COMPLETE - V2 COMMUNITIES ARCHITECTURE")
1414 t.Logf("%s", divider)
1415 t.Logf("\n🎯 Complete Flow Tested:")
1416 t.Logf(" 1. HTTP Request → Service Layer")
1417 t.Logf(" 2. Service → PDS Account Creation (com.atproto.server.createAccount)")
1418 t.Logf(" 3. Service → PDS Record Write (at://community_did/profile/self)")
1419 t.Logf(" 4. PDS → Jetstream Firehose (REAL WebSocket subscription!)")
1420 t.Logf(" 5. Jetstream → Consumer Event Handler")
1421 t.Logf(" 6. Consumer → AppView PostgreSQL Database")
1422 t.Logf(" 7. AppView DB → XRPC HTTP Endpoints")
1423 t.Logf(" 8. XRPC → Client Response")
1424 t.Logf("\n✅ V2 Architecture Verified:")
1425 t.Logf(" ✓ Community owns its own PDS account")
1426 t.Logf(" ✓ Community owns its own repository (at://community_did/...)")
1427 t.Logf(" ✓ PDS manages signing keypair (we only store credentials)")
1428 t.Logf(" ✓ Real Jetstream firehose event consumption")
1429 t.Logf(" ✓ True portability (community can migrate instances)")
1430 t.Logf(" ✓ Full atProto compliance")
1431 t.Logf("\n%s", divider)
1432 t.Logf("🚀 V2 Communities: Production Ready!")
1433 t.Logf("%s\n", divider)
1434}
1435
1436// Helper: create and index a community (simulates consumer indexing for fast test setup)
1437// NOTE: This simulates the firehose event for speed. For TRUE E2E testing with real
1438// Jetstream WebSocket subscription, see "Part 2: Real Jetstream Firehose Consumption" above.
1439func createAndIndexCommunity(t *testing.T, service communities.Service, consumer *jetstream.CommunityEventConsumer, instanceDID, pdsURL string) *communities.Community {
1440 // Use nanoseconds % 1 billion to get unique but short names
1441 // This avoids handle collisions when creating multiple communities quickly
1442 uniqueID := time.Now().UnixNano() % 1000000000
1443 req := communities.CreateCommunityRequest{
1444 Name: fmt.Sprintf("test-%d", uniqueID),
1445 DisplayName: "Test Community",
1446 Description: "Test",
1447 Visibility: "public",
1448 CreatedByDID: instanceDID,
1449 HostedByDID: instanceDID,
1450 AllowExternalDiscovery: true,
1451 }
1452
1453 community, err := service.CreateCommunity(context.Background(), req)
1454 if err != nil {
1455 t.Fatalf("Failed to create: %v", err)
1456 }
1457
1458 // Fetch from PDS to get full record
1459 // V2: Record lives in community's own repository (at://community.DID/...)
1460 collection := "social.coves.community.profile"
1461 rkey := utils.ExtractRKeyFromURI(community.RecordURI)
1462
1463 pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
1464 pdsURL, community.DID, collection, rkey))
1465 if pdsErr != nil {
1466 t.Fatalf("Failed to fetch PDS record: %v", pdsErr)
1467 }
1468 defer func() {
1469 if closeErr := pdsResp.Body.Close(); closeErr != nil {
1470 t.Logf("Failed to close PDS response: %v", closeErr)
1471 }
1472 }()
1473
1474 var pdsRecord struct {
1475 Value map[string]interface{} `json:"value"`
1476 CID string `json:"cid"`
1477 }
1478 if decodeErr := json.NewDecoder(pdsResp.Body).Decode(&pdsRecord); decodeErr != nil {
1479 t.Fatalf("Failed to decode PDS record: %v", decodeErr)
1480 }
1481
1482 // Simulate firehose event for fast indexing
1483 // V2: Event comes from community's DID (community owns the repo)
1484 // NOTE: This bypasses real Jetstream WebSocket for speed. Real firehose testing
1485 // happens in "Part 2: Real Jetstream Firehose Consumption" above.
1486 event := jetstream.JetstreamEvent{
1487 Did: community.DID,
1488 TimeUS: time.Now().UnixMicro(),
1489 Kind: "commit",
1490 Commit: &jetstream.CommitEvent{
1491 Rev: "test",
1492 Operation: "create",
1493 Collection: collection,
1494 RKey: rkey,
1495 CID: pdsRecord.CID,
1496 Record: pdsRecord.Value,
1497 },
1498 }
1499
1500 if handleErr := consumer.HandleEvent(context.Background(), &event); handleErr != nil {
1501 t.Logf("Warning: failed to handle event: %v", handleErr)
1502 }
1503
1504 return community
1505}
1506
1507// queryPDSAccount queries the PDS to verify an account exists
1508// Returns the account's DID and handle if found
1509func queryPDSAccount(pdsURL, handle string) (string, string, error) {
1510 // Use com.atproto.identity.resolveHandle to verify account exists
1511 resp, err := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.identity.resolveHandle?handle=%s", pdsURL, handle))
1512 if err != nil {
1513 return "", "", fmt.Errorf("failed to query PDS: %w", err)
1514 }
1515 defer func() { _ = resp.Body.Close() }()
1516
1517 if resp.StatusCode != http.StatusOK {
1518 body, readErr := io.ReadAll(resp.Body)
1519 if readErr != nil {
1520 return "", "", fmt.Errorf("account not found (status %d, failed to read body: %w)", resp.StatusCode, readErr)
1521 }
1522 return "", "", fmt.Errorf("account not found (status %d): %s", resp.StatusCode, string(body))
1523 }
1524
1525 var result struct {
1526 DID string `json:"did"`
1527 }
1528
1529 if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
1530 return "", "", fmt.Errorf("failed to decode response: %w", err)
1531 }
1532
1533 return result.DID, handle, nil
1534}
1535
1536// subscribeToJetstream subscribes to real Jetstream firehose and processes events
1537// This enables TRUE E2E testing: PDS → Jetstream → Consumer → AppView
1538func subscribeToJetstream(
1539 ctx context.Context,
1540 jetstreamURL string,
1541 targetDID string,
1542 consumer *jetstream.CommunityEventConsumer,
1543 eventChan chan<- *jetstream.JetstreamEvent,
1544 errorChan chan<- error,
1545 done <-chan bool,
1546) error {
1547 // Import needed for websocket
1548 // Note: We'll use the gorilla websocket library
1549 conn, _, err := websocket.DefaultDialer.Dial(jetstreamURL, nil)
1550 if err != nil {
1551 return fmt.Errorf("failed to connect to Jetstream: %w", err)
1552 }
1553 defer func() { _ = conn.Close() }()
1554
1555 // Read messages until we find our event or receive done signal
1556 for {
1557 select {
1558 case <-done:
1559 return nil
1560 case <-ctx.Done():
1561 return ctx.Err()
1562 default:
1563 // Set read deadline to avoid blocking forever
1564 if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
1565 return fmt.Errorf("failed to set read deadline: %w", err)
1566 }
1567
1568 var event jetstream.JetstreamEvent
1569 err := conn.ReadJSON(&event)
1570 if err != nil {
1571 // Check if it's a timeout (expected)
1572 if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
1573 return nil
1574 }
1575 if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
1576 continue // Timeout is expected, keep listening
1577 }
1578 // For other errors, don't retry reading from a broken connection
1579 return fmt.Errorf("failed to read Jetstream message: %w", err)
1580 }
1581
1582 // Check if this is the event we're looking for
1583 if event.Did == targetDID && event.Kind == "commit" {
1584 // Process the event through the consumer
1585 if err := consumer.HandleEvent(ctx, &event); err != nil {
1586 return fmt.Errorf("failed to process event: %w", err)
1587 }
1588
1589 // Send to channel so test can verify
1590 select {
1591 case eventChan <- &event:
1592 return nil
1593 case <-time.After(1 * time.Second):
1594 return fmt.Errorf("timeout sending event to channel")
1595 }
1596 }
1597 }
1598 }
1599}