A community based topic aggregation platform built on atproto
bretton.dev
1package communities
2
3import (
4 "Coves/internal/atproto/utils"
5 "bytes"
6 "context"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "io"
11 "log"
12 "net/http"
13 "regexp"
14 "strings"
15 "sync"
16 "time"
17)
18
19// Community handle validation regex (DNS-valid handle: name.community.instance.com)
20// Matches standard DNS hostname format (RFC 1035)
21var communityHandleRegex = regexp.MustCompile(`^([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$`)
22
23// DNS label validation (RFC 1035: 1-63 chars, alphanumeric + hyphen, can't start/end with hyphen)
24var dnsLabelRegex = regexp.MustCompile(`^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$`)
25
26// Domain validation (simplified - checks for valid DNS hostname structure)
27var domainRegex = regexp.MustCompile(`^([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)*[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$`)
28
29type communityService struct {
30 // Interfaces and pointers first (better alignment)
31 repo Repository
32 provisioner *PDSAccountProvisioner
33
34 // Token refresh concurrency control
35 // Each community gets its own mutex to prevent concurrent refresh attempts
36 refreshMutexes map[string]*sync.Mutex
37
38 // Strings
39 pdsURL string
40 instanceDID string
41 instanceDomain string
42 pdsAccessToken string
43
44 // Sync primitives last
45 mapMutex sync.RWMutex // Protects refreshMutexes map itself
46}
47
48const (
49 // Maximum recommended size for mutex cache (warning threshold, not hard limit)
50 // At 10,000 entries × 16 bytes = ~160KB memory (negligible overhead)
51 // Map can grow larger in production - even 100,000 entries = 1.6MB is acceptable
52 maxMutexCacheSize = 10000
53)
54
55// NewCommunityService creates a new community service
56func NewCommunityService(repo Repository, pdsURL, instanceDID, instanceDomain string, provisioner *PDSAccountProvisioner) Service {
57 // SECURITY: Basic validation that did:web domain matches configured instanceDomain
58 // This catches honest configuration mistakes but NOT malicious code modifications
59 // Full verification (Phase 2) requires fetching DID document from domain
60 // See: docs/PRD_BACKLOG.md - "did:web Domain Verification"
61 if strings.HasPrefix(instanceDID, "did:web:") {
62 didDomain := strings.TrimPrefix(instanceDID, "did:web:")
63 if didDomain != instanceDomain {
64 log.Printf("⚠️ SECURITY WARNING: Instance DID domain (%s) doesn't match configured domain (%s)",
65 didDomain, instanceDomain)
66 log.Printf(" This could indicate a configuration error or potential domain spoofing attempt")
67 log.Printf(" Communities will be hosted by: %s", instanceDID)
68 }
69 }
70
71 return &communityService{
72 repo: repo,
73 pdsURL: pdsURL,
74 instanceDID: instanceDID,
75 instanceDomain: instanceDomain,
76 provisioner: provisioner,
77 refreshMutexes: make(map[string]*sync.Mutex),
78 }
79}
80
81// SetPDSAccessToken sets the PDS access token for authentication
82// This should be called after creating a session for the Coves instance DID on the PDS
83func (s *communityService) SetPDSAccessToken(token string) {
84 s.pdsAccessToken = token
85}
86
87// CreateCommunity creates a new community via write-forward to PDS
88// V2 Flow:
89// 1. Service creates PDS account for community (PDS generates signing keypair)
90// 2. Service writes community profile to COMMUNITY's own repository
91// 3. Firehose emits event
92// 4. Consumer indexes to AppView DB
93//
94// V2 Architecture:
95// - Community owns its own repository (at://community_did/social.coves.community.profile/self)
96// - PDS manages the signing keypair (we never see it)
97// - We store PDS credentials to act on behalf of the community
98// - Community can migrate to other instances (future V2.1 with rotation keys)
99func (s *communityService) CreateCommunity(ctx context.Context, req CreateCommunityRequest) (*Community, error) {
100 // Apply defaults before validation
101 if req.Visibility == "" {
102 req.Visibility = "public"
103 }
104
105 // SECURITY: Auto-populate hostedByDID from instance configuration
106 // Clients MUST NOT provide this field - it's derived from the instance receiving the request
107 // This prevents malicious instances from claiming to host communities for domains they don't own
108 req.HostedByDID = s.instanceDID
109
110 // Validate request
111 if err := s.validateCreateRequest(req); err != nil {
112 return nil, err
113 }
114
115 // V2: Provision a real PDS account for this community
116 // This calls com.atproto.server.createAccount internally
117 // The PDS will:
118 // 1. Generate a signing keypair (stored in PDS, we never see it)
119 // 2. Create a DID (did:plc:xxx)
120 // 3. Return credentials (DID, tokens)
121 pdsAccount, err := s.provisioner.ProvisionCommunityAccount(ctx, req.Name)
122 if err != nil {
123 return nil, fmt.Errorf("failed to provision PDS account for community: %w", err)
124 }
125
126 // Validate the atProto handle
127 if validateErr := s.ValidateHandle(pdsAccount.Handle); validateErr != nil {
128 return nil, fmt.Errorf("generated atProto handle is invalid: %w", validateErr)
129 }
130
131 // Build community profile record
132 profile := map[string]interface{}{
133 "$type": "social.coves.community.profile",
134 "handle": pdsAccount.Handle, // atProto handle (e.g., gaming.community.coves.social)
135 "name": req.Name, // Short name for !mentions (e.g., "gaming")
136 "visibility": req.Visibility,
137 "hostedBy": s.instanceDID, // V2: Instance hosts, community owns
138 "createdBy": req.CreatedByDID,
139 "createdAt": time.Now().Format(time.RFC3339),
140 "federation": map[string]interface{}{
141 "allowExternalDiscovery": req.AllowExternalDiscovery,
142 },
143 }
144
145 // Add optional fields
146 if req.DisplayName != "" {
147 profile["displayName"] = req.DisplayName
148 }
149 if req.Description != "" {
150 profile["description"] = req.Description
151 }
152 if len(req.Rules) > 0 {
153 profile["rules"] = req.Rules
154 }
155 if len(req.Categories) > 0 {
156 profile["categories"] = req.Categories
157 }
158 if req.Language != "" {
159 profile["language"] = req.Language
160 }
161
162 // Initialize counts
163 profile["memberCount"] = 0
164 profile["subscriberCount"] = 0
165
166 // TODO: Handle avatar and banner blobs
167 // For now, we'll skip blob uploads. This would require:
168 // 1. Upload blob to PDS via com.atproto.repo.uploadBlob
169 // 2. Get blob ref (CID)
170 // 3. Add to profile record
171
172 // V2: Write to COMMUNITY's own repository (not instance repo!)
173 // Repository: at://COMMUNITY_DID/social.coves.community.profile/self
174 // Authenticate using community's access token
175 recordURI, recordCID, err := s.createRecordOnPDSAs(
176 ctx,
177 pdsAccount.DID, // repo = community's DID (community owns its repo!)
178 "social.coves.community.profile",
179 "self", // canonical rkey for profile
180 profile,
181 pdsAccount.AccessToken, // authenticate as the community
182 )
183 if err != nil {
184 return nil, fmt.Errorf("failed to create community profile record: %w", err)
185 }
186
187 // Build Community object with PDS credentials AND cryptographic keys
188 community := &Community{
189 DID: pdsAccount.DID, // Community's DID (owns the repo!)
190 Handle: pdsAccount.Handle, // atProto handle (e.g., gaming.community.coves.social)
191 Name: req.Name,
192 DisplayName: req.DisplayName,
193 Description: req.Description,
194 OwnerDID: pdsAccount.DID, // V2: Community owns itself
195 CreatedByDID: req.CreatedByDID,
196 HostedByDID: req.HostedByDID,
197 PDSEmail: pdsAccount.Email,
198 PDSPassword: pdsAccount.Password,
199 PDSAccessToken: pdsAccount.AccessToken,
200 PDSRefreshToken: pdsAccount.RefreshToken,
201 PDSURL: pdsAccount.PDSURL,
202 Visibility: req.Visibility,
203 AllowExternalDiscovery: req.AllowExternalDiscovery,
204 MemberCount: 0,
205 SubscriberCount: 0,
206 CreatedAt: time.Now(),
207 UpdatedAt: time.Now(),
208 RecordURI: recordURI,
209 RecordCID: recordCID,
210 // V2: Cryptographic keys for portability (will be encrypted by repository)
211 RotationKeyPEM: pdsAccount.RotationKeyPEM, // CRITICAL: Enables DID migration
212 SigningKeyPEM: pdsAccount.SigningKeyPEM, // For atproto operations
213 }
214
215 // CRITICAL: Persist PDS credentials immediately to database
216 // The Jetstream consumer will eventually index the community profile from the firehose,
217 // but it won't have the PDS credentials. We must store them now so we can:
218 // 1. Update the community profile later (using its own credentials)
219 // 2. Re-authenticate if access tokens expire
220 _, err = s.repo.Create(ctx, community)
221 if err != nil {
222 return nil, fmt.Errorf("failed to persist community with credentials: %w", err)
223 }
224
225 return community, nil
226}
227
228// GetCommunity retrieves a community from AppView DB
229// identifier can be either a DID or handle
230func (s *communityService) GetCommunity(ctx context.Context, identifier string) (*Community, error) {
231 if identifier == "" {
232 return nil, ErrInvalidInput
233 }
234
235 // Determine if identifier is DID or handle
236 if strings.HasPrefix(identifier, "did:") {
237 return s.repo.GetByDID(ctx, identifier)
238 }
239
240 if strings.HasPrefix(identifier, "!") {
241 return s.repo.GetByHandle(ctx, identifier)
242 }
243
244 return nil, NewValidationError("identifier", "must be a DID or handle")
245}
246
247// GetByDID retrieves a community by its DID
248// Exported for use by post service when validating community references
249func (s *communityService) GetByDID(ctx context.Context, did string) (*Community, error) {
250 if did == "" {
251 return nil, ErrInvalidInput
252 }
253
254 if !strings.HasPrefix(did, "did:") {
255 return nil, NewValidationError("did", "must be a valid DID")
256 }
257
258 return s.repo.GetByDID(ctx, did)
259}
260
261// UpdateCommunity updates a community via write-forward to PDS
262func (s *communityService) UpdateCommunity(ctx context.Context, req UpdateCommunityRequest) (*Community, error) {
263 if req.CommunityDID == "" {
264 return nil, NewValidationError("communityDid", "required")
265 }
266
267 if req.UpdatedByDID == "" {
268 return nil, NewValidationError("updatedByDid", "required")
269 }
270
271 // Get existing community
272 existing, err := s.repo.GetByDID(ctx, req.CommunityDID)
273 if err != nil {
274 return nil, err
275 }
276
277 // CRITICAL: Ensure fresh PDS access token before write operation
278 // Community PDS tokens expire every ~2 hours and must be refreshed
279 existing, err = s.EnsureFreshToken(ctx, existing)
280 if err != nil {
281 return nil, fmt.Errorf("failed to ensure fresh credentials: %w", err)
282 }
283
284 // Authorization: verify user is the creator
285 // TODO(Communities-Auth): Add moderator check when moderation system is implemented
286 if existing.CreatedByDID != req.UpdatedByDID {
287 return nil, ErrUnauthorized
288 }
289
290 // Build updated profile record (start with existing)
291 profile := map[string]interface{}{
292 "$type": "social.coves.community.profile",
293 "handle": existing.Handle,
294 "name": existing.Name,
295 "owner": existing.OwnerDID,
296 "createdBy": existing.CreatedByDID,
297 "hostedBy": existing.HostedByDID,
298 "createdAt": existing.CreatedAt.Format(time.RFC3339),
299 }
300
301 // Apply updates
302 if req.DisplayName != nil {
303 profile["displayName"] = *req.DisplayName
304 } else {
305 profile["displayName"] = existing.DisplayName
306 }
307
308 if req.Description != nil {
309 profile["description"] = *req.Description
310 } else {
311 profile["description"] = existing.Description
312 }
313
314 if req.Visibility != nil {
315 profile["visibility"] = *req.Visibility
316 } else {
317 profile["visibility"] = existing.Visibility
318 }
319
320 if req.AllowExternalDiscovery != nil {
321 profile["federation"] = map[string]interface{}{
322 "allowExternalDiscovery": *req.AllowExternalDiscovery,
323 }
324 } else {
325 profile["federation"] = map[string]interface{}{
326 "allowExternalDiscovery": existing.AllowExternalDiscovery,
327 }
328 }
329
330 // Preserve moderation settings (even if empty)
331 // These fields are optional but should not be erased on update
332 if req.ModerationType != nil {
333 profile["moderationType"] = *req.ModerationType
334 } else if existing.ModerationType != "" {
335 profile["moderationType"] = existing.ModerationType
336 }
337
338 if len(req.ContentWarnings) > 0 {
339 profile["contentWarnings"] = req.ContentWarnings
340 } else if len(existing.ContentWarnings) > 0 {
341 profile["contentWarnings"] = existing.ContentWarnings
342 }
343
344 // Preserve counts
345 profile["memberCount"] = existing.MemberCount
346 profile["subscriberCount"] = existing.SubscriberCount
347
348 // V2: Community profiles always use "self" as rkey
349 // (No need to extract from URI - it's always "self" for V2 communities)
350
351 // V2 CRITICAL FIX: Write-forward using COMMUNITY's own DID and credentials
352 // Repository: at://COMMUNITY_DID/social.coves.community.profile/self
353 // Authenticate as the community (not as instance!)
354 if existing.PDSAccessToken == "" {
355 return nil, fmt.Errorf("community %s missing PDS credentials - cannot update", existing.DID)
356 }
357
358 recordURI, recordCID, err := s.putRecordOnPDSAs(
359 ctx,
360 existing.DID, // repo = community's own DID (V2!)
361 "social.coves.community.profile",
362 "self", // V2: always "self"
363 profile,
364 existing.PDSAccessToken, // authenticate as the community
365 )
366 if err != nil {
367 return nil, fmt.Errorf("failed to update community on PDS: %w", err)
368 }
369
370 // Return updated community representation
371 // Actual AppView DB update happens via Jetstream consumer
372 updated := *existing
373 if req.DisplayName != nil {
374 updated.DisplayName = *req.DisplayName
375 }
376 if req.Description != nil {
377 updated.Description = *req.Description
378 }
379 if req.Visibility != nil {
380 updated.Visibility = *req.Visibility
381 }
382 if req.AllowExternalDiscovery != nil {
383 updated.AllowExternalDiscovery = *req.AllowExternalDiscovery
384 }
385 if req.ModerationType != nil {
386 updated.ModerationType = *req.ModerationType
387 }
388 if len(req.ContentWarnings) > 0 {
389 updated.ContentWarnings = req.ContentWarnings
390 }
391 updated.RecordURI = recordURI
392 updated.RecordCID = recordCID
393 updated.UpdatedAt = time.Now()
394
395 return &updated, nil
396}
397
398// getOrCreateRefreshMutex returns a mutex for the given community DID
399// Thread-safe with read-lock fast path for existing entries
400// SAFETY: Does NOT evict entries to avoid race condition where:
401// 1. Thread A holds mutex for community-123
402// 2. Thread B evicts community-123 from map
403// 3. Thread C creates NEW mutex for community-123
404// 4. Now two threads can refresh community-123 concurrently (mutex defeated!)
405func (s *communityService) getOrCreateRefreshMutex(did string) *sync.Mutex {
406 // Fast path: check if mutex already exists (read lock)
407 s.mapMutex.RLock()
408 mutex, exists := s.refreshMutexes[did]
409 s.mapMutex.RUnlock()
410
411 if exists {
412 return mutex
413 }
414
415 // Slow path: create new mutex (write lock)
416 s.mapMutex.Lock()
417 defer s.mapMutex.Unlock()
418
419 // Double-check after acquiring write lock (another goroutine might have created it)
420 mutex, exists = s.refreshMutexes[did]
421 if exists {
422 return mutex
423 }
424
425 // Create new mutex
426 mutex = &sync.Mutex{}
427 s.refreshMutexes[did] = mutex
428
429 // SAFETY: No eviction to prevent race condition
430 // Map will grow beyond maxMutexCacheSize but this is safer than evicting in-use mutexes
431 if len(s.refreshMutexes) > maxMutexCacheSize {
432 memoryKB := len(s.refreshMutexes) * 16 / 1024
433 log.Printf("[TOKEN-REFRESH] WARN: Mutex cache size (%d) exceeds recommended limit (%d) - this is safe but may indicate high community churn. Memory usage: ~%d KB",
434 len(s.refreshMutexes), maxMutexCacheSize, memoryKB)
435 }
436
437 return mutex
438}
439
440// ensureFreshToken checks if a community's access token needs refresh and updates if needed
441// Returns updated community with fresh credentials (or original if no refresh needed)
442// Thread-safe: Uses per-community mutex to prevent concurrent refresh attempts
443// EnsureFreshToken ensures the community's PDS access token is valid
444// Exported for use by post service when writing posts to community repos
445func (s *communityService) EnsureFreshToken(ctx context.Context, community *Community) (*Community, error) {
446 // Get or create mutex for this specific community DID
447 mutex := s.getOrCreateRefreshMutex(community.DID)
448
449 // Lock for this specific community (allows other communities to refresh concurrently)
450 mutex.Lock()
451 defer mutex.Unlock()
452
453 // Re-fetch community from DB (another goroutine might have already refreshed it)
454 fresh, err := s.repo.GetByDID(ctx, community.DID)
455 if err != nil {
456 return nil, fmt.Errorf("failed to re-fetch community: %w", err)
457 }
458
459 // Check if token needs refresh (5-minute buffer before expiration)
460 needsRefresh, err := NeedsRefresh(fresh.PDSAccessToken)
461 if err != nil {
462 log.Printf("[TOKEN-REFRESH] Community: %s, Event: token_parse_failed, Error: %v", fresh.DID, err)
463 return nil, fmt.Errorf("failed to check token expiration: %w", err)
464 }
465
466 if !needsRefresh {
467 // Token still valid, no refresh needed
468 return fresh, nil
469 }
470
471 log.Printf("[TOKEN-REFRESH] Community: %s, Event: token_refresh_started, Message: Access token expiring soon", fresh.DID)
472
473 // Attempt token refresh using refresh token
474 newAccessToken, newRefreshToken, err := refreshPDSToken(ctx, fresh.PDSURL, fresh.PDSAccessToken, fresh.PDSRefreshToken)
475 if err != nil {
476 // Check if refresh token expired (need password fallback)
477 if strings.Contains(err.Error(), "expired or invalid") {
478 log.Printf("[TOKEN-REFRESH] Community: %s, Event: refresh_token_expired, Message: Re-authenticating with password", fresh.DID)
479
480 // Fallback: Re-authenticate with stored password
481 newAccessToken, newRefreshToken, err = reauthenticateWithPassword(
482 ctx,
483 fresh.PDSURL,
484 fresh.PDSEmail,
485 fresh.PDSPassword, // Retrieved decrypted from DB
486 )
487 if err != nil {
488 log.Printf("[TOKEN-REFRESH] Community: %s, Event: password_auth_failed, Error: %v", fresh.DID, err)
489 return nil, fmt.Errorf("failed to re-authenticate community: %w", err)
490 }
491
492 log.Printf("[TOKEN-REFRESH] Community: %s, Event: password_fallback_success, Message: Re-authenticated after refresh token expiry", fresh.DID)
493 } else {
494 log.Printf("[TOKEN-REFRESH] Community: %s, Event: refresh_failed, Error: %v", fresh.DID, err)
495 return nil, fmt.Errorf("failed to refresh token: %w", err)
496 }
497 }
498
499 // CRITICAL: Update database with new tokens immediately
500 // Refresh tokens are SINGLE-USE - old one is now invalid
501 // Use retry logic to handle transient DB failures
502 const maxRetries = 3
503 var updateErr error
504 for attempt := 0; attempt < maxRetries; attempt++ {
505 updateErr = s.repo.UpdateCredentials(ctx, fresh.DID, newAccessToken, newRefreshToken)
506 if updateErr == nil {
507 break // Success
508 }
509
510 log.Printf("[TOKEN-REFRESH] Community: %s, Event: db_update_retry, Attempt: %d/%d, Error: %v",
511 fresh.DID, attempt+1, maxRetries, updateErr)
512
513 if attempt < maxRetries-1 {
514 // Exponential backoff: 100ms, 200ms, 400ms
515 backoff := time.Duration(1<<attempt) * 100 * time.Millisecond
516 time.Sleep(backoff)
517 }
518 }
519
520 if updateErr != nil {
521 // CRITICAL: Community is now locked out - old refresh token invalid, new one not saved
522 log.Printf("[TOKEN-REFRESH] CRITICAL: Community %s LOCKED OUT - failed to persist credentials after %d retries: %v",
523 fresh.DID, maxRetries, updateErr)
524 // TODO: Send alert to monitoring system (add in Beta)
525 return nil, fmt.Errorf("failed to persist refreshed credentials after %d retries (COMMUNITY LOCKED OUT): %w",
526 maxRetries, updateErr)
527 }
528
529 // Return updated community object with fresh tokens
530 updatedCommunity := *fresh
531 updatedCommunity.PDSAccessToken = newAccessToken
532 updatedCommunity.PDSRefreshToken = newRefreshToken
533
534 log.Printf("[TOKEN-REFRESH] Community: %s, Event: token_refreshed, Message: Access token refreshed successfully", fresh.DID)
535
536 return &updatedCommunity, nil
537}
538
539// ListCommunities queries AppView DB for communities with filters
540func (s *communityService) ListCommunities(ctx context.Context, req ListCommunitiesRequest) ([]*Community, int, error) {
541 // Set defaults
542 if req.Limit <= 0 || req.Limit > 100 {
543 req.Limit = 50
544 }
545
546 return s.repo.List(ctx, req)
547}
548
549// SearchCommunities performs fuzzy search in AppView DB
550func (s *communityService) SearchCommunities(ctx context.Context, req SearchCommunitiesRequest) ([]*Community, int, error) {
551 if req.Query == "" {
552 return nil, 0, NewValidationError("query", "search query is required")
553 }
554
555 // Set defaults
556 if req.Limit <= 0 || req.Limit > 100 {
557 req.Limit = 50
558 }
559
560 return s.repo.Search(ctx, req)
561}
562
563// SubscribeToCommunity creates a subscription via write-forward to PDS
564func (s *communityService) SubscribeToCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string, contentVisibility int) (*Subscription, error) {
565 if userDID == "" {
566 return nil, NewValidationError("userDid", "required")
567 }
568 if userAccessToken == "" {
569 return nil, NewValidationError("userAccessToken", "required")
570 }
571
572 // Clamp contentVisibility to valid range (1-5), default to 3 if 0 or invalid
573 if contentVisibility <= 0 || contentVisibility > 5 {
574 contentVisibility = 3
575 }
576
577 // Resolve community identifier to DID
578 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
579 if err != nil {
580 return nil, err
581 }
582
583 // Verify community exists
584 community, err := s.repo.GetByDID(ctx, communityDID)
585 if err != nil {
586 return nil, err
587 }
588
589 // Check visibility - can't subscribe to private communities without invitation (TODO)
590 if community.Visibility == "private" {
591 return nil, ErrUnauthorized
592 }
593
594 // Build subscription record
595 // CRITICAL: Collection is social.coves.community.subscription (RECORD TYPE), not social.coves.community.subscribe (XRPC procedure)
596 // This record will be created in the USER's repository: at://user_did/social.coves.community.subscription/{tid}
597 // Following atProto conventions, we use "subject" field to reference the community
598 subRecord := map[string]interface{}{
599 "$type": "social.coves.community.subscription",
600 "subject": communityDID, // atProto convention: "subject" for entity references
601 "createdAt": time.Now().Format(time.RFC3339),
602 "contentVisibility": contentVisibility,
603 }
604
605 // Write-forward: create subscription record in user's repo using their access token
606 // The collection parameter refers to the record type in the repository
607 recordURI, recordCID, err := s.createRecordOnPDSAs(ctx, userDID, "social.coves.community.subscription", "", subRecord, userAccessToken)
608 if err != nil {
609 return nil, fmt.Errorf("failed to create subscription on PDS: %w", err)
610 }
611
612 // Return subscription representation
613 subscription := &Subscription{
614 UserDID: userDID,
615 CommunityDID: communityDID,
616 ContentVisibility: contentVisibility,
617 SubscribedAt: time.Now(),
618 RecordURI: recordURI,
619 RecordCID: recordCID,
620 }
621
622 return subscription, nil
623}
624
625// UnsubscribeFromCommunity removes a subscription via PDS delete
626func (s *communityService) UnsubscribeFromCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string) error {
627 if userDID == "" {
628 return NewValidationError("userDid", "required")
629 }
630 if userAccessToken == "" {
631 return NewValidationError("userAccessToken", "required")
632 }
633
634 // Resolve community identifier
635 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
636 if err != nil {
637 return err
638 }
639
640 // Get the subscription from AppView to find the record key
641 subscription, err := s.repo.GetSubscription(ctx, userDID, communityDID)
642 if err != nil {
643 return err
644 }
645
646 // Extract rkey from record URI (at://did/collection/rkey)
647 rkey := utils.ExtractRKeyFromURI(subscription.RecordURI)
648 if rkey == "" {
649 return fmt.Errorf("invalid subscription record URI")
650 }
651
652 // Write-forward: delete record from PDS using user's access token
653 // CRITICAL: Delete from social.coves.community.subscription (RECORD TYPE), not social.coves.community.unsubscribe
654 if err := s.deleteRecordOnPDSAs(ctx, userDID, "social.coves.community.subscription", rkey, userAccessToken); err != nil {
655 return fmt.Errorf("failed to delete subscription on PDS: %w", err)
656 }
657
658 return nil
659}
660
661// GetUserSubscriptions queries AppView DB for user's subscriptions
662func (s *communityService) GetUserSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*Subscription, error) {
663 if limit <= 0 || limit > 100 {
664 limit = 50
665 }
666
667 return s.repo.ListSubscriptions(ctx, userDID, limit, offset)
668}
669
670// GetCommunitySubscribers queries AppView DB for community subscribers
671func (s *communityService) GetCommunitySubscribers(ctx context.Context, communityIdentifier string, limit, offset int) ([]*Subscription, error) {
672 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
673 if err != nil {
674 return nil, err
675 }
676
677 if limit <= 0 || limit > 100 {
678 limit = 50
679 }
680
681 return s.repo.ListSubscribers(ctx, communityDID, limit, offset)
682}
683
684// GetMembership retrieves membership info from AppView DB
685func (s *communityService) GetMembership(ctx context.Context, userDID, communityIdentifier string) (*Membership, error) {
686 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
687 if err != nil {
688 return nil, err
689 }
690
691 return s.repo.GetMembership(ctx, userDID, communityDID)
692}
693
694// ListCommunityMembers queries AppView DB for members
695func (s *communityService) ListCommunityMembers(ctx context.Context, communityIdentifier string, limit, offset int) ([]*Membership, error) {
696 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
697 if err != nil {
698 return nil, err
699 }
700
701 if limit <= 0 || limit > 100 {
702 limit = 50
703 }
704
705 return s.repo.ListMembers(ctx, communityDID, limit, offset)
706}
707
708// BlockCommunity blocks a community via write-forward to PDS
709func (s *communityService) BlockCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string) (*CommunityBlock, error) {
710 if userDID == "" {
711 return nil, NewValidationError("userDid", "required")
712 }
713 if userAccessToken == "" {
714 return nil, NewValidationError("userAccessToken", "required")
715 }
716
717 // Resolve community identifier (also verifies community exists)
718 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
719 if err != nil {
720 return nil, err
721 }
722
723 // Build block record
724 // CRITICAL: Collection is social.coves.community.block (RECORD TYPE)
725 // This record will be created in the USER's repository: at://user_did/social.coves.community.block/{tid}
726 // Following atProto conventions and Bluesky's app.bsky.graph.block pattern
727 blockRecord := map[string]interface{}{
728 "$type": "social.coves.community.block",
729 "subject": communityDID, // DID of community being blocked
730 "createdAt": time.Now().Format(time.RFC3339),
731 }
732
733 // Write-forward: create block record in user's repo using their access token
734 // Note: We don't check for existing blocks first because:
735 // 1. The PDS may reject duplicates (depending on implementation)
736 // 2. The repository layer handles idempotency with ON CONFLICT DO NOTHING
737 // 3. This avoids a race condition where two concurrent requests both pass the check
738 recordURI, recordCID, err := s.createRecordOnPDSAs(ctx, userDID, "social.coves.community.block", "", blockRecord, userAccessToken)
739 if err != nil {
740 // Check if this is a duplicate/conflict error from PDS
741 // PDS should return 409 Conflict for duplicate records, but we also check common error messages
742 // for compatibility with different PDS implementations
743 errMsg := err.Error()
744 isDuplicate := strings.Contains(errMsg, "status 409") || // HTTP 409 Conflict
745 strings.Contains(errMsg, "duplicate") ||
746 strings.Contains(errMsg, "already exists") ||
747 strings.Contains(errMsg, "AlreadyExists")
748
749 if isDuplicate {
750 // Fetch and return existing block from our indexed view
751 existingBlock, getErr := s.repo.GetBlock(ctx, userDID, communityDID)
752 if getErr == nil {
753 // Block exists in our index - return it
754 return existingBlock, nil
755 }
756 // Only treat as "already exists" if the error is ErrBlockNotFound (race condition)
757 // Any other error (DB outage, connection failure, etc.) should bubble up
758 if errors.Is(getErr, ErrBlockNotFound) {
759 // Race condition: PDS has the block but Jetstream hasn't indexed it yet
760 // Return typed conflict error so handler can return 409 instead of 500
761 // This is normal in eventually-consistent systems
762 return nil, ErrBlockAlreadyExists
763 }
764 // Real datastore error - bubble it up so operators see the failure
765 return nil, fmt.Errorf("PDS reported duplicate block but failed to fetch from index: %w", getErr)
766 }
767 return nil, fmt.Errorf("failed to create block on PDS: %w", err)
768 }
769
770 // Return block representation
771 block := &CommunityBlock{
772 UserDID: userDID,
773 CommunityDID: communityDID,
774 BlockedAt: time.Now(),
775 RecordURI: recordURI,
776 RecordCID: recordCID,
777 }
778
779 return block, nil
780}
781
782// UnblockCommunity removes a block via PDS delete
783func (s *communityService) UnblockCommunity(ctx context.Context, userDID, userAccessToken, communityIdentifier string) error {
784 if userDID == "" {
785 return NewValidationError("userDid", "required")
786 }
787 if userAccessToken == "" {
788 return NewValidationError("userAccessToken", "required")
789 }
790
791 // Resolve community identifier
792 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
793 if err != nil {
794 return err
795 }
796
797 // Get the block from AppView to find the record key
798 block, err := s.repo.GetBlock(ctx, userDID, communityDID)
799 if err != nil {
800 return err
801 }
802
803 // Extract rkey from record URI (at://did/collection/rkey)
804 rkey := utils.ExtractRKeyFromURI(block.RecordURI)
805 if rkey == "" {
806 return fmt.Errorf("invalid block record URI")
807 }
808
809 // Write-forward: delete record from PDS using user's access token
810 if err := s.deleteRecordOnPDSAs(ctx, userDID, "social.coves.community.block", rkey, userAccessToken); err != nil {
811 return fmt.Errorf("failed to delete block on PDS: %w", err)
812 }
813
814 return nil
815}
816
817// GetBlockedCommunities queries AppView DB for user's blocks
818func (s *communityService) GetBlockedCommunities(ctx context.Context, userDID string, limit, offset int) ([]*CommunityBlock, error) {
819 if limit <= 0 || limit > 100 {
820 limit = 50
821 }
822
823 return s.repo.ListBlockedCommunities(ctx, userDID, limit, offset)
824}
825
826// IsBlocked checks if a user has blocked a community
827func (s *communityService) IsBlocked(ctx context.Context, userDID, communityIdentifier string) (bool, error) {
828 communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
829 if err != nil {
830 return false, err
831 }
832
833 return s.repo.IsBlocked(ctx, userDID, communityDID)
834}
835
836// ValidateHandle checks if a community handle is valid
837func (s *communityService) ValidateHandle(handle string) error {
838 if handle == "" {
839 return NewValidationError("handle", "required")
840 }
841
842 if !communityHandleRegex.MatchString(handle) {
843 return ErrInvalidHandle
844 }
845
846 return nil
847}
848
849// ResolveCommunityIdentifier converts a community identifier to a DID
850// Following Bluesky's pattern with Coves extensions:
851//
852// Accepts (like Bluesky's at-identifier):
853// 1. DID: did:plc:abc123 (pass through)
854// 2. Canonical handle: gardening.community.coves.social (atProto standard)
855// 3. At-identifier: @gardening.community.coves.social (strip @ prefix)
856//
857// Coves-specific extensions:
858// 4. Scoped format: !gardening@coves.social (parse and resolve)
859//
860// Returns: DID string
861func (s *communityService) ResolveCommunityIdentifier(ctx context.Context, identifier string) (string, error) {
862 identifier = strings.TrimSpace(identifier)
863
864 if identifier == "" {
865 return "", ErrInvalidInput
866 }
867
868 // 1. DID - verify it exists and return (Bluesky standard)
869 if strings.HasPrefix(identifier, "did:") {
870 _, err := s.repo.GetByDID(ctx, identifier)
871 if err != nil {
872 if IsNotFound(err) {
873 return "", fmt.Errorf("community not found for DID %s: %w", identifier, err)
874 }
875 return "", fmt.Errorf("failed to verify community DID %s: %w", identifier, err)
876 }
877 return identifier, nil
878 }
879
880 // 2. Scoped format: !name@instance (Coves-specific)
881 if strings.HasPrefix(identifier, "!") {
882 return s.resolveScopedIdentifier(ctx, identifier)
883 }
884
885 // 3. At-identifier format: @handle (Bluesky standard - strip @ prefix)
886 identifier = strings.TrimPrefix(identifier, "@")
887
888 // 4. Canonical handle: name.community.instance.com (Bluesky standard)
889 if strings.Contains(identifier, ".") {
890 community, err := s.repo.GetByHandle(ctx, strings.ToLower(identifier))
891 if err != nil {
892 return "", fmt.Errorf("community not found for handle %s: %w", identifier, err)
893 }
894 return community.DID, nil
895 }
896
897 return "", NewValidationError("identifier", "must be a DID, handle, or scoped identifier (!name@instance)")
898}
899
900// resolveScopedIdentifier handles Coves-specific !name@instance format
901// Formats accepted:
902//
903// !gardening@coves.social -> gardening.community.coves.social
904func (s *communityService) resolveScopedIdentifier(ctx context.Context, scoped string) (string, error) {
905 // Remove ! prefix
906 scoped = strings.TrimPrefix(scoped, "!")
907
908 var name string
909 var instanceDomain string
910
911 // Parse !name@instance
912 if !strings.Contains(scoped, "@") {
913 return "", NewValidationError("identifier", "scoped identifier must include @ symbol (!name@instance)")
914 }
915
916 parts := strings.SplitN(scoped, "@", 2)
917 name = strings.TrimSpace(parts[0])
918 instanceDomain = strings.TrimSpace(parts[1])
919
920 // Validate name format
921 if name == "" {
922 return "", NewValidationError("identifier", "community name cannot be empty")
923 }
924
925 // Validate name is a valid DNS label (RFC 1035)
926 // Must be 1-63 chars, alphanumeric + hyphen, can't start/end with hyphen
927 if !isValidDNSLabel(name) {
928 return "", NewValidationError("identifier", "community name must be valid DNS label (alphanumeric and hyphens only, 1-63 chars, cannot start or end with hyphen)")
929 }
930
931 // Validate instance domain format
932 if !isValidDomain(instanceDomain) {
933 return "", NewValidationError("identifier", "invalid instance domain format")
934 }
935
936 // Normalize domain to lowercase (DNS is case-insensitive)
937 // This fixes the bug where !gardening@Coves.social would fail lookup
938 instanceDomain = strings.ToLower(instanceDomain)
939
940 // Validate the instance matches this server
941 if !s.isLocalInstance(instanceDomain) {
942 return "", NewValidationError("identifier",
943 fmt.Sprintf("community is not hosted on this instance (expected @%s)", s.instanceDomain))
944 }
945
946 // Construct canonical handle: {name}.community.{instanceDomain}
947 // Both name and instanceDomain are normalized to lowercase for consistent DB lookup
948 canonicalHandle := fmt.Sprintf("%s.community.%s",
949 strings.ToLower(name),
950 instanceDomain) // Already normalized to lowercase on line 923
951
952 // Look up by canonical handle
953 community, err := s.repo.GetByHandle(ctx, canonicalHandle)
954 if err != nil {
955 return "", fmt.Errorf("community not found for scoped identifier !%s@%s: %w", name, instanceDomain, err)
956 }
957
958 return community.DID, nil
959}
960
961// isLocalInstance checks if the provided domain matches this instance
962func (s *communityService) isLocalInstance(domain string) bool {
963 // Normalize both domains
964 domain = strings.ToLower(strings.TrimSpace(domain))
965 instanceDomain := strings.ToLower(s.instanceDomain)
966
967 // Direct match
968 return domain == instanceDomain
969}
970
971// Validation helpers
972
973// isValidDNSLabel validates that a string is a valid DNS label per RFC 1035
974// - 1-63 characters
975// - Alphanumeric and hyphens only
976// - Cannot start or end with hyphen
977func isValidDNSLabel(label string) bool {
978 return dnsLabelRegex.MatchString(label)
979}
980
981// isValidDomain validates that a string is a valid domain name
982// Simplified validation - checks basic DNS hostname structure
983func isValidDomain(domain string) bool {
984 if domain == "" || len(domain) > 253 {
985 return false
986 }
987 return domainRegex.MatchString(domain)
988}
989
990func (s *communityService) validateCreateRequest(req CreateCommunityRequest) error {
991 if req.Name == "" {
992 return NewValidationError("name", "required")
993 }
994
995 // DNS label limit: 63 characters per label
996 // Community handle format: {name}.community.{instanceDomain}
997 // The first label is just req.Name, so it must be <= 63 chars
998 if len(req.Name) > 63 {
999 return NewValidationError("name", "must be 63 characters or less (DNS label limit)")
1000 }
1001
1002 // Name can only contain alphanumeric and hyphens
1003 // Must start and end with alphanumeric (not hyphen)
1004 nameRegex := regexp.MustCompile(`^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$`)
1005 if !nameRegex.MatchString(req.Name) {
1006 return NewValidationError("name", "must contain only alphanumeric characters and hyphens")
1007 }
1008
1009 if req.Description != "" && len(req.Description) > 3000 {
1010 return NewValidationError("description", "must be 3000 characters or less")
1011 }
1012
1013 // Visibility should already be set with default in CreateCommunity
1014 if req.Visibility != "public" && req.Visibility != "unlisted" && req.Visibility != "private" {
1015 return ErrInvalidVisibility
1016 }
1017
1018 if req.CreatedByDID == "" {
1019 return NewValidationError("createdByDid", "required")
1020 }
1021
1022 // hostedByDID is auto-populated by the service layer, no validation needed
1023 // The handler ensures clients cannot provide this field
1024
1025 return nil
1026}
1027
1028// PDS write-forward helpers
1029
1030// createRecordOnPDSAs creates a record with a specific access token (for V2 community auth)
1031func (s *communityService) createRecordOnPDSAs(ctx context.Context, repoDID, collection, rkey string, record map[string]interface{}, accessToken string) (string, string, error) {
1032 endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.createRecord", strings.TrimSuffix(s.pdsURL, "/"))
1033
1034 payload := map[string]interface{}{
1035 "repo": repoDID,
1036 "collection": collection,
1037 "record": record,
1038 }
1039
1040 if rkey != "" {
1041 payload["rkey"] = rkey
1042 }
1043
1044 return s.callPDSWithAuth(ctx, "POST", endpoint, payload, accessToken)
1045}
1046
1047// putRecordOnPDSAs updates a record with a specific access token (for V2 community auth)
1048func (s *communityService) putRecordOnPDSAs(ctx context.Context, repoDID, collection, rkey string, record map[string]interface{}, accessToken string) (string, string, error) {
1049 endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.putRecord", strings.TrimSuffix(s.pdsURL, "/"))
1050
1051 payload := map[string]interface{}{
1052 "repo": repoDID,
1053 "collection": collection,
1054 "rkey": rkey,
1055 "record": record,
1056 }
1057
1058 return s.callPDSWithAuth(ctx, "POST", endpoint, payload, accessToken)
1059}
1060
1061// deleteRecordOnPDSAs deletes a record with a specific access token (for user-scoped deletions)
1062func (s *communityService) deleteRecordOnPDSAs(ctx context.Context, repoDID, collection, rkey, accessToken string) error {
1063 endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.deleteRecord", strings.TrimSuffix(s.pdsURL, "/"))
1064
1065 payload := map[string]interface{}{
1066 "repo": repoDID,
1067 "collection": collection,
1068 "rkey": rkey,
1069 }
1070
1071 _, _, err := s.callPDSWithAuth(ctx, "POST", endpoint, payload, accessToken)
1072 return err
1073}
1074
1075// callPDSWithAuth makes a PDS call with a specific access token (V2: for community authentication)
1076func (s *communityService) callPDSWithAuth(ctx context.Context, method, endpoint string, payload map[string]interface{}, accessToken string) (string, string, error) {
1077 jsonData, err := json.Marshal(payload)
1078 if err != nil {
1079 return "", "", fmt.Errorf("failed to marshal payload: %w", err)
1080 }
1081
1082 req, err := http.NewRequestWithContext(ctx, method, endpoint, bytes.NewBuffer(jsonData))
1083 if err != nil {
1084 return "", "", fmt.Errorf("failed to create request: %w", err)
1085 }
1086 req.Header.Set("Content-Type", "application/json")
1087
1088 // Add authentication with provided access token
1089 if accessToken != "" {
1090 req.Header.Set("Authorization", "Bearer "+accessToken)
1091 }
1092
1093 // Dynamic timeout based on operation type
1094 // Write operations (createAccount, createRecord, putRecord) are slower due to:
1095 // - Keypair generation
1096 // - DID PLC registration
1097 // - Database writes on PDS
1098 timeout := 10 * time.Second // Default for read operations
1099 if strings.Contains(endpoint, "createAccount") ||
1100 strings.Contains(endpoint, "createRecord") ||
1101 strings.Contains(endpoint, "putRecord") {
1102 timeout = 30 * time.Second // Extended timeout for write operations
1103 }
1104
1105 client := &http.Client{Timeout: timeout}
1106 resp, err := client.Do(req)
1107 if err != nil {
1108 return "", "", fmt.Errorf("failed to call PDS: %w", err)
1109 }
1110 defer func() {
1111 if closeErr := resp.Body.Close(); closeErr != nil {
1112 log.Printf("Failed to close response body: %v", closeErr)
1113 }
1114 }()
1115
1116 body, err := io.ReadAll(resp.Body)
1117 if err != nil {
1118 return "", "", fmt.Errorf("failed to read response: %w", err)
1119 }
1120
1121 if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
1122 return "", "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
1123 }
1124
1125 // Parse response to extract URI and CID
1126 var result struct {
1127 URI string `json:"uri"`
1128 CID string `json:"cid"`
1129 }
1130 if err := json.Unmarshal(body, &result); err != nil {
1131 // For delete operations, there might not be a response body
1132 if method == "POST" && strings.Contains(endpoint, "deleteRecord") {
1133 return "", "", nil
1134 }
1135 return "", "", fmt.Errorf("failed to parse PDS response: %w", err)
1136 }
1137
1138 return result.URI, result.CID, nil
1139}
1140
1141// Helper functions