A community based topic aggregation platform built on atproto
bretton.dev
1package postgres
2
3import (
4 "Coves/internal/core/communities"
5 "context"
6 "database/sql"
7 "fmt"
8 "log"
9 "strings"
10
11 "github.com/lib/pq"
12)
13
14type postgresCommunityRepo struct {
15 db *sql.DB
16}
17
18// NewCommunityRepository creates a new PostgreSQL community repository
19func NewCommunityRepository(db *sql.DB) communities.Repository {
20 return &postgresCommunityRepo{db: db}
21}
22
23// Create inserts a new community into the communities table
24func (r *postgresCommunityRepo) Create(ctx context.Context, community *communities.Community) (*communities.Community, error) {
25 // Validate that handle is always provided (constructed by consumer)
26 if community.Handle == "" {
27 return nil, fmt.Errorf("handle is required (should be constructed by consumer before insert)")
28 }
29
30 query := `
31 INSERT INTO communities (
32 did, handle, name, display_name, description, description_facets,
33 avatar_cid, banner_cid, owner_did, created_by_did, hosted_by_did,
34 pds_email, pds_password_encrypted,
35 pds_access_token_encrypted, pds_refresh_token_encrypted, pds_url,
36 visibility, allow_external_discovery, moderation_type, content_warnings,
37 member_count, subscriber_count, post_count,
38 federated_from, federated_id, created_at, updated_at,
39 record_uri, record_cid
40 ) VALUES (
41 $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11,
42 $12,
43 CASE WHEN $13 != '' THEN pgp_sym_encrypt($13, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1)) ELSE NULL END,
44 CASE WHEN $14 != '' THEN pgp_sym_encrypt($14, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1)) ELSE NULL END,
45 CASE WHEN $15 != '' THEN pgp_sym_encrypt($15, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1)) ELSE NULL END,
46 $16,
47 $17, $18, $19, $20,
48 $21, $22, $23, $24, $25, $26, $27, $28, $29
49 )
50 RETURNING id, created_at, updated_at`
51
52 // Handle JSONB field - use sql.NullString with valid JSON or NULL
53 var descFacets interface{}
54 if len(community.DescriptionFacets) > 0 {
55 descFacets = community.DescriptionFacets
56 } else {
57 descFacets = nil
58 }
59
60 err := r.db.QueryRowContext(ctx, query,
61 community.DID,
62 community.Handle, // Always non-empty - constructed by AppView consumer
63 community.Name,
64 nullString(community.DisplayName),
65 nullString(community.Description),
66 descFacets,
67 nullString(community.AvatarCID),
68 nullString(community.BannerCID),
69 community.OwnerDID,
70 community.CreatedByDID,
71 community.HostedByDID,
72 // V2.0: PDS credentials for community account (encrypted at rest)
73 nullString(community.PDSEmail),
74 nullString(community.PDSPassword), // Encrypted by pgp_sym_encrypt
75 nullString(community.PDSAccessToken), // Encrypted by pgp_sym_encrypt
76 nullString(community.PDSRefreshToken), // Encrypted by pgp_sym_encrypt
77 nullString(community.PDSURL),
78 // V2.0: No key columns - PDS manages all keys
79 community.Visibility,
80 community.AllowExternalDiscovery,
81 nullString(community.ModerationType),
82 pq.Array(community.ContentWarnings),
83 community.MemberCount,
84 community.SubscriberCount,
85 community.PostCount,
86 nullString(community.FederatedFrom),
87 nullString(community.FederatedID),
88 community.CreatedAt,
89 community.UpdatedAt,
90 nullString(community.RecordURI),
91 nullString(community.RecordCID),
92 ).Scan(&community.ID, &community.CreatedAt, &community.UpdatedAt)
93 if err != nil {
94 // Check for unique constraint violations
95 if strings.Contains(err.Error(), "duplicate key") {
96 if strings.Contains(err.Error(), "communities_did_key") {
97 return nil, communities.ErrCommunityAlreadyExists
98 }
99 if strings.Contains(err.Error(), "communities_handle_key") {
100 return nil, communities.ErrHandleTaken
101 }
102 }
103 return nil, fmt.Errorf("failed to create community: %w", err)
104 }
105
106 return community, nil
107}
108
109// GetByDID retrieves a community by its DID
110// Note: PDS credentials are included (for internal service use only)
111// Handlers MUST use json:"-" tags to prevent credential exposure in APIs
112//
113// V2.0: Key columns not included - PDS manages all keys
114func (r *postgresCommunityRepo) GetByDID(ctx context.Context, did string) (*communities.Community, error) {
115 community := &communities.Community{}
116 query := `
117 SELECT id, did, handle, name, display_name, description, description_facets,
118 avatar_cid, banner_cid, owner_did, created_by_did, hosted_by_did,
119 pds_email,
120 CASE
121 WHEN pds_password_encrypted IS NOT NULL
122 THEN pgp_sym_decrypt(pds_password_encrypted, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1))
123 ELSE NULL
124 END as pds_password,
125 CASE
126 WHEN pds_access_token_encrypted IS NOT NULL
127 THEN pgp_sym_decrypt(pds_access_token_encrypted, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1))
128 ELSE NULL
129 END as pds_access_token,
130 CASE
131 WHEN pds_refresh_token_encrypted IS NOT NULL
132 THEN pgp_sym_decrypt(pds_refresh_token_encrypted, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1))
133 ELSE NULL
134 END as pds_refresh_token,
135 pds_url,
136 visibility, allow_external_discovery, moderation_type, content_warnings,
137 member_count, subscriber_count, post_count,
138 federated_from, federated_id, created_at, updated_at,
139 record_uri, record_cid
140 FROM communities
141 WHERE did = $1`
142
143 var displayName, description, avatarCID, bannerCID, moderationType sql.NullString
144 var federatedFrom, federatedID, recordURI, recordCID sql.NullString
145 var pdsEmail, pdsPassword, pdsAccessToken, pdsRefreshToken, pdsURL sql.NullString
146 var descFacets []byte
147 var contentWarnings []string
148
149 err := r.db.QueryRowContext(ctx, query, did).Scan(
150 &community.ID, &community.DID, &community.Handle, &community.Name,
151 &displayName, &description, &descFacets,
152 &avatarCID, &bannerCID,
153 &community.OwnerDID, &community.CreatedByDID, &community.HostedByDID,
154 // V2.0: PDS credentials (decrypted from pgp_sym_encrypt)
155 &pdsEmail, &pdsPassword, &pdsAccessToken, &pdsRefreshToken, &pdsURL,
156 &community.Visibility, &community.AllowExternalDiscovery,
157 &moderationType, pq.Array(&contentWarnings),
158 &community.MemberCount, &community.SubscriberCount, &community.PostCount,
159 &federatedFrom, &federatedID,
160 &community.CreatedAt, &community.UpdatedAt,
161 &recordURI, &recordCID,
162 )
163
164 if err == sql.ErrNoRows {
165 return nil, communities.ErrCommunityNotFound
166 }
167 if err != nil {
168 return nil, fmt.Errorf("failed to get community by DID: %w", err)
169 }
170
171 // Map nullable fields
172 community.DisplayName = displayName.String
173 community.Description = description.String
174 community.AvatarCID = avatarCID.String
175 community.BannerCID = bannerCID.String
176 community.PDSEmail = pdsEmail.String
177 community.PDSPassword = pdsPassword.String
178 community.PDSAccessToken = pdsAccessToken.String
179 community.PDSRefreshToken = pdsRefreshToken.String
180 community.PDSURL = pdsURL.String
181 // V2.0: No key fields - PDS manages all keys
182 community.RotationKeyPEM = "" // Empty - PDS-managed
183 community.SigningKeyPEM = "" // Empty - PDS-managed
184 community.ModerationType = moderationType.String
185 community.ContentWarnings = contentWarnings
186 community.FederatedFrom = federatedFrom.String
187 community.FederatedID = federatedID.String
188 community.RecordURI = recordURI.String
189 community.RecordCID = recordCID.String
190 if descFacets != nil {
191 community.DescriptionFacets = descFacets
192 }
193
194 return community, nil
195}
196
197// GetByHandle retrieves a community by its scoped handle
198func (r *postgresCommunityRepo) GetByHandle(ctx context.Context, handle string) (*communities.Community, error) {
199 community := &communities.Community{}
200 query := `
201 SELECT id, did, handle, name, display_name, description, description_facets,
202 avatar_cid, banner_cid, owner_did, created_by_did, hosted_by_did,
203 visibility, allow_external_discovery, moderation_type, content_warnings,
204 member_count, subscriber_count, post_count,
205 federated_from, federated_id, created_at, updated_at,
206 record_uri, record_cid
207 FROM communities
208 WHERE handle = $1`
209
210 var displayName, description, avatarCID, bannerCID, moderationType sql.NullString
211 var federatedFrom, federatedID, recordURI, recordCID sql.NullString
212 var descFacets []byte
213 var contentWarnings []string
214
215 err := r.db.QueryRowContext(ctx, query, handle).Scan(
216 &community.ID, &community.DID, &community.Handle, &community.Name,
217 &displayName, &description, &descFacets,
218 &avatarCID, &bannerCID,
219 &community.OwnerDID, &community.CreatedByDID, &community.HostedByDID,
220 &community.Visibility, &community.AllowExternalDiscovery,
221 &moderationType, pq.Array(&contentWarnings),
222 &community.MemberCount, &community.SubscriberCount, &community.PostCount,
223 &federatedFrom, &federatedID,
224 &community.CreatedAt, &community.UpdatedAt,
225 &recordURI, &recordCID,
226 )
227
228 if err == sql.ErrNoRows {
229 return nil, communities.ErrCommunityNotFound
230 }
231 if err != nil {
232 return nil, fmt.Errorf("failed to get community by handle: %w", err)
233 }
234
235 // Map nullable fields
236 community.DisplayName = displayName.String
237 community.Description = description.String
238 community.AvatarCID = avatarCID.String
239 community.BannerCID = bannerCID.String
240 community.ModerationType = moderationType.String
241 community.ContentWarnings = contentWarnings
242 community.FederatedFrom = federatedFrom.String
243 community.FederatedID = federatedID.String
244 community.RecordURI = recordURI.String
245 community.RecordCID = recordCID.String
246 if descFacets != nil {
247 community.DescriptionFacets = descFacets
248 }
249
250 return community, nil
251}
252
253// Update modifies an existing community's metadata
254func (r *postgresCommunityRepo) Update(ctx context.Context, community *communities.Community) (*communities.Community, error) {
255 query := `
256 UPDATE communities
257 SET display_name = $2, description = $3, description_facets = $4,
258 avatar_cid = $5, banner_cid = $6,
259 visibility = $7, allow_external_discovery = $8,
260 moderation_type = $9, content_warnings = $10,
261 updated_at = NOW(),
262 record_uri = $11, record_cid = $12
263 WHERE did = $1
264 RETURNING updated_at`
265
266 // Handle JSONB field - use sql.NullString with valid JSON or NULL
267 var descFacets interface{}
268 if len(community.DescriptionFacets) > 0 {
269 descFacets = community.DescriptionFacets
270 } else {
271 descFacets = nil
272 }
273
274 err := r.db.QueryRowContext(ctx, query,
275 community.DID,
276 nullString(community.DisplayName),
277 nullString(community.Description),
278 descFacets,
279 nullString(community.AvatarCID),
280 nullString(community.BannerCID),
281 community.Visibility,
282 community.AllowExternalDiscovery,
283 nullString(community.ModerationType),
284 pq.Array(community.ContentWarnings),
285 nullString(community.RecordURI),
286 nullString(community.RecordCID),
287 ).Scan(&community.UpdatedAt)
288
289 if err == sql.ErrNoRows {
290 return nil, communities.ErrCommunityNotFound
291 }
292 if err != nil {
293 return nil, fmt.Errorf("failed to update community: %w", err)
294 }
295
296 return community, nil
297}
298
299// UpdateCredentials atomically updates community's PDS access and refresh tokens
300// CRITICAL: Both tokens must be updated together because refresh tokens are single-use
301// After a successful token refresh, the old refresh token is immediately revoked by the PDS
302func (r *postgresCommunityRepo) UpdateCredentials(ctx context.Context, did, accessToken, refreshToken string) error {
303 query := `
304 UPDATE communities
305 SET
306 pds_access_token_encrypted = pgp_sym_encrypt($2, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1)),
307 pds_refresh_token_encrypted = pgp_sym_encrypt($3, (SELECT encode(key_data, 'hex') FROM encryption_keys WHERE id = 1)),
308 updated_at = NOW()
309 WHERE did = $1
310 RETURNING did`
311
312 var returnedDID string
313 err := r.db.QueryRowContext(ctx, query, did, accessToken, refreshToken).Scan(&returnedDID)
314
315 if err == sql.ErrNoRows {
316 return communities.ErrCommunityNotFound
317 }
318 if err != nil {
319 return fmt.Errorf("failed to update credentials: %w", err)
320 }
321
322 return nil
323}
324
325// Delete removes a community from the database
326func (r *postgresCommunityRepo) Delete(ctx context.Context, did string) error {
327 query := `DELETE FROM communities WHERE did = $1`
328
329 result, err := r.db.ExecContext(ctx, query, did)
330 if err != nil {
331 return fmt.Errorf("failed to delete community: %w", err)
332 }
333
334 rowsAffected, err := result.RowsAffected()
335 if err != nil {
336 return fmt.Errorf("failed to check delete result: %w", err)
337 }
338
339 if rowsAffected == 0 {
340 return communities.ErrCommunityNotFound
341 }
342
343 return nil
344}
345
346// List retrieves communities with filtering and pagination
347func (r *postgresCommunityRepo) List(ctx context.Context, req communities.ListCommunitiesRequest) ([]*communities.Community, error) {
348 // Build query with filters
349 whereClauses := []string{}
350 args := []interface{}{}
351 argCount := 1
352
353 if req.Visibility != "" {
354 whereClauses = append(whereClauses, fmt.Sprintf("visibility = $%d", argCount))
355 args = append(args, req.Visibility)
356 argCount++
357 }
358
359 // TODO: Add category filter when DB schema supports it
360 // if req.Category != "" { ... }
361
362 // TODO: Add language filter when DB schema supports it
363 // if req.Language != "" { ... }
364
365 whereClause := ""
366 if len(whereClauses) > 0 {
367 whereClause = "WHERE " + strings.Join(whereClauses, " AND ")
368 }
369
370 // Build sort clause - map sort enum to DB columns
371 sortColumn := "subscriber_count" // default: popular
372 sortOrder := "DESC"
373
374 switch req.Sort {
375 case "popular":
376 // Most subscribers (default)
377 sortColumn = "subscriber_count"
378 sortOrder = "DESC"
379 case "active":
380 // Most posts/activity
381 sortColumn = "post_count"
382 sortOrder = "DESC"
383 case "new":
384 // Recently created
385 sortColumn = "created_at"
386 sortOrder = "DESC"
387 case "alphabetical":
388 // Sorted by name A-Z
389 sortColumn = "name"
390 sortOrder = "ASC"
391 default:
392 // Fallback to popular if empty or invalid (should be validated in handler)
393 sortColumn = "subscriber_count"
394 sortOrder = "DESC"
395 }
396
397 // Get communities with pagination
398 query := fmt.Sprintf(`
399 SELECT id, did, handle, name, display_name, description, description_facets,
400 avatar_cid, banner_cid, owner_did, created_by_did, hosted_by_did,
401 visibility, allow_external_discovery, moderation_type, content_warnings,
402 member_count, subscriber_count, post_count,
403 federated_from, federated_id, created_at, updated_at,
404 record_uri, record_cid
405 FROM communities
406 %s
407 ORDER BY %s %s
408 LIMIT $%d OFFSET $%d`,
409 whereClause, sortColumn, sortOrder, argCount, argCount+1)
410
411 args = append(args, req.Limit, req.Offset)
412
413 rows, err := r.db.QueryContext(ctx, query, args...)
414 if err != nil {
415 return nil, fmt.Errorf("failed to list communities: %w", err)
416 }
417 defer func() {
418 if closeErr := rows.Close(); closeErr != nil {
419 log.Printf("Failed to close rows: %v", closeErr)
420 }
421 }()
422
423 result := []*communities.Community{}
424 for rows.Next() {
425 community := &communities.Community{}
426 var displayName, description, avatarCID, bannerCID, moderationType sql.NullString
427 var federatedFrom, federatedID, recordURI, recordCID sql.NullString
428 var descFacets []byte
429 var contentWarnings []string
430
431 scanErr := rows.Scan(
432 &community.ID, &community.DID, &community.Handle, &community.Name,
433 &displayName, &description, &descFacets,
434 &avatarCID, &bannerCID,
435 &community.OwnerDID, &community.CreatedByDID, &community.HostedByDID,
436 &community.Visibility, &community.AllowExternalDiscovery,
437 &moderationType, pq.Array(&contentWarnings),
438 &community.MemberCount, &community.SubscriberCount, &community.PostCount,
439 &federatedFrom, &federatedID,
440 &community.CreatedAt, &community.UpdatedAt,
441 &recordURI, &recordCID,
442 )
443 if scanErr != nil {
444 return nil, fmt.Errorf("failed to scan community: %w", scanErr)
445 }
446
447 // Map nullable fields
448 community.DisplayName = displayName.String
449 community.Description = description.String
450 community.AvatarCID = avatarCID.String
451 community.BannerCID = bannerCID.String
452 community.ModerationType = moderationType.String
453 community.ContentWarnings = contentWarnings
454 community.FederatedFrom = federatedFrom.String
455 community.FederatedID = federatedID.String
456 community.RecordURI = recordURI.String
457 community.RecordCID = recordCID.String
458 if descFacets != nil {
459 community.DescriptionFacets = descFacets
460 }
461
462 result = append(result, community)
463 }
464
465 if err = rows.Err(); err != nil {
466 return nil, fmt.Errorf("error iterating communities: %w", err)
467 }
468
469 return result, nil
470}
471
472// Search searches communities by name/description using fuzzy matching
473func (r *postgresCommunityRepo) Search(ctx context.Context, req communities.SearchCommunitiesRequest) ([]*communities.Community, int, error) {
474 // Build query with fuzzy search and visibility filter
475 whereClauses := []string{
476 "(name ILIKE '%' || $1 || '%' OR description ILIKE '%' || $1 || '%')",
477 }
478 args := []interface{}{req.Query}
479 argCount := 2
480
481 if req.Visibility != "" {
482 whereClauses = append(whereClauses, fmt.Sprintf("visibility = $%d", argCount))
483 args = append(args, req.Visibility)
484 argCount++
485 }
486
487 whereClause := "WHERE " + strings.Join(whereClauses, " AND ")
488
489 // Get total count
490 countQuery := fmt.Sprintf("SELECT COUNT(*) FROM communities %s", whereClause)
491 var totalCount int
492 err := r.db.QueryRowContext(ctx, countQuery, args...).Scan(&totalCount)
493 if err != nil {
494 return nil, 0, fmt.Errorf("failed to count search results: %w", err)
495 }
496
497 // Search with relevance ranking using pg_trgm similarity
498 // Filter out results with very low relevance (< 0.2) to avoid noise
499 query := fmt.Sprintf(`
500 SELECT id, did, handle, name, display_name, description, description_facets,
501 avatar_cid, banner_cid, owner_did, created_by_did, hosted_by_did,
502 visibility, allow_external_discovery, moderation_type, content_warnings,
503 member_count, subscriber_count, post_count,
504 federated_from, federated_id, created_at, updated_at,
505 record_uri, record_cid,
506 similarity(name, $1) + similarity(COALESCE(description, ''), $1) as relevance
507 FROM communities
508 %s AND (similarity(name, $1) + similarity(COALESCE(description, ''), $1)) > 0.2
509 ORDER BY relevance DESC, member_count DESC
510 LIMIT $%d OFFSET $%d`,
511 whereClause, argCount, argCount+1)
512
513 args = append(args, req.Limit, req.Offset)
514
515 rows, err := r.db.QueryContext(ctx, query, args...)
516 if err != nil {
517 return nil, 0, fmt.Errorf("failed to search communities: %w", err)
518 }
519 defer func() {
520 if closeErr := rows.Close(); closeErr != nil {
521 log.Printf("Failed to close rows: %v", closeErr)
522 }
523 }()
524
525 result := []*communities.Community{}
526 for rows.Next() {
527 community := &communities.Community{}
528 var displayName, description, avatarCID, bannerCID, moderationType sql.NullString
529 var federatedFrom, federatedID, recordURI, recordCID sql.NullString
530 var descFacets []byte
531 var contentWarnings []string
532 var relevance float64
533
534 scanErr := rows.Scan(
535 &community.ID, &community.DID, &community.Handle, &community.Name,
536 &displayName, &description, &descFacets,
537 &avatarCID, &bannerCID,
538 &community.OwnerDID, &community.CreatedByDID, &community.HostedByDID,
539 &community.Visibility, &community.AllowExternalDiscovery,
540 &moderationType, pq.Array(&contentWarnings),
541 &community.MemberCount, &community.SubscriberCount, &community.PostCount,
542 &federatedFrom, &federatedID,
543 &community.CreatedAt, &community.UpdatedAt,
544 &recordURI, &recordCID,
545 &relevance,
546 )
547 if scanErr != nil {
548 return nil, 0, fmt.Errorf("failed to scan community: %w", scanErr)
549 }
550
551 // Map nullable fields
552 community.DisplayName = displayName.String
553 community.Description = description.String
554 community.AvatarCID = avatarCID.String
555 community.BannerCID = bannerCID.String
556 community.ModerationType = moderationType.String
557 community.ContentWarnings = contentWarnings
558 community.FederatedFrom = federatedFrom.String
559 community.FederatedID = federatedID.String
560 community.RecordURI = recordURI.String
561 community.RecordCID = recordCID.String
562 if descFacets != nil {
563 community.DescriptionFacets = descFacets
564 }
565
566 result = append(result, community)
567 }
568
569 if err = rows.Err(); err != nil {
570 return nil, 0, fmt.Errorf("error iterating search results: %w", err)
571 }
572
573 return result, totalCount, nil
574}
575
576// Helper functions
577func nullString(s string) sql.NullString {
578 return sql.NullString{String: s, Valid: s != ""}
579}