···

       
1
       
 
       
package e2e

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
4
       
 
       
	"bytes"

     

       
5
       
 
       
	"context"

     

       
6
       
 
       
	"database/sql"

     
···

       
11
       
 
       
	"testing"

     

       
12
       
 
       
	"time"

     

       
13
       
 
       


     

       
14
       
-
       
	"Coves/internal/atproto/identity"

     

       
15
       
-
       
	"Coves/internal/atproto/jetstream"

     

       
16
       
-
       
	"Coves/internal/core/users"

     

       
17
       
-
       
	"Coves/internal/db/postgres"

     

       
18
       
 
       
	_ "github.com/lib/pq"

     

       
19
       
 
       
	"github.com/pressly/goose/v3"

     

       
20
       
 
       
)

     
···

       
31
       
 
       
//   - Test database on localhost:5434

     

       
32
       
 
       
//

     

       
33
       
 
       
// Run with:

     

       
34
       
-
       
//   make e2e-up  # Start infrastructure

     

       
35
       
-
       
//   go run ./cmd/server &  # Start AppView

     

       
36
       
-
       
//   go test ./tests/integration -run TestE2E_UserSignup -v

     

       
0
       
 
       

     

       
37
       
 
       
func TestE2E_UserSignup(t *testing.T) {

     

       
38
       
 
       
	if testing.Short() {

     

       
39
       
 
       
		t.Skip("Skipping E2E test in short mode")

     
···

       
55
       
 
       
	}

     

       
56
       
 
       


     

       
57
       
 
       
	db := setupTestDB(t)

     

       
58
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
59
       
 
       


     

       
60
       
 
       
	// Set up services

     

       
61
       
 
       
	userRepo := postgres.NewUserRepository(db)

     
···

       
253
       
 
       
	if err != nil {

     

       
254
       
 
       
		return "", fmt.Errorf("failed to create invite code: %w", err)

     

       
255
       
 
       
	}

     

       
256
       
-
       
	defer resp.Body.Close()

     

       
257
       
 
       


     

       
258
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
259
       
 
       
		var errorResp map[string]interface{}

     

       
260
       
-
       
		json.NewDecoder(resp.Body).Decode(&errorResp)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
261
       
 
       
		return "", fmt.Errorf("PDS admin API returned status %d: %v", resp.StatusCode, errorResp)

     

       
262
       
 
       
	}

     

       
263
       
 
       


     
···

       
303
       
 
       
	if err != nil {

     

       
304
       
 
       
		return "", fmt.Errorf("failed to call signup endpoint: %w", err)

     

       
305
       
 
       
	}

     

       
306
       
-
       
	defer resp.Body.Close()

     

       
307
       
 
       


     

       
308
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
309
       
 
       
		var errorResp map[string]interface{}

     

       
310
       
-
       
		json.NewDecoder(resp.Body).Decode(&errorResp)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
311
       
 
       
		return "", fmt.Errorf("signup endpoint returned status %d: %v", resp.StatusCode, errorResp)

     

       
312
       
 
       
	}

     

       
313
       
 
       


     

       
314
       
 
       
	var result struct {

     

       
315
       
-
       
		DID         string `json:"did"`

     

       
316
       
-
       
		Handle      string `json:"handle"`

     

       
317
       
-
       
		AccessJwt   string `json:"accessJwt"`

     

       
318
       
-
       
		RefreshJwt  string `json:"refreshJwt"`

     

       
319
       
 
       
	}

     

       
320
       
 
       


     

       
321
       
 
       
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {

     
···

       
334
       
 
       
		t.Logf("PDS not available: %v", err)

     

       
335
       
 
       
		return false

     

       
336
       
 
       
	}

     

       
337
       
-
       
	defer resp.Body.Close()

     

       
338
       
 
       
	return resp.StatusCode == http.StatusOK

     

       
339
       
 
       
}

     

       
340
       
 
       


     
···

       
346
       
 
       
		t.Logf("Jetstream not available: %v", err)

     

       
347
       
 
       
		return false

     

       
348
       
 
       
	}

     

       
349
       
-
       
	defer resp.Body.Close()

     

       
350
       
 
       
	return resp.StatusCode == http.StatusOK

     

       
351
       
 
       
}

     

       
352
       
 
       


     
···

       
357
       
 
       
		t.Logf("AppView not available: %v", err)

     

       
358
       
 
       
		return false

     

       
359
       
 
       
	}

     

       
360
       
-
       
	defer resp.Body.Close()

     

       
361
       
 
       
	return resp.StatusCode == http.StatusOK

     

       
362
       
 
       
}

     

       
363
       
 
       


     
···

       
391
       
 
       
		t.Fatalf("Failed to connect to test database: %v", err)

     

       
392
       
 
       
	}

     

       
393
       
 
       


     

       
394
       
-
       
	if err := db.Ping(); err != nil {

     

       
395
       
-
       
		t.Fatalf("Failed to ping test database: %v", err)

     

       
396
       
 
       
	}

     

       
397
       
 
       


     

       
398
       
-
       
	if err := goose.SetDialect("postgres"); err != nil {

     

       
399
       
-
       
		t.Fatalf("Failed to set goose dialect: %v", err)

     

       
400
       
 
       
	}

     

       
401
       
 
       


     

       
402
       
-
       
	if err := goose.Up(db, "../../internal/db/migrations"); err != nil {

     

       
403
       
-
       
		t.Fatalf("Failed to run migrations: %v", err)

     

       
404
       
 
       
	}

     

       
405
       
 
       


     

       
406
       
 
       
	// Clean up any existing test data

     

···

       
1
       
 
       
package e2e

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
+
       
	"Coves/internal/atproto/identity"

     

       
5
       
+
       
	"Coves/internal/atproto/jetstream"

     

       
6
       
+
       
	"Coves/internal/core/users"

     

       
7
       
+
       
	"Coves/internal/db/postgres"

     

       
8
       
 
       
	"bytes"

     

       
9
       
 
       
	"context"

     

       
10
       
 
       
	"database/sql"

     
···

       
15
       
 
       
	"testing"

     

       
16
       
 
       
	"time"

     

       
17
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
18
       
 
       
	_ "github.com/lib/pq"

     

       
19
       
 
       
	"github.com/pressly/goose/v3"

     

       
20
       
 
       
)

     
···

       
31
       
 
       
//   - Test database on localhost:5434

     

       
32
       
 
       
//

     

       
33
       
 
       
// Run with:

     

       
34
       
+
       
//

     

       
35
       
+
       
//	make e2e-up  # Start infrastructure

     

       
36
       
+
       
//	go run ./cmd/server &  # Start AppView

     

       
37
       
+
       
//	go test ./tests/integration -run TestE2E_UserSignup -v

     

       
38
       
 
       
func TestE2E_UserSignup(t *testing.T) {

     

       
39
       
 
       
	if testing.Short() {

     

       
40
       
 
       
		t.Skip("Skipping E2E test in short mode")

     
···

       
56
       
 
       
	}

     

       
57
       
 
       


     

       
58
       
 
       
	db := setupTestDB(t)

     

       
59
       
+
       
	defer func() {

     

       
60
       
+
       
		if err := db.Close(); err != nil {

     

       
61
       
+
       
			t.Logf("Failed to close database: %v", err)

     

       
62
       
+
       
		}

     

       
63
       
+
       
	}()

     

       
64
       
 
       


     

       
65
       
 
       
	// Set up services

     

       
66
       
 
       
	userRepo := postgres.NewUserRepository(db)

     
···

       
258
       
 
       
	if err != nil {

     

       
259
       
 
       
		return "", fmt.Errorf("failed to create invite code: %w", err)

     

       
260
       
 
       
	}

     

       
261
       
+
       
	defer func() { _ = resp.Body.Close() }()

     

       
262
       
 
       


     

       
263
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
264
       
 
       
		var errorResp map[string]interface{}

     

       
265
       
+
       
		if err := json.NewDecoder(resp.Body).Decode(&errorResp); err != nil {

     

       
266
       
+
       
			return "", fmt.Errorf("PDS admin API returned status %d (failed to decode error: %w)", resp.StatusCode, err)

     

       
267
       
+
       
		}

     

       
268
       
 
       
		return "", fmt.Errorf("PDS admin API returned status %d: %v", resp.StatusCode, errorResp)

     

       
269
       
 
       
	}

     

       
270
       
 
       


     
···

       
310
       
 
       
	if err != nil {

     

       
311
       
 
       
		return "", fmt.Errorf("failed to call signup endpoint: %w", err)

     

       
312
       
 
       
	}

     

       
313
       
+
       
	defer func() { _ = resp.Body.Close() }()

     

       
314
       
 
       


     

       
315
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
316
       
 
       
		var errorResp map[string]interface{}

     

       
317
       
+
       
		if err := json.NewDecoder(resp.Body).Decode(&errorResp); err != nil {

     

       
318
       
+
       
			return "", fmt.Errorf("signup endpoint returned status %d (failed to decode error: %w)", resp.StatusCode, err)

     

       
319
       
+
       
		}

     

       
320
       
 
       
		return "", fmt.Errorf("signup endpoint returned status %d: %v", resp.StatusCode, errorResp)

     

       
321
       
 
       
	}

     

       
322
       
 
       


     

       
323
       
 
       
	var result struct {

     

       
324
       
+
       
		DID        string `json:"did"`

     

       
325
       
+
       
		Handle     string `json:"handle"`

     

       
326
       
+
       
		AccessJwt  string `json:"accessJwt"`

     

       
327
       
+
       
		RefreshJwt string `json:"refreshJwt"`

     

       
328
       
 
       
	}

     

       
329
       
 
       


     

       
330
       
 
       
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {

     
···

       
343
       
 
       
		t.Logf("PDS not available: %v", err)

     

       
344
       
 
       
		return false

     

       
345
       
 
       
	}

     

       
346
       
+
       
	defer func() { _ = resp.Body.Close() }()

     

       
347
       
 
       
	return resp.StatusCode == http.StatusOK

     

       
348
       
 
       
}

     

       
349
       
 
       


     
···

       
355
       
 
       
		t.Logf("Jetstream not available: %v", err)

     

       
356
       
 
       
		return false

     

       
357
       
 
       
	}

     

       
358
       
+
       
	defer func() { _ = resp.Body.Close() }()

     

       
359
       
 
       
	return resp.StatusCode == http.StatusOK

     

       
360
       
 
       
}

     

       
361
       
 
       


     
···

       
366
       
 
       
		t.Logf("AppView not available: %v", err)

     

       
367
       
 
       
		return false

     

       
368
       
 
       
	}

     

       
369
       
+
       
	defer func() { _ = resp.Body.Close() }()

     

       
370
       
 
       
	return resp.StatusCode == http.StatusOK

     

       
371
       
 
       
}

     

       
372
       
 
       


     
···

       
400
       
 
       
		t.Fatalf("Failed to connect to test database: %v", err)

     

       
401
       
 
       
	}

     

       
402
       
 
       


     

       
403
       
+
       
	if pingErr := db.Ping(); pingErr != nil {

     

       
404
       
+
       
		t.Fatalf("Failed to ping test database: %v", pingErr)

     

       
405
       
 
       
	}

     

       
406
       
 
       


     

       
407
       
+
       
	if dialectErr := goose.SetDialect("postgres"); dialectErr != nil {

     

       
408
       
+
       
		t.Fatalf("Failed to set goose dialect: %v", dialectErr)

     

       
409
       
 
       
	}

     

       
410
       
 
       


     

       
411
       
+
       
	if migrateErr := goose.Up(db, "../../internal/db/migrations"); migrateErr != nil {

     

       
412
       
+
       
		t.Fatalf("Failed to run migrations: %v", migrateErr)

     

       
413
       
 
       
	}

     

       
414
       
 
       


     

       
415
       
 
       
	// Clean up any existing test data

     

···

       
1
       
 
       
package integration

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
4
       
 
       
	"context"

     

       
5
       
 
       
	"fmt"

     

       
6
       
 
       
	"testing"

     

       
7
       
 
       
	"time"

     

       
8
       
-
       


     

       
9
       
-
       
	"Coves/internal/atproto/did"

     

       
10
       
-
       
	"Coves/internal/atproto/jetstream"

     

       
11
       
-
       
	"Coves/internal/core/communities"

     

       
12
       
-
       
	"Coves/internal/db/postgres"

     

       
13
       
 
       
)

     

       
14
       
 
       


     

       
15
       
 
       
func TestCommunityConsumer_HandleCommunityProfile(t *testing.T) {

     

       
16
       
 
       
	db := setupTestDB(t)

     

       
17
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
18
       
 
       


     

       
19
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
20
       
 
       
	consumer := jetstream.NewCommunityEventConsumer(repo)

     
···

       
22
       
 
       
	ctx := context.Background()

     

       
23
       
 
       


     

       
24
       
 
       
	t.Run("creates community from firehose event", func(t *testing.T) {

     

       
25
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
26
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
27
       
 
       


     

       
28
       
 
       
		// Simulate a Jetstream commit event

     
···

       
57
       
 
       
		}

     

       
58
       
 
       


     

       
59
       
 
       
		// Handle the event

     

       
60
       
-
       
		err := consumer.HandleEvent(ctx, event)

     

       
61
       
-
       
		if err != nil {

     

       
62
       
 
       
			t.Fatalf("Failed to handle event: %v", err)

     

       
63
       
 
       
		}

     

       
64
       
 
       


     
···

       
80
       
 
       
	})

     

       
81
       
 
       


     

       
82
       
 
       
	t.Run("updates existing community", func(t *testing.T) {

     

       
83
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
84
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
85
       
 
       
		handle := fmt.Sprintf("!update-test-%s@coves.local", uniqueSuffix)

     

       
86
       
 
       


     
···

       
100
       
 
       
			UpdatedAt:              time.Now(),

     

       
101
       
 
       
		}

     

       
102
       
 
       


     

       
103
       
-
       
		_, err := repo.Create(ctx, initialCommunity)

     

       
104
       
-
       
		if err != nil {

     

       
105
       
 
       
			t.Fatalf("Failed to create initial community: %v", err)

     

       
106
       
 
       
		}

     

       
107
       
 
       


     
···

       
137
       
 
       
		}

     

       
138
       
 
       


     

       
139
       
 
       
		// Handle the update

     

       
140
       
-
       
		err = consumer.HandleEvent(ctx, updateEvent)

     

       
141
       
-
       
		if err != nil {

     

       
142
       
 
       
			t.Fatalf("Failed to handle update event: %v", err)

     

       
143
       
 
       
		}

     

       
144
       
 
       


     
···

       
163
       
 
       
	})

     

       
164
       
 
       


     

       
165
       
 
       
	t.Run("deletes community", func(t *testing.T) {

     

       
166
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
167
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
168
       
 
       


     

       
169
       
 
       
		// Create community to delete

     
···

       
179
       
 
       
			UpdatedAt:    time.Now(),

     

       
180
       
 
       
		}

     

       
181
       
 
       


     

       
182
       
-
       
		_, err := repo.Create(ctx, community)

     

       
183
       
-
       
		if err != nil {

     

       
184
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
185
       
 
       
		}

     

       
186
       
 
       


     
···

       
198
       
 
       
		}

     

       
199
       
 
       


     

       
200
       
 
       
		// Handle the delete

     

       
201
       
-
       
		err = consumer.HandleEvent(ctx, deleteEvent)

     

       
202
       
-
       
		if err != nil {

     

       
203
       
 
       
			t.Fatalf("Failed to handle delete event: %v", err)

     

       
204
       
 
       
		}

     

       
205
       
 
       


     

       
206
       
 
       
		// Verify community was deleted

     

       
207
       
-
       
		_, err = repo.GetByDID(ctx, communityDID)

     

       
208
       
-
       
		if err != communities.ErrCommunityNotFound {

     

       
209
       
 
       
			t.Errorf("Expected ErrCommunityNotFound, got: %v", err)

     

       
210
       
 
       
		}

     

       
211
       
 
       
	})

     
···

       
213
       
 
       


     

       
214
       
 
       
func TestCommunityConsumer_HandleSubscription(t *testing.T) {

     

       
215
       
 
       
	db := setupTestDB(t)

     

       
216
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
217
       
 
       


     

       
218
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
219
       
 
       
	consumer := jetstream.NewCommunityEventConsumer(repo)

     
···

       
222
       
 
       


     

       
223
       
 
       
	t.Run("creates subscription from event", func(t *testing.T) {

     

       
224
       
 
       
		// Create a community first

     

       
225
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
226
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
227
       
 
       


     

       
228
       
 
       
		community := &communities.Community{

     
···

       
237
       
 
       
			UpdatedAt:    time.Now(),

     

       
238
       
 
       
		}

     

       
239
       
 
       


     

       
240
       
-
       
		_, err := repo.Create(ctx, community)

     

       
241
       
-
       
		if err != nil {

     

       
242
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
243
       
 
       
		}

     

       
244
       
 
       


     
···

       
261
       
 
       
		}

     

       
262
       
 
       


     

       
263
       
 
       
		// Handle the subscription

     

       
264
       
-
       
		err = consumer.HandleEvent(ctx, subEvent)

     

       
265
       
-
       
		if err != nil {

     

       
266
       
 
       
			t.Fatalf("Failed to handle subscription event: %v", err)

     

       
267
       
 
       
		}

     

       
268
       
 
       


     
···

       
293
       
 
       


     

       
294
       
 
       
func TestCommunityConsumer_IgnoresNonCommunityEvents(t *testing.T) {

     

       
295
       
 
       
	db := setupTestDB(t)

     

       
296
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
297
       
 
       


     

       
298
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
299
       
 
       
	consumer := jetstream.NewCommunityEventConsumer(repo)

     

···

       
1
       
 
       
package integration

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
+
       
	"Coves/internal/atproto/did"

     

       
5
       
+
       
	"Coves/internal/atproto/jetstream"

     

       
6
       
+
       
	"Coves/internal/core/communities"

     

       
7
       
+
       
	"Coves/internal/db/postgres"

     

       
8
       
 
       
	"context"

     

       
9
       
 
       
	"fmt"

     

       
10
       
 
       
	"testing"

     

       
11
       
 
       
	"time"

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
12
       
 
       
)

     

       
13
       
 
       


     

       
14
       
 
       
func TestCommunityConsumer_HandleCommunityProfile(t *testing.T) {

     

       
15
       
 
       
	db := setupTestDB(t)

     

       
16
       
+
       
	defer func() {

     

       
17
       
+
       
		if err := db.Close(); err != nil {

     

       
18
       
+
       
			t.Logf("Failed to close database: %v", err)

     

       
19
       
+
       
		}

     

       
20
       
+
       
	}()

     

       
21
       
 
       


     

       
22
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
23
       
 
       
	consumer := jetstream.NewCommunityEventConsumer(repo)

     
···

       
25
       
 
       
	ctx := context.Background()

     

       
26
       
 
       


     

       
27
       
 
       
	t.Run("creates community from firehose event", func(t *testing.T) {

     

       
28
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
29
       
+
       
		if err != nil {

     

       
30
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
31
       
+
       
		}

     

       
32
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
33
       
 
       


     

       
34
       
 
       
		// Simulate a Jetstream commit event

     
···

       
63
       
 
       
		}

     

       
64
       
 
       


     

       
65
       
 
       
		// Handle the event

     

       
66
       
+
       
		if err := consumer.HandleEvent(ctx, event); err != nil {

     

       
0
       
 
       

     

       
67
       
 
       
			t.Fatalf("Failed to handle event: %v", err)

     

       
68
       
 
       
		}

     

       
69
       
 
       


     
···

       
85
       
 
       
	})

     

       
86
       
 
       


     

       
87
       
 
       
	t.Run("updates existing community", func(t *testing.T) {

     

       
88
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
89
       
+
       
		if err != nil {

     

       
90
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
91
       
+
       
		}

     

       
92
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
93
       
 
       
		handle := fmt.Sprintf("!update-test-%s@coves.local", uniqueSuffix)

     

       
94
       
 
       


     
···

       
108
       
 
       
			UpdatedAt:              time.Now(),

     

       
109
       
 
       
		}

     

       
110
       
 
       


     

       
111
       
+
       
		if _, err := repo.Create(ctx, initialCommunity); err != nil {

     

       
0
       
 
       

     

       
112
       
 
       
			t.Fatalf("Failed to create initial community: %v", err)

     

       
113
       
 
       
		}

     

       
114
       
 
       


     
···

       
144
       
 
       
		}

     

       
145
       
 
       


     

       
146
       
 
       
		// Handle the update

     

       
147
       
+
       
		if err := consumer.HandleEvent(ctx, updateEvent); err != nil {

     

       
0
       
 
       

     

       
148
       
 
       
			t.Fatalf("Failed to handle update event: %v", err)

     

       
149
       
 
       
		}

     

       
150
       
 
       


     
···

       
169
       
 
       
	})

     

       
170
       
 
       


     

       
171
       
 
       
	t.Run("deletes community", func(t *testing.T) {

     

       
172
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
173
       
+
       
		if err != nil {

     

       
174
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
175
       
+
       
		}

     

       
176
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
177
       
 
       


     

       
178
       
 
       
		// Create community to delete

     
···

       
188
       
 
       
			UpdatedAt:    time.Now(),

     

       
189
       
 
       
		}

     

       
190
       
 
       


     

       
191
       
+
       
		if _, err := repo.Create(ctx, community); err != nil {

     

       
0
       
 
       

     

       
192
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
193
       
 
       
		}

     

       
194
       
 
       


     
···

       
206
       
 
       
		}

     

       
207
       
 
       


     

       
208
       
 
       
		// Handle the delete

     

       
209
       
+
       
		if err := consumer.HandleEvent(ctx, deleteEvent); err != nil {

     

       
0
       
 
       

     

       
210
       
 
       
			t.Fatalf("Failed to handle delete event: %v", err)

     

       
211
       
 
       
		}

     

       
212
       
 
       


     

       
213
       
 
       
		// Verify community was deleted

     

       
214
       
+
       
		if _, err := repo.GetByDID(ctx, communityDID); err != communities.ErrCommunityNotFound {

     

       
0
       
 
       

     

       
215
       
 
       
			t.Errorf("Expected ErrCommunityNotFound, got: %v", err)

     

       
216
       
 
       
		}

     

       
217
       
 
       
	})

     
···

       
219
       
 
       


     

       
220
       
 
       
func TestCommunityConsumer_HandleSubscription(t *testing.T) {

     

       
221
       
 
       
	db := setupTestDB(t)

     

       
222
       
+
       
	defer func() {

     

       
223
       
+
       
		if err := db.Close(); err != nil {

     

       
224
       
+
       
			t.Logf("Failed to close database: %v", err)

     

       
225
       
+
       
		}

     

       
226
       
+
       
	}()

     

       
227
       
 
       


     

       
228
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
229
       
 
       
	consumer := jetstream.NewCommunityEventConsumer(repo)

     
···

       
232
       
 
       


     

       
233
       
 
       
	t.Run("creates subscription from event", func(t *testing.T) {

     

       
234
       
 
       
		// Create a community first

     

       
235
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
236
       
+
       
		if err != nil {

     

       
237
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
238
       
+
       
		}

     

       
239
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
240
       
 
       


     

       
241
       
 
       
		community := &communities.Community{

     
···

       
250
       
 
       
			UpdatedAt:    time.Now(),

     

       
251
       
 
       
		}

     

       
252
       
 
       


     

       
253
       
+
       
		if _, err := repo.Create(ctx, community); err != nil {

     

       
0
       
 
       

     

       
254
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
255
       
 
       
		}

     

       
256
       
 
       


     
···

       
273
       
 
       
		}

     

       
274
       
 
       


     

       
275
       
 
       
		// Handle the subscription

     

       
276
       
+
       
		if err := consumer.HandleEvent(ctx, subEvent); err != nil {

     

       
0
       
 
       

     

       
277
       
 
       
			t.Fatalf("Failed to handle subscription event: %v", err)

     

       
278
       
 
       
		}

     

       
279
       
 
       


     
···

       
304
       
 
       


     

       
305
       
 
       
func TestCommunityConsumer_IgnoresNonCommunityEvents(t *testing.T) {

     

       
306
       
 
       
	db := setupTestDB(t)

     

       
307
       
+
       
	defer func() {

     

       
308
       
+
       
		if err := db.Close(); err != nil {

     

       
309
       
+
       
			t.Logf("Failed to close database: %v", err)

     

       
310
       
+
       
		}

     

       
311
       
+
       
	}()

     

       
312
       
 
       


     

       
313
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
314
       
 
       
	consumer := jetstream.NewCommunityEventConsumer(repo)

     

···

       
1
       
 
       
package integration

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
4
       
 
       
	"context"

     

       
5
       
 
       
	"fmt"

     

       
6
       
 
       
	"testing"

     

       
7
       
 
       
	"time"

     

       
8
       
-
       


     

       
9
       
-
       
	"Coves/internal/atproto/did"

     

       
10
       
-
       
	"Coves/internal/core/communities"

     

       
11
       
-
       
	"Coves/internal/db/postgres"

     

       
12
       
 
       
)

     

       
13
       
 
       


     

       
14
       
 
       
// TestCommunityRepository_CredentialPersistence tests that PDS credentials are properly persisted

     

       
15
       
 
       
func TestCommunityRepository_CredentialPersistence(t *testing.T) {

     

       
16
       
 
       
	db := setupTestDB(t)

     

       
17
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
18
       
 
       


     

       
19
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
20
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
21
       
 
       
	ctx := context.Background()

     

       
22
       
 
       


     

       
23
       
 
       
	t.Run("persists PDS credentials on create", func(t *testing.T) {

     

       
24
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
25
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
26
       
 
       


     

       
27
       
 
       
		community := &communities.Community{

     
···

       
75
       
 
       
	})

     

       
76
       
 
       


     

       
77
       
 
       
	t.Run("handles empty credentials gracefully", func(t *testing.T) {

     

       
78
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
79
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
80
       
 
       


     

       
81
       
 
       
		// Community without PDS credentials (e.g., from Jetstream consumer)

     
···

       
122
       
 
       
// TestCommunityRepository_EncryptedCredentials tests encryption at rest

     

       
123
       
 
       
func TestCommunityRepository_EncryptedCredentials(t *testing.T) {

     

       
124
       
 
       
	db := setupTestDB(t)

     

       
125
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
126
       
 
       


     

       
127
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
128
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
129
       
 
       
	ctx := context.Background()

     

       
130
       
 
       


     

       
131
       
 
       
	t.Run("credentials are encrypted in database", func(t *testing.T) {

     

       
132
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
133
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
134
       
 
       


     

       
135
       
 
       
		accessToken := "sensitive_access_token_xyz123"

     
···

       
152
       
 
       
			UpdatedAt:       time.Now(),

     

       
153
       
 
       
		}

     

       
154
       
 
       


     

       
155
       
-
       
		_, err := repo.Create(ctx, community)

     

       
156
       
-
       
		if err != nil {

     

       
157
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
158
       
 
       
		}

     

       
159
       
 
       


     
···

       
164
       
 
       
			FROM communities

     

       
165
       
 
       
			WHERE did = $1

     

       
166
       
 
       
		`

     

       
167
       
-
       
		err = db.QueryRowContext(ctx, query, communityDID).Scan(&encryptedAccess, &encryptedRefresh)

     

       
168
       
-
       
		if err != nil {

     

       
169
       
 
       
			t.Fatalf("Failed to query encrypted data: %v", err)

     

       
170
       
 
       
		}

     

       
171
       
 
       


     
···

       
200
       
 
       
	})

     

       
201
       
 
       


     

       
202
       
 
       
	t.Run("encryption handles special characters", func(t *testing.T) {

     

       
203
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
204
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
205
       
 
       


     

       
206
       
 
       
		// Token with special characters

     
···

       
220
       
 
       
			UpdatedAt:       time.Now(),

     

       
221
       
 
       
		}

     

       
222
       
 
       


     

       
223
       
-
       
		_, err := repo.Create(ctx, community)

     

       
224
       
-
       
		if err != nil {

     

       
225
       
 
       
			t.Fatalf("Failed to create community with special chars: %v", err)

     

       
226
       
 
       
		}

     

       
227
       
 
       


     
···

       
239
       
 
       
// TestCommunityRepository_V2OwnershipModel tests that communities are self-owned

     

       
240
       
 
       
func TestCommunityRepository_V2OwnershipModel(t *testing.T) {

     

       
241
       
 
       
	db := setupTestDB(t)

     

       
242
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
243
       
 
       


     

       
244
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
245
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
246
       
 
       
	ctx := context.Background()

     

       
247
       
 
       


     

       
248
       
 
       
	t.Run("V2 communities are self-owned", func(t *testing.T) {

     

       
249
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
250
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
251
       
 
       


     

       
252
       
 
       
		community := &communities.Community{

     

···

       
1
       
 
       
package integration

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
+
       
	"Coves/internal/atproto/did"

     

       
5
       
+
       
	"Coves/internal/core/communities"

     

       
6
       
+
       
	"Coves/internal/db/postgres"

     

       
7
       
 
       
	"context"

     

       
8
       
 
       
	"fmt"

     

       
9
       
 
       
	"testing"

     

       
10
       
 
       
	"time"

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
11
       
 
       
)

     

       
12
       
 
       


     

       
13
       
 
       
// TestCommunityRepository_CredentialPersistence tests that PDS credentials are properly persisted

     

       
14
       
 
       
func TestCommunityRepository_CredentialPersistence(t *testing.T) {

     

       
15
       
 
       
	db := setupTestDB(t)

     

       
16
       
+
       
	defer func() {

     

       
17
       
+
       
		if err := db.Close(); err != nil {

     

       
18
       
+
       
			t.Logf("Failed to close database: %v", err)

     

       
19
       
+
       
		}

     

       
20
       
+
       
	}()

     

       
21
       
 
       


     

       
22
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
23
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
24
       
 
       
	ctx := context.Background()

     

       
25
       
 
       


     

       
26
       
 
       
	t.Run("persists PDS credentials on create", func(t *testing.T) {

     

       
27
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
28
       
+
       
		if err != nil {

     

       
29
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
30
       
+
       
		}

     

       
31
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
32
       
 
       


     

       
33
       
 
       
		community := &communities.Community{

     
···

       
81
       
 
       
	})

     

       
82
       
 
       


     

       
83
       
 
       
	t.Run("handles empty credentials gracefully", func(t *testing.T) {

     

       
84
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
85
       
+
       
		if err != nil {

     

       
86
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
87
       
+
       
		}

     

       
88
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
89
       
 
       


     

       
90
       
 
       
		// Community without PDS credentials (e.g., from Jetstream consumer)

     
···

       
131
       
 
       
// TestCommunityRepository_EncryptedCredentials tests encryption at rest

     

       
132
       
 
       
func TestCommunityRepository_EncryptedCredentials(t *testing.T) {

     

       
133
       
 
       
	db := setupTestDB(t)

     

       
134
       
+
       
	defer func() {

     

       
135
       
+
       
		if err := db.Close(); err != nil {

     

       
136
       
+
       
			t.Logf("Failed to close database: %v", err)

     

       
137
       
+
       
		}

     

       
138
       
+
       
	}()

     

       
139
       
 
       


     

       
140
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
141
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
142
       
 
       
	ctx := context.Background()

     

       
143
       
 
       


     

       
144
       
 
       
	t.Run("credentials are encrypted in database", func(t *testing.T) {

     

       
145
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
146
       
+
       
		if err != nil {

     

       
147
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
148
       
+
       
		}

     

       
149
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
150
       
 
       


     

       
151
       
 
       
		accessToken := "sensitive_access_token_xyz123"

     
···

       
168
       
 
       
			UpdatedAt:       time.Now(),

     

       
169
       
 
       
		}

     

       
170
       
 
       


     

       
171
       
+
       
		if _, err := repo.Create(ctx, community); err != nil {

     

       
0
       
 
       

     

       
172
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
173
       
 
       
		}

     

       
174
       
 
       


     
···

       
179
       
 
       
			FROM communities

     

       
180
       
 
       
			WHERE did = $1

     

       
181
       
 
       
		`

     

       
182
       
+
       
		if err := db.QueryRowContext(ctx, query, communityDID).Scan(&encryptedAccess, &encryptedRefresh); err != nil {

     

       
0
       
 
       

     

       
183
       
 
       
			t.Fatalf("Failed to query encrypted data: %v", err)

     

       
184
       
 
       
		}

     

       
185
       
 
       


     
···

       
214
       
 
       
	})

     

       
215
       
 
       


     

       
216
       
 
       
	t.Run("encryption handles special characters", func(t *testing.T) {

     

       
217
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
218
       
+
       
		if err != nil {

     

       
219
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
220
       
+
       
		}

     

       
221
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
222
       
 
       


     

       
223
       
 
       
		// Token with special characters

     
···

       
237
       
 
       
			UpdatedAt:       time.Now(),

     

       
238
       
 
       
		}

     

       
239
       
 
       


     

       
240
       
+
       
		if _, err := repo.Create(ctx, community); err != nil {

     

       
0
       
 
       

     

       
241
       
 
       
			t.Fatalf("Failed to create community with special chars: %v", err)

     

       
242
       
 
       
		}

     

       
243
       
 
       


     
···

       
255
       
 
       
// TestCommunityRepository_V2OwnershipModel tests that communities are self-owned

     

       
256
       
 
       
func TestCommunityRepository_V2OwnershipModel(t *testing.T) {

     

       
257
       
 
       
	db := setupTestDB(t)

     

       
258
       
+
       
	defer func() {

     

       
259
       
+
       
		if err := db.Close(); err != nil {

     

       
260
       
+
       
			t.Logf("Failed to close database: %v", err)

     

       
261
       
+
       
		}

     

       
262
       
+
       
	}()

     

       
263
       
 
       


     

       
264
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
265
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
266
       
 
       
	ctx := context.Background()

     

       
267
       
 
       


     

       
268
       
 
       
	t.Run("V2 communities are self-owned", func(t *testing.T) {

     

       
269
       
+
       
		communityDID, err := didGen.GenerateCommunityDID()

     

       
270
       
+
       
		if err != nil {

     

       
271
       
+
       
			t.Fatalf("Failed to generate community DID: %v", err)

     

       
272
       
+
       
		}

     

       
273
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
274
       
 
       


     

       
275
       
 
       
		community := &communities.Community{

     

···

       
1
       
 
       
package integration

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
4
       
 
       
	"bytes"

     

       
5
       
 
       
	"context"

     

       
6
       
 
       
	"database/sql"

     
···

       
19
       
 
       
	"github.com/gorilla/websocket"

     

       
20
       
 
       
	_ "github.com/lib/pq"

     

       
21
       
 
       
	"github.com/pressly/goose/v3"

     

       
22
       
-
       


     

       
23
       
-
       
	"Coves/internal/api/routes"

     

       
24
       
-
       
	"Coves/internal/atproto/did"

     

       
25
       
-
       
	"Coves/internal/atproto/identity"

     

       
26
       
-
       
	"Coves/internal/atproto/jetstream"

     

       
27
       
-
       
	"Coves/internal/core/communities"

     

       
28
       
-
       
	"Coves/internal/core/users"

     

       
29
       
-
       
	"Coves/internal/db/postgres"

     

       
30
       
 
       
)

     

       
31
       
 
       


     

       
32
       
 
       
// TestCommunity_E2E is a TRUE end-to-end test covering the complete flow:

     
···

       
55
       
 
       
	if err != nil {

     

       
56
       
 
       
		t.Fatalf("Failed to connect to test database: %v", err)

     

       
57
       
 
       
	}

     

       
58
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
59
       
 
       


     

       
60
       
 
       
	// Run migrations

     

       
61
       
-
       
	if err := goose.SetDialect("postgres"); err != nil {

     

       
62
       
-
       
		t.Fatalf("Failed to set goose dialect: %v", err)

     

       
63
       
 
       
	}

     

       
64
       
-
       
	if err := goose.Up(db, "../../internal/db/migrations"); err != nil {

     

       
65
       
-
       
		t.Fatalf("Failed to run migrations: %v", err)

     

       
66
       
 
       
	}

     

       
67
       
 
       


     

       
68
       
 
       
	// Check if PDS is running

     
···

       
75
       
 
       
	if err != nil {

     

       
76
       
 
       
		t.Skipf("PDS not running at %s: %v", pdsURL, err)

     

       
77
       
 
       
	}

     

       
78
       
-
       
	healthResp.Body.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
79
       
 
       


     

       
80
       
 
       
	// Setup dependencies

     

       
81
       
 
       
	communityRepo := postgres.NewCommunityRepository(db)

     
···

       
208
       
 
       
		if err != nil {

     

       
209
       
 
       
			t.Fatalf("Failed to query PDS: %v", err)

     

       
210
       
 
       
		}

     

       
211
       
-
       
		defer pdsResp.Body.Close()

     

       
212
       
 
       


     

       
213
       
 
       
		if pdsResp.StatusCode != http.StatusOK {

     

       
214
       
-
       
			body, _ := io.ReadAll(pdsResp.Body)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
215
       
 
       
			t.Fatalf("PDS returned status %d: %s", pdsResp.StatusCode, string(body))

     

       
216
       
 
       
		}

     

       
217
       
 
       


     

       
218
       
 
       
		var pdsRecord struct {

     

       
0
       
 
       

     

       
219
       
 
       
			URI   string                 `json:"uri"`

     

       
220
       
 
       
			CID   string                 `json:"cid"`

     

       
221
       
-
       
			Value map[string]interface{} `json:"value"`

     

       
222
       
 
       
		}

     

       
223
       
 
       


     

       
224
       
 
       
		if err := json.NewDecoder(pdsResp.Body).Decode(&pdsRecord); err != nil {

     
···

       
230
       
 
       
		t.Logf("   CID: %s", pdsRecord.CID)

     

       
231
       
 
       


     

       
232
       
 
       
		// Print full record for inspection

     

       
233
       
-
       
		recordJSON, _ := json.MarshalIndent(pdsRecord.Value, "   ", "  ")

     

       
234
       
-
       
		t.Logf("   Record value:\n   %s", string(recordJSON))

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
235
       
 
       


     

       
236
       
 
       
		// V2: DID is NOT in the record - it's in the repository URI

     

       
237
       
 
       
		// The record should have handle, name, etc. but no 'did' field

     
···

       
330
       
 
       
	// Part 3: XRPC HTTP Endpoints

     

       
331
       
 
       
	// ====================================================================================

     

       
332
       
 
       
	t.Run("3. XRPC HTTP Endpoints", func(t *testing.T) {

     

       
333
       
-
       


     

       
334
       
 
       
		t.Run("Create via XRPC endpoint", func(t *testing.T) {

     

       
335
       
 
       
			// Use Unix timestamp (seconds) instead of UnixNano to keep handle short

     

       
336
       
 
       
			createReq := map[string]interface{}{

     
···

       
343
       
 
       
				"allowExternalDiscovery": true,

     

       
344
       
 
       
			}

     

       
345
       
 
       


     

       
346
       
-
       
			reqBody, _ := json.Marshal(createReq)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
347
       
 
       


     

       
348
       
 
       
			// Step 1: Client POSTs to XRPC endpoint

     

       
349
       
 
       
			t.Logf("📡 Client → POST /xrpc/social.coves.community.create")

     
···

       
356
       
 
       
			if err != nil {

     

       
357
       
 
       
				t.Fatalf("Failed to POST: %v", err)

     

       
358
       
 
       
			}

     

       
359
       
-
       
			defer resp.Body.Close()

     

       
360
       
 
       


     

       
361
       
 
       
			if resp.StatusCode != http.StatusOK {

     

       
362
       
-
       
				body, _ := io.ReadAll(resp.Body)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
363
       
 
       
				t.Logf("❌ XRPC Create Failed")

     

       
364
       
 
       
				t.Logf("   Status: %d", resp.StatusCode)

     

       
365
       
 
       
				t.Logf("   Response: %s", string(body))

     
···

       
373
       
 
       
				Handle string `json:"handle"`

     

       
374
       
 
       
			}

     

       
375
       
 
       


     

       
376
       
-
       
			json.NewDecoder(resp.Body).Decode(&createResp)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
377
       
 
       


     

       
378
       
 
       
			t.Logf("✅ XRPC response received:")

     

       
379
       
 
       
			t.Logf("   DID:    %s", createResp.DID)

     
···

       
409
       
 
       
					CID: createResp.CID,

     

       
410
       
 
       
				},

     

       
411
       
 
       
			}

     

       
412
       
-
       
			consumer.HandleEvent(context.Background(), &event)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
413
       
 
       


     

       
414
       
 
       
			// Step 3: Verify it's indexed in AppView

     

       
415
       
 
       
			t.Logf("🔍 Querying AppView to verify indexing...")

     
···

       
443
       
 
       
			if err != nil {

     

       
444
       
 
       
				t.Fatalf("Failed to GET: %v", err)

     

       
445
       
 
       
			}

     

       
446
       
-
       
			defer resp.Body.Close()

     

       
447
       
 
       


     

       
448
       
 
       
			if resp.StatusCode != http.StatusOK {

     

       
449
       
-
       
				body, _ := io.ReadAll(resp.Body)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
450
       
 
       
				t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))

     

       
451
       
 
       
			}

     

       
452
       
 
       


     

       
453
       
 
       
			var getCommunity communities.Community

     

       
454
       
-
       
			json.NewDecoder(resp.Body).Decode(&getCommunity)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
455
       
 
       


     

       
456
       
-
       
			t.Logf("✅ Retrieved via XRPC HTTP endpoint:")

     

       
457
       
 
       
			t.Logf("   DID:         %s", getCommunity.DID)

     

       
458
       
 
       
			t.Logf("   DisplayName: %s", getCommunity.DisplayName)

     

       
459
       
 
       


     
···

       
473
       
 
       
			if err != nil {

     

       
474
       
 
       
				t.Fatalf("Failed to GET list: %v", err)

     

       
475
       
 
       
			}

     

       
476
       
-
       
			defer resp.Body.Close()

     

       
477
       
 
       


     

       
478
       
 
       
			if resp.StatusCode != http.StatusOK {

     

       
479
       
-
       
				body, _ := io.ReadAll(resp.Body)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
480
       
 
       
				t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))

     

       
481
       
 
       
			}

     

       
482
       
 
       


     
···

       
485
       
 
       
				Total       int                     `json:"total"`

     

       
486
       
 
       
			}

     

       
487
       
 
       


     

       
488
       
-
       
			json.NewDecoder(resp.Body).Decode(&listResp)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
489
       
 
       


     

       
490
       
 
       
			t.Logf("✅ Listed %d communities via XRPC", len(listResp.Communities))

     

       
491
       
 
       


     
···

       
547
       
 
       
	collection := "social.coves.community.profile"

     

       
548
       
 
       
	rkey := extractRKeyFromURI(community.RecordURI)

     

       
549
       
 
       


     

       
550
       
-
       
	pdsResp, _ := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",

     

       
551
       
 
       
		pdsURL, instanceDID, collection, rkey))

     

       
552
       
-
       
	defer pdsResp.Body.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
553
       
 
       


     

       
554
       
 
       
	var pdsRecord struct {

     

       
555
       
-
       
		CID   string                 `json:"cid"`

     

       
556
       
 
       
		Value map[string]interface{} `json:"value"`

     

       
0
       
 
       

     

       
557
       
 
       
	}

     

       
558
       
-
       
	json.NewDecoder(pdsResp.Body).Decode(&pdsRecord)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
559
       
 
       


     

       
560
       
 
       
	// Simulate firehose event

     

       
561
       
 
       
	event := jetstream.JetstreamEvent{

     
···

       
572
       
 
       
		},

     

       
573
       
 
       
	}

     

       
574
       
 
       


     

       
575
       
-
       
	consumer.HandleEvent(context.Background(), &event)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
576
       
 
       


     

       
577
       
 
       
	return community

     

       
578
       
 
       
}

     
···

       
594
       
 
       
		"password":   password,

     

       
595
       
 
       
	}

     

       
596
       
 
       


     

       
597
       
-
       
	reqBody, _ := json.Marshal(sessionReq)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
598
       
 
       
	resp, err := http.Post(

     

       
599
       
 
       
		pdsURL+"/xrpc/com.atproto.server.createSession",

     

       
600
       
 
       
		"application/json",

     
···

       
603
       
 
       
	if err != nil {

     

       
604
       
 
       
		return "", "", fmt.Errorf("failed to create session: %w", err)

     

       
605
       
 
       
	}

     

       
606
       
-
       
	defer resp.Body.Close()

     

       
607
       
 
       


     

       
608
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
609
       
-
       
		body, _ := io.ReadAll(resp.Body)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
610
       
 
       
		return "", "", fmt.Errorf("PDS auth failed (status %d): %s", resp.StatusCode, string(body))

     

       
611
       
 
       
	}

     

       
612
       
 
       


     
···

       
665
       
 
       
	if err != nil {

     

       
666
       
 
       
		return "", "", fmt.Errorf("failed to query PDS: %w", err)

     

       
667
       
 
       
	}

     

       
668
       
-
       
	defer resp.Body.Close()

     

       
669
       
 
       


     

       
670
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
671
       
-
       
		body, _ := io.ReadAll(resp.Body)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
672
       
 
       
		return "", "", fmt.Errorf("account not found (status %d): %s", resp.StatusCode, string(body))

     

       
673
       
 
       
	}

     

       
674
       
 
       


     
···

       
700
       
 
       
	if err != nil {

     

       
701
       
 
       
		return fmt.Errorf("failed to connect to Jetstream: %w", err)

     

       
702
       
 
       
	}

     

       
703
       
-
       
	defer conn.Close()

     

       
704
       
 
       


     

       
705
       
 
       
	// Read messages until we find our event or receive done signal

     

       
706
       
 
       
	for {

     
···

       
711
       
 
       
			return ctx.Err()

     

       
712
       
 
       
		default:

     

       
713
       
 
       
			// Set read deadline to avoid blocking forever

     

       
714
       
-
       
			conn.SetReadDeadline(time.Now().Add(5 * time.Second))

     

       
0
       
 
       

     

       
0
       
 
       

     

       
715
       
 
       


     

       
716
       
 
       
			var event jetstream.JetstreamEvent

     

       
717
       
 
       
			err := conn.ReadJSON(&event)

     

···

       
1
       
 
       
package integration

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
+
       
	"Coves/internal/api/routes"

     

       
5
       
+
       
	"Coves/internal/atproto/did"

     

       
6
       
+
       
	"Coves/internal/atproto/identity"

     

       
7
       
+
       
	"Coves/internal/atproto/jetstream"

     

       
8
       
+
       
	"Coves/internal/core/communities"

     

       
9
       
+
       
	"Coves/internal/core/users"

     

       
10
       
+
       
	"Coves/internal/db/postgres"

     

       
11
       
 
       
	"bytes"

     

       
12
       
 
       
	"context"

     

       
13
       
 
       
	"database/sql"

     
···

       
26
       
 
       
	"github.com/gorilla/websocket"

     

       
27
       
 
       
	_ "github.com/lib/pq"

     

       
28
       
 
       
	"github.com/pressly/goose/v3"

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
29
       
 
       
)

     

       
30
       
 
       


     

       
31
       
 
       
// TestCommunity_E2E is a TRUE end-to-end test covering the complete flow:

     
···

       
54
       
 
       
	if err != nil {

     

       
55
       
 
       
		t.Fatalf("Failed to connect to test database: %v", err)

     

       
56
       
 
       
	}

     

       
57
       
+
       
	defer func() {

     

       
58
       
+
       
		if closeErr := db.Close(); closeErr != nil {

     

       
59
       
+
       
			t.Logf("Failed to close database: %v", closeErr)

     

       
60
       
+
       
		}

     

       
61
       
+
       
	}()

     

       
62
       
 
       


     

       
63
       
 
       
	// Run migrations

     

       
64
       
+
       
	if dialectErr := goose.SetDialect("postgres"); dialectErr != nil {

     

       
65
       
+
       
		t.Fatalf("Failed to set goose dialect: %v", dialectErr)

     

       
66
       
 
       
	}

     

       
67
       
+
       
	if migrateErr := goose.Up(db, "../../internal/db/migrations"); migrateErr != nil {

     

       
68
       
+
       
		t.Fatalf("Failed to run migrations: %v", migrateErr)

     

       
69
       
 
       
	}

     

       
70
       
 
       


     

       
71
       
 
       
	// Check if PDS is running

     
···

       
78
       
 
       
	if err != nil {

     

       
79
       
 
       
		t.Skipf("PDS not running at %s: %v", pdsURL, err)

     

       
80
       
 
       
	}

     

       
81
       
+
       
	func() {

     

       
82
       
+
       
		if closeErr := healthResp.Body.Close(); closeErr != nil {

     

       
83
       
+
       
			t.Logf("Failed to close health response: %v", closeErr)

     

       
84
       
+
       
		}

     

       
85
       
+
       
	}()

     

       
86
       
 
       


     

       
87
       
 
       
	// Setup dependencies

     

       
88
       
 
       
	communityRepo := postgres.NewCommunityRepository(db)

     
···

       
215
       
 
       
		if err != nil {

     

       
216
       
 
       
			t.Fatalf("Failed to query PDS: %v", err)

     

       
217
       
 
       
		}

     

       
218
       
+
       
		defer func() { _ = pdsResp.Body.Close() }()

     

       
219
       
 
       


     

       
220
       
 
       
		if pdsResp.StatusCode != http.StatusOK {

     

       
221
       
+
       
			body, readErr := io.ReadAll(pdsResp.Body)

     

       
222
       
+
       
			if readErr != nil {

     

       
223
       
+
       
				t.Fatalf("PDS returned status %d (failed to read body: %v)", pdsResp.StatusCode, readErr)

     

       
224
       
+
       
			}

     

       
225
       
 
       
			t.Fatalf("PDS returned status %d: %s", pdsResp.StatusCode, string(body))

     

       
226
       
 
       
		}

     

       
227
       
 
       


     

       
228
       
 
       
		var pdsRecord struct {

     

       
229
       
+
       
			Value map[string]interface{} `json:"value"`

     

       
230
       
 
       
			URI   string                 `json:"uri"`

     

       
231
       
 
       
			CID   string                 `json:"cid"`

     

       
0
       
 
       

     

       
232
       
 
       
		}

     

       
233
       
 
       


     

       
234
       
 
       
		if err := json.NewDecoder(pdsResp.Body).Decode(&pdsRecord); err != nil {

     
···

       
240
       
 
       
		t.Logf("   CID: %s", pdsRecord.CID)

     

       
241
       
 
       


     

       
242
       
 
       
		// Print full record for inspection

     

       
243
       
+
       
		recordJSON, marshalErr := json.MarshalIndent(pdsRecord.Value, "   ", "  ")

     

       
244
       
+
       
		if marshalErr != nil {

     

       
245
       
+
       
			t.Logf("   Failed to marshal record: %v", marshalErr)

     

       
246
       
+
       
		} else {

     

       
247
       
+
       
			t.Logf("   Record value:\n   %s", string(recordJSON))

     

       
248
       
+
       
		}

     

       
249
       
 
       


     

       
250
       
 
       
		// V2: DID is NOT in the record - it's in the repository URI

     

       
251
       
 
       
		// The record should have handle, name, etc. but no 'did' field

     
···

       
344
       
 
       
	// Part 3: XRPC HTTP Endpoints

     

       
345
       
 
       
	// ====================================================================================

     

       
346
       
 
       
	t.Run("3. XRPC HTTP Endpoints", func(t *testing.T) {

     

       
0
       
 
       

     

       
347
       
 
       
		t.Run("Create via XRPC endpoint", func(t *testing.T) {

     

       
348
       
 
       
			// Use Unix timestamp (seconds) instead of UnixNano to keep handle short

     

       
349
       
 
       
			createReq := map[string]interface{}{

     
···

       
356
       
 
       
				"allowExternalDiscovery": true,

     

       
357
       
 
       
			}

     

       
358
       
 
       


     

       
359
       
+
       
			reqBody, marshalErr := json.Marshal(createReq)

     

       
360
       
+
       
			if marshalErr != nil {

     

       
361
       
+
       
				t.Fatalf("Failed to marshal request: %v", marshalErr)

     

       
362
       
+
       
			}

     

       
363
       
 
       


     

       
364
       
 
       
			// Step 1: Client POSTs to XRPC endpoint

     

       
365
       
 
       
			t.Logf("📡 Client → POST /xrpc/social.coves.community.create")

     
···

       
372
       
 
       
			if err != nil {

     

       
373
       
 
       
				t.Fatalf("Failed to POST: %v", err)

     

       
374
       
 
       
			}

     

       
375
       
+
       
			defer func() { _ = resp.Body.Close() }()

     

       
376
       
 
       


     

       
377
       
 
       
			if resp.StatusCode != http.StatusOK {

     

       
378
       
+
       
				body, readErr := io.ReadAll(resp.Body)

     

       
379
       
+
       
				if readErr != nil {

     

       
380
       
+
       
					t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)

     

       
381
       
+
       
				}

     

       
382
       
 
       
				t.Logf("❌ XRPC Create Failed")

     

       
383
       
 
       
				t.Logf("   Status: %d", resp.StatusCode)

     

       
384
       
 
       
				t.Logf("   Response: %s", string(body))

     
···

       
392
       
 
       
				Handle string `json:"handle"`

     

       
393
       
 
       
			}

     

       
394
       
 
       


     

       
395
       
+
       
			if err := json.NewDecoder(resp.Body).Decode(&createResp); err != nil {

     

       
396
       
+
       
				t.Fatalf("Failed to decode create response: %v", err)

     

       
397
       
+
       
			}

     

       
398
       
 
       


     

       
399
       
 
       
			t.Logf("✅ XRPC response received:")

     

       
400
       
 
       
			t.Logf("   DID:    %s", createResp.DID)

     
···

       
430
       
 
       
					CID: createResp.CID,

     

       
431
       
 
       
				},

     

       
432
       
 
       
			}

     

       
433
       
+
       
			if handleErr := consumer.HandleEvent(context.Background(), &event); handleErr != nil {

     

       
434
       
+
       
				t.Logf("Warning: failed to handle event: %v", handleErr)

     

       
435
       
+
       
			}

     

       
436
       
 
       


     

       
437
       
 
       
			// Step 3: Verify it's indexed in AppView

     

       
438
       
 
       
			t.Logf("🔍 Querying AppView to verify indexing...")

     
···

       
466
       
 
       
			if err != nil {

     

       
467
       
 
       
				t.Fatalf("Failed to GET: %v", err)

     

       
468
       
 
       
			}

     

       
469
       
+
       
			defer func() { _ = resp.Body.Close() }()

     

       
470
       
 
       


     

       
471
       
 
       
			if resp.StatusCode != http.StatusOK {

     

       
472
       
+
       
				body, readErr := io.ReadAll(resp.Body)

     

       
473
       
+
       
				if readErr != nil {

     

       
474
       
+
       
					t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)

     

       
475
       
+
       
				}

     

       
476
       
 
       
				t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))

     

       
477
       
 
       
			}

     

       
478
       
 
       


     

       
479
       
 
       
			var getCommunity communities.Community

     

       
480
       
+
       
			if err := json.NewDecoder(resp.Body).Decode(&getCommunity); err != nil {

     

       
481
       
+
       
				t.Fatalf("Failed to decode get response: %v", err)

     

       
482
       
+
       
			}

     

       
483
       
 
       


     

       
484
       
+
       
			t.Logf("Retrieved via XRPC HTTP endpoint:")

     

       
485
       
 
       
			t.Logf("   DID:         %s", getCommunity.DID)

     

       
486
       
 
       
			t.Logf("   DisplayName: %s", getCommunity.DisplayName)

     

       
487
       
 
       


     
···

       
501
       
 
       
			if err != nil {

     

       
502
       
 
       
				t.Fatalf("Failed to GET list: %v", err)

     

       
503
       
 
       
			}

     

       
504
       
+
       
			defer func() { _ = resp.Body.Close() }()

     

       
505
       
 
       


     

       
506
       
 
       
			if resp.StatusCode != http.StatusOK {

     

       
507
       
+
       
				body, readErr := io.ReadAll(resp.Body)

     

       
508
       
+
       
				if readErr != nil {

     

       
509
       
+
       
					t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)

     

       
510
       
+
       
				}

     

       
511
       
 
       
				t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))

     

       
512
       
 
       
			}

     

       
513
       
 
       


     
···

       
516
       
 
       
				Total       int                     `json:"total"`

     

       
517
       
 
       
			}

     

       
518
       
 
       


     

       
519
       
+
       
			if err := json.NewDecoder(resp.Body).Decode(&listResp); err != nil {

     

       
520
       
+
       
				t.Fatalf("Failed to decode list response: %v", err)

     

       
521
       
+
       
			}

     

       
522
       
 
       


     

       
523
       
 
       
			t.Logf("✅ Listed %d communities via XRPC", len(listResp.Communities))

     

       
524
       
 
       


     
···

       
580
       
 
       
	collection := "social.coves.community.profile"

     

       
581
       
 
       
	rkey := extractRKeyFromURI(community.RecordURI)

     

       
582
       
 
       


     

       
583
       
+
       
	pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",

     

       
584
       
 
       
		pdsURL, instanceDID, collection, rkey))

     

       
585
       
+
       
	if pdsErr != nil {

     

       
586
       
+
       
		t.Fatalf("Failed to fetch PDS record: %v", pdsErr)

     

       
587
       
+
       
	}

     

       
588
       
+
       
	defer func() {

     

       
589
       
+
       
		if closeErr := pdsResp.Body.Close(); closeErr != nil {

     

       
590
       
+
       
			t.Logf("Failed to close PDS response: %v", closeErr)

     

       
591
       
+
       
		}

     

       
592
       
+
       
	}()

     

       
593
       
 
       


     

       
594
       
 
       
	var pdsRecord struct {

     

       
0
       
 
       

     

       
595
       
 
       
		Value map[string]interface{} `json:"value"`

     

       
596
       
+
       
		CID   string                 `json:"cid"`

     

       
597
       
 
       
	}

     

       
598
       
+
       
	if decodeErr := json.NewDecoder(pdsResp.Body).Decode(&pdsRecord); decodeErr != nil {

     

       
599
       
+
       
		t.Fatalf("Failed to decode PDS record: %v", decodeErr)

     

       
600
       
+
       
	}

     

       
601
       
 
       


     

       
602
       
 
       
	// Simulate firehose event

     

       
603
       
 
       
	event := jetstream.JetstreamEvent{

     
···

       
614
       
 
       
		},

     

       
615
       
 
       
	}

     

       
616
       
 
       


     

       
617
       
+
       
	if handleErr := consumer.HandleEvent(context.Background(), &event); handleErr != nil {

     

       
618
       
+
       
		t.Logf("Warning: failed to handle event: %v", handleErr)

     

       
619
       
+
       
	}

     

       
620
       
 
       


     

       
621
       
 
       
	return community

     

       
622
       
 
       
}

     
···

       
638
       
 
       
		"password":   password,

     

       
639
       
 
       
	}

     

       
640
       
 
       


     

       
641
       
+
       
	reqBody, marshalErr := json.Marshal(sessionReq)

     

       
642
       
+
       
	if marshalErr != nil {

     

       
643
       
+
       
		return "", "", fmt.Errorf("failed to marshal session request: %w", marshalErr)

     

       
644
       
+
       
	}

     

       
645
       
 
       
	resp, err := http.Post(

     

       
646
       
 
       
		pdsURL+"/xrpc/com.atproto.server.createSession",

     

       
647
       
 
       
		"application/json",

     
···

       
650
       
 
       
	if err != nil {

     

       
651
       
 
       
		return "", "", fmt.Errorf("failed to create session: %w", err)

     

       
652
       
 
       
	}

     

       
653
       
+
       
	defer func() { _ = resp.Body.Close() }()

     

       
654
       
 
       


     

       
655
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
656
       
+
       
		body, readErr := io.ReadAll(resp.Body)

     

       
657
       
+
       
		if readErr != nil {

     

       
658
       
+
       
			return "", "", fmt.Errorf("PDS auth failed (status %d, failed to read body: %w)", resp.StatusCode, readErr)

     

       
659
       
+
       
		}

     

       
660
       
 
       
		return "", "", fmt.Errorf("PDS auth failed (status %d): %s", resp.StatusCode, string(body))

     

       
661
       
 
       
	}

     

       
662
       
 
       


     
···

       
715
       
 
       
	if err != nil {

     

       
716
       
 
       
		return "", "", fmt.Errorf("failed to query PDS: %w", err)

     

       
717
       
 
       
	}

     

       
718
       
+
       
	defer func() { _ = resp.Body.Close() }()

     

       
719
       
 
       


     

       
720
       
 
       
	if resp.StatusCode != http.StatusOK {

     

       
721
       
+
       
		body, readErr := io.ReadAll(resp.Body)

     

       
722
       
+
       
		if readErr != nil {

     

       
723
       
+
       
			return "", "", fmt.Errorf("account not found (status %d, failed to read body: %w)", resp.StatusCode, readErr)

     

       
724
       
+
       
		}

     

       
725
       
 
       
		return "", "", fmt.Errorf("account not found (status %d): %s", resp.StatusCode, string(body))

     

       
726
       
 
       
	}

     

       
727
       
 
       


     
···

       
753
       
 
       
	if err != nil {

     

       
754
       
 
       
		return fmt.Errorf("failed to connect to Jetstream: %w", err)

     

       
755
       
 
       
	}

     

       
756
       
+
       
	defer func() { _ = conn.Close() }()

     

       
757
       
 
       


     

       
758
       
 
       
	// Read messages until we find our event or receive done signal

     

       
759
       
 
       
	for {

     
···

       
764
       
 
       
			return ctx.Err()

     

       
765
       
 
       
		default:

     

       
766
       
 
       
			// Set read deadline to avoid blocking forever

     

       
767
       
+
       
			if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {

     

       
768
       
+
       
				return fmt.Errorf("failed to set read deadline: %w", err)

     

       
769
       
+
       
			}

     

       
770
       
 
       


     

       
771
       
 
       
			var event jetstream.JetstreamEvent

     

       
772
       
 
       
			err := conn.ReadJSON(&event)

     

···

       
1
       
 
       
package integration

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
4
       
 
       
	"context"

     

       
5
       
 
       
	"fmt"

     

       
6
       
 
       
	"testing"

     

       
7
       
 
       
	"time"

     

       
8
       
-
       


     

       
9
       
-
       
	"Coves/internal/atproto/did"

     

       
10
       
-
       
	"Coves/internal/core/communities"

     

       
11
       
-
       
	"Coves/internal/db/postgres"

     

       
12
       
 
       
)

     

       
13
       
 
       


     

       
14
       
 
       
func TestCommunityRepository_Create(t *testing.T) {

     

       
15
       
 
       
	db := setupTestDB(t)

     

       
16
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
17
       
 
       


     

       
18
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
19
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
20
       
 
       
	ctx := context.Background()

     

       
21
       
 
       


     

       
22
       
 
       
	t.Run("creates community successfully", func(t *testing.T) {

     

       
23
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
24
       
 
       
		// Generate unique handle using timestamp to avoid collisions

     

       
25
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
26
       
 
       
		community := &communities.Community{

     
···

       
52
       
 
       
	})

     

       
53
       
 
       


     

       
54
       
 
       
	t.Run("returns error for duplicate DID", func(t *testing.T) {

     

       
55
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
56
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
57
       
 
       
		community := &communities.Community{

     

       
58
       
 
       
			DID:          communityDID,

     
···

       
67
       
 
       
		}

     

       
68
       
 
       


     

       
69
       
 
       
		// Create first time

     

       
70
       
-
       
		_, err := repo.Create(ctx, community)

     

       
71
       
-
       
		if err != nil {

     

       
72
       
 
       
			t.Fatalf("First create failed: %v", err)

     

       
73
       
 
       
		}

     

       
74
       
 
       


     

       
75
       
 
       
		// Try to create again with same DID

     

       
76
       
-
       
		_, err = repo.Create(ctx, community)

     

       
77
       
-
       
		if err != communities.ErrCommunityAlreadyExists {

     

       
78
       
 
       
			t.Errorf("Expected ErrCommunityAlreadyExists, got: %v", err)

     

       
79
       
 
       
		}

     

       
80
       
 
       
	})

     
···

       
84
       
 
       
		handle := fmt.Sprintf("!unique-handle-%s@coves.local", uniqueSuffix)

     

       
85
       
 
       


     

       
86
       
 
       
		// First community

     

       
87
       
-
       
		did1, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
88
       
 
       
		community1 := &communities.Community{

     

       
89
       
 
       
			DID:          did1,

     

       
90
       
 
       
			Handle:       handle,

     
···

       
97
       
 
       
			UpdatedAt:    time.Now(),

     

       
98
       
 
       
		}

     

       
99
       
 
       


     

       
100
       
-
       
		_, err := repo.Create(ctx, community1)

     

       
101
       
-
       
		if err != nil {

     

       
102
       
 
       
			t.Fatalf("First create failed: %v", err)

     

       
103
       
 
       
		}

     

       
104
       
 
       


     

       
105
       
 
       
		// Second community with different DID but same handle

     

       
106
       
-
       
		did2, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
107
       
 
       
		community2 := &communities.Community{

     

       
108
       
 
       
			DID:          did2,

     

       
109
       
 
       
			Handle:       handle, // Same handle!

     
···

       
116
       
 
       
			UpdatedAt:    time.Now(),

     

       
117
       
 
       
		}

     

       
118
       
 
       


     

       
119
       
-
       
		_, err = repo.Create(ctx, community2)

     

       
120
       
-
       
		if err != communities.ErrHandleTaken {

     

       
121
       
 
       
			t.Errorf("Expected ErrHandleTaken, got: %v", err)

     

       
122
       
 
       
		}

     

       
123
       
 
       
	})

     
···

       
125
       
 
       


     

       
126
       
 
       
func TestCommunityRepository_GetByDID(t *testing.T) {

     

       
127
       
 
       
	db := setupTestDB(t)

     

       
128
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
129
       
 
       


     

       
130
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
131
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
132
       
 
       
	ctx := context.Background()

     

       
133
       
 
       


     

       
134
       
 
       
	t.Run("retrieves existing community", func(t *testing.T) {

     

       
135
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
136
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
137
       
 
       
		community := &communities.Community{

     

       
138
       
 
       
			DID:          communityDID,

     
···

       
170
       
 
       
	})

     

       
171
       
 
       


     

       
172
       
 
       
	t.Run("returns error for non-existent community", func(t *testing.T) {

     

       
173
       
-
       
		fakeDID, _ := didGen.GenerateCommunityDID()

     

       
174
       
-
       
		_, err := repo.GetByDID(ctx, fakeDID)

     

       
175
       
-
       
		if err != communities.ErrCommunityNotFound {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
176
       
 
       
			t.Errorf("Expected ErrCommunityNotFound, got: %v", err)

     

       
177
       
 
       
		}

     

       
178
       
 
       
	})

     
···

       
180
       
 
       


     

       
181
       
 
       
func TestCommunityRepository_GetByHandle(t *testing.T) {

     

       
182
       
 
       
	db := setupTestDB(t)

     

       
183
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
184
       
 
       


     

       
185
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
186
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
187
       
 
       
	ctx := context.Background()

     

       
188
       
 
       


     

       
189
       
 
       
	t.Run("retrieves community by handle", func(t *testing.T) {

     

       
190
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
191
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
192
       
 
       
		handle := fmt.Sprintf("!handle-lookup-%s@coves.local", uniqueSuffix)

     

       
193
       
 
       


     
···

       
203
       
 
       
			UpdatedAt:    time.Now(),

     

       
204
       
 
       
		}

     

       
205
       
 
       


     

       
206
       
-
       
		_, err := repo.Create(ctx, community)

     

       
207
       
-
       
		if err != nil {

     

       
208
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
209
       
 
       
		}

     

       
210
       
 
       


     
···

       
224
       
 
       


     

       
225
       
 
       
func TestCommunityRepository_Subscriptions(t *testing.T) {

     

       
226
       
 
       
	db := setupTestDB(t)

     

       
227
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
228
       
 
       


     

       
229
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
230
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     

       
231
       
 
       
	ctx := context.Background()

     

       
232
       
 
       


     

       
233
       
 
       
	// Create a community for subscription tests

     

       
234
       
-
       
	communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
235
       
 
       
	uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
236
       
 
       
	community := &communities.Community{

     

       
237
       
 
       
		DID:          communityDID,

     
···

       
245
       
 
       
		UpdatedAt:    time.Now(),

     

       
246
       
 
       
	}

     

       
247
       
 
       


     

       
248
       
-
       
	_, err := repo.Create(ctx, community)

     

       
249
       
-
       
	if err != nil {

     

       
250
       
 
       
		t.Fatalf("Failed to create community: %v", err)

     

       
251
       
 
       
	}

     

       
252
       
 
       


     
···

       
274
       
 
       
			SubscribedAt: time.Now(),

     

       
275
       
 
       
		}

     

       
276
       
 
       


     

       
277
       
-
       
		_, err := repo.Subscribe(ctx, sub)

     

       
278
       
-
       
		if err != nil {

     

       
279
       
 
       
			t.Fatalf("First subscription failed: %v", err)

     

       
280
       
 
       
		}

     

       
281
       
 
       


     
···

       
314
       
 
       


     

       
315
       
 
       
func TestCommunityRepository_List(t *testing.T) {

     

       
316
       
 
       
	db := setupTestDB(t)

     

       
317
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
318
       
 
       


     

       
319
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
320
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     
···

       
324
       
 
       
		// Create multiple communities

     

       
325
       
 
       
		baseSuffix := time.Now().UnixNano()

     

       
326
       
 
       
		for i := 0; i < 5; i++ {

     

       
327
       
-
       
			communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
328
       
 
       
			community := &communities.Community{

     

       
329
       
 
       
				DID:          communityDID,

     

       
330
       
 
       
				Handle:       fmt.Sprintf("!list-test-%d-%d@coves.local", baseSuffix, i),

     
···

       
336
       
 
       
				CreatedAt:    time.Now(),

     

       
337
       
 
       
				UpdatedAt:    time.Now(),

     

       
338
       
 
       
			}

     

       
339
       
-
       
			_, err := repo.Create(ctx, community)

     

       
340
       
-
       
			if err != nil {

     

       
341
       
 
       
				t.Fatalf("Failed to create community %d: %v", i, err)

     

       
342
       
 
       
			}

     

       
343
       
 
       
			time.Sleep(10 * time.Millisecond) // Ensure different timestamps

     
···

       
365
       
 
       


     

       
366
       
 
       
	t.Run("filters by visibility", func(t *testing.T) {

     

       
367
       
 
       
		// Create an unlisted community

     

       
368
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
369
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
370
       
 
       
		community := &communities.Community{

     

       
371
       
 
       
			DID:          communityDID,

     
···

       
379
       
 
       
			UpdatedAt:    time.Now(),

     

       
380
       
 
       
		}

     

       
381
       
 
       


     

       
382
       
-
       
		_, err := repo.Create(ctx, community)

     

       
383
       
-
       
		if err != nil {

     

       
384
       
 
       
			t.Fatalf("Failed to create unlisted community: %v", err)

     

       
385
       
 
       
		}

     

       
386
       
 
       


     
···

       
407
       
 
       


     

       
408
       
 
       
func TestCommunityRepository_Search(t *testing.T) {

     

       
409
       
 
       
	db := setupTestDB(t)

     

       
410
       
-
       
	defer db.Close()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
411
       
 
       


     

       
412
       
 
       
	repo := postgres.NewCommunityRepository(db)

     

       
413
       
 
       
	didGen := did.NewGenerator(true, "https://plc.directory")

     
···

       
415
       
 
       


     

       
416
       
 
       
	t.Run("searches communities by name", func(t *testing.T) {

     

       
417
       
 
       
		// Create a community with searchable name

     

       
418
       
-
       
		communityDID, _ := didGen.GenerateCommunityDID()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
419
       
 
       
		uniqueSuffix := fmt.Sprintf("%d", time.Now().UnixNano())

     

       
420
       
 
       
		community := &communities.Community{

     

       
421
       
 
       
			DID:          communityDID,

     
···

       
431
       
 
       
			UpdatedAt:    time.Now(),

     

       
432
       
 
       
		}

     

       
433
       
 
       


     

       
434
       
-
       
		_, err := repo.Create(ctx, community)

     

       
435
       
-
       
		if err != nil {

     

       
436
       
 
       
			t.Fatalf("Failed to create community: %v", err)

     

       
437
       
 
       
		}

     

       
438