···
14
+
"Coves/internal/atproto/did"
17
+
// Community handle validation regex (!name@instance)
18
+
var 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])?$`)
20
+
type communityService struct {
22
+
didGen *did.Generator
23
+
pdsURL string // PDS URL for write-forward operations
24
+
instanceDID string // DID of this Coves instance
25
+
pdsAccessToken string // Access token for authenticating to PDS as the instance
28
+
// NewCommunityService creates a new community service
29
+
func NewCommunityService(repo Repository, didGen *did.Generator, pdsURL string, instanceDID string) Service {
30
+
return &communityService{
34
+
instanceDID: instanceDID,
38
+
// SetPDSAccessToken sets the PDS access token for authentication
39
+
// This should be called after creating a session for the Coves instance DID on the PDS
40
+
func (s *communityService) SetPDSAccessToken(token string) {
41
+
s.pdsAccessToken = token
44
+
// CreateCommunity creates a new community via write-forward to PDS
45
+
// Flow: Service -> PDS (creates record) -> Firehose -> Consumer -> AppView DB
46
+
func (s *communityService) CreateCommunity(ctx context.Context, req CreateCommunityRequest) (*Community, error) {
47
+
// Apply defaults before validation
48
+
if req.Visibility == "" {
49
+
req.Visibility = "public"
53
+
if err := s.validateCreateRequest(req); err != nil {
57
+
// Generate a unique DID for the community
58
+
communityDID, err := s.didGen.GenerateCommunityDID()
60
+
return nil, fmt.Errorf("failed to generate community DID: %w", err)
63
+
// Build scoped handle: !{name}@{instance}
64
+
instanceDomain := extractDomain(s.instanceDID)
65
+
if instanceDomain == "" {
66
+
instanceDomain = "coves.local" // Fallback for testing
68
+
handle := fmt.Sprintf("!%s@%s", req.Name, instanceDomain)
70
+
// Validate the generated handle
71
+
if err := s.ValidateHandle(handle); err != nil {
72
+
return nil, fmt.Errorf("generated handle is invalid: %w", err)
75
+
// Build community profile record
76
+
profile := map[string]interface{}{
77
+
"$type": "social.coves.community.profile",
78
+
"did": communityDID, // Unique identifier for this community
81
+
"visibility": req.Visibility,
82
+
"owner": s.instanceDID, // V1: instance owns the community
83
+
"createdBy": req.CreatedByDID,
84
+
"hostedBy": req.HostedByDID,
85
+
"createdAt": time.Now().Format(time.RFC3339),
86
+
"federation": map[string]interface{}{
87
+
"allowExternalDiscovery": req.AllowExternalDiscovery,
91
+
// Add optional fields
92
+
if req.DisplayName != "" {
93
+
profile["displayName"] = req.DisplayName
95
+
if req.Description != "" {
96
+
profile["description"] = req.Description
98
+
if len(req.Rules) > 0 {
99
+
profile["rules"] = req.Rules
101
+
if len(req.Categories) > 0 {
102
+
profile["categories"] = req.Categories
104
+
if req.Language != "" {
105
+
profile["language"] = req.Language
108
+
// Initialize counts
109
+
profile["memberCount"] = 0
110
+
profile["subscriberCount"] = 0
112
+
// TODO: Handle avatar and banner blobs
113
+
// For now, we'll skip blob uploads. This would require:
114
+
// 1. Upload blob to PDS via com.atproto.repo.uploadBlob
115
+
// 2. Get blob ref (CID)
116
+
// 3. Add to profile record
118
+
// Write-forward to PDS: create the community profile record in the INSTANCE's repository
119
+
// The instance owns all community records, community DID is just metadata in the record
120
+
// Record will be at: at://INSTANCE_DID/social.coves.community.profile/COMMUNITY_RKEY
121
+
recordURI, recordCID, err := s.createRecordOnPDS(ctx, s.instanceDID, "social.coves.community.profile", "", profile)
123
+
return nil, fmt.Errorf("failed to create community on PDS: %w", err)
126
+
// Return a Community object representing what was created
127
+
// Note: This won't be in AppView DB until the Jetstream consumer processes it
128
+
community := &Community{
132
+
DisplayName: req.DisplayName,
133
+
Description: req.Description,
134
+
OwnerDID: s.instanceDID,
135
+
CreatedByDID: req.CreatedByDID,
136
+
HostedByDID: req.HostedByDID,
137
+
Visibility: req.Visibility,
138
+
AllowExternalDiscovery: req.AllowExternalDiscovery,
140
+
SubscriberCount: 0,
141
+
CreatedAt: time.Now(),
142
+
UpdatedAt: time.Now(),
143
+
RecordURI: recordURI,
144
+
RecordCID: recordCID,
147
+
return community, nil
150
+
// GetCommunity retrieves a community from AppView DB
151
+
// identifier can be either a DID or handle
152
+
func (s *communityService) GetCommunity(ctx context.Context, identifier string) (*Community, error) {
153
+
if identifier == "" {
154
+
return nil, ErrInvalidInput
157
+
// Determine if identifier is DID or handle
158
+
if strings.HasPrefix(identifier, "did:") {
159
+
return s.repo.GetByDID(ctx, identifier)
162
+
if strings.HasPrefix(identifier, "!") {
163
+
return s.repo.GetByHandle(ctx, identifier)
166
+
return nil, NewValidationError("identifier", "must be a DID or handle")
169
+
// UpdateCommunity updates a community via write-forward to PDS
170
+
func (s *communityService) UpdateCommunity(ctx context.Context, req UpdateCommunityRequest) (*Community, error) {
171
+
if req.CommunityDID == "" {
172
+
return nil, NewValidationError("communityDid", "required")
175
+
if req.UpdatedByDID == "" {
176
+
return nil, NewValidationError("updatedByDid", "required")
179
+
// Get existing community
180
+
existing, err := s.repo.GetByDID(ctx, req.CommunityDID)
185
+
// Authorization: verify user is the creator
186
+
// TODO(Communities-Auth): Add moderator check when moderation system is implemented
187
+
if existing.CreatedByDID != req.UpdatedByDID {
188
+
return nil, ErrUnauthorized
191
+
// Build updated profile record (start with existing)
192
+
profile := map[string]interface{}{
193
+
"$type": "social.coves.community.profile",
194
+
"handle": existing.Handle,
195
+
"name": existing.Name,
196
+
"owner": existing.OwnerDID,
197
+
"createdBy": existing.CreatedByDID,
198
+
"hostedBy": existing.HostedByDID,
199
+
"createdAt": existing.CreatedAt.Format(time.RFC3339),
203
+
if req.DisplayName != nil {
204
+
profile["displayName"] = *req.DisplayName
206
+
profile["displayName"] = existing.DisplayName
209
+
if req.Description != nil {
210
+
profile["description"] = *req.Description
212
+
profile["description"] = existing.Description
215
+
if req.Visibility != nil {
216
+
profile["visibility"] = *req.Visibility
218
+
profile["visibility"] = existing.Visibility
221
+
if req.AllowExternalDiscovery != nil {
222
+
profile["federation"] = map[string]interface{}{
223
+
"allowExternalDiscovery": *req.AllowExternalDiscovery,
226
+
profile["federation"] = map[string]interface{}{
227
+
"allowExternalDiscovery": existing.AllowExternalDiscovery,
231
+
if req.ModerationType != nil {
232
+
profile["moderationType"] = *req.ModerationType
235
+
if len(req.ContentWarnings) > 0 {
236
+
profile["contentWarnings"] = req.ContentWarnings
240
+
profile["memberCount"] = existing.MemberCount
241
+
profile["subscriberCount"] = existing.SubscriberCount
243
+
// Extract rkey from existing record URI (communities live in instance's repo)
244
+
rkey := extractRKeyFromURI(existing.RecordURI)
246
+
return nil, fmt.Errorf("invalid community record URI: %s", existing.RecordURI)
249
+
// Write-forward: update record on PDS using INSTANCE DID (communities are stored in instance repo)
250
+
recordURI, recordCID, err := s.putRecordOnPDS(ctx, s.instanceDID, "social.coves.community.profile", rkey, profile)
252
+
return nil, fmt.Errorf("failed to update community on PDS: %w", err)
255
+
// Return updated community representation
256
+
// Actual AppView DB update happens via Jetstream consumer
257
+
updated := *existing
258
+
if req.DisplayName != nil {
259
+
updated.DisplayName = *req.DisplayName
261
+
if req.Description != nil {
262
+
updated.Description = *req.Description
264
+
if req.Visibility != nil {
265
+
updated.Visibility = *req.Visibility
267
+
if req.AllowExternalDiscovery != nil {
268
+
updated.AllowExternalDiscovery = *req.AllowExternalDiscovery
270
+
if req.ModerationType != nil {
271
+
updated.ModerationType = *req.ModerationType
273
+
if len(req.ContentWarnings) > 0 {
274
+
updated.ContentWarnings = req.ContentWarnings
276
+
updated.RecordURI = recordURI
277
+
updated.RecordCID = recordCID
278
+
updated.UpdatedAt = time.Now()
280
+
return &updated, nil
283
+
// ListCommunities queries AppView DB for communities with filters
284
+
func (s *communityService) ListCommunities(ctx context.Context, req ListCommunitiesRequest) ([]*Community, int, error) {
286
+
if req.Limit <= 0 || req.Limit > 100 {
290
+
return s.repo.List(ctx, req)
293
+
// SearchCommunities performs fuzzy search in AppView DB
294
+
func (s *communityService) SearchCommunities(ctx context.Context, req SearchCommunitiesRequest) ([]*Community, int, error) {
295
+
if req.Query == "" {
296
+
return nil, 0, NewValidationError("query", "search query is required")
300
+
if req.Limit <= 0 || req.Limit > 100 {
304
+
return s.repo.Search(ctx, req)
307
+
// SubscribeToCommunity creates a subscription via write-forward to PDS
308
+
func (s *communityService) SubscribeToCommunity(ctx context.Context, userDID, communityIdentifier string) (*Subscription, error) {
310
+
return nil, NewValidationError("userDid", "required")
313
+
// Resolve community identifier to DID
314
+
communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
319
+
// Verify community exists
320
+
community, err := s.repo.GetByDID(ctx, communityDID)
325
+
// Check visibility - can't subscribe to private communities without invitation (TODO)
326
+
if community.Visibility == "private" {
327
+
return nil, ErrUnauthorized
330
+
// Build subscription record
331
+
subRecord := map[string]interface{}{
332
+
"$type": "social.coves.community.subscribe",
333
+
"community": communityDID,
336
+
// Write-forward: create subscription record in user's repo
337
+
recordURI, recordCID, err := s.createRecordOnPDS(ctx, userDID, "social.coves.community.subscribe", "", subRecord)
339
+
return nil, fmt.Errorf("failed to create subscription on PDS: %w", err)
342
+
// Return subscription representation
343
+
subscription := &Subscription{
345
+
CommunityDID: communityDID,
346
+
SubscribedAt: time.Now(),
347
+
RecordURI: recordURI,
348
+
RecordCID: recordCID,
351
+
return subscription, nil
354
+
// UnsubscribeFromCommunity removes a subscription via PDS delete
355
+
func (s *communityService) UnsubscribeFromCommunity(ctx context.Context, userDID, communityIdentifier string) error {
357
+
return NewValidationError("userDid", "required")
360
+
// Resolve community identifier
361
+
communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
366
+
// Get the subscription from AppView to find the record key
367
+
subscription, err := s.repo.GetSubscription(ctx, userDID, communityDID)
372
+
// Extract rkey from record URI (at://did/collection/rkey)
373
+
rkey := extractRKeyFromURI(subscription.RecordURI)
375
+
return fmt.Errorf("invalid subscription record URI")
378
+
// Write-forward: delete record from PDS
379
+
if err := s.deleteRecordOnPDS(ctx, userDID, "social.coves.community.subscribe", rkey); err != nil {
380
+
return fmt.Errorf("failed to delete subscription on PDS: %w", err)
386
+
// GetUserSubscriptions queries AppView DB for user's subscriptions
387
+
func (s *communityService) GetUserSubscriptions(ctx context.Context, userDID string, limit, offset int) ([]*Subscription, error) {
388
+
if limit <= 0 || limit > 100 {
392
+
return s.repo.ListSubscriptions(ctx, userDID, limit, offset)
395
+
// GetCommunitySubscribers queries AppView DB for community subscribers
396
+
func (s *communityService) GetCommunitySubscribers(ctx context.Context, communityIdentifier string, limit, offset int) ([]*Subscription, error) {
397
+
communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
402
+
if limit <= 0 || limit > 100 {
406
+
return s.repo.ListSubscribers(ctx, communityDID, limit, offset)
409
+
// GetMembership retrieves membership info from AppView DB
410
+
func (s *communityService) GetMembership(ctx context.Context, userDID, communityIdentifier string) (*Membership, error) {
411
+
communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
416
+
return s.repo.GetMembership(ctx, userDID, communityDID)
419
+
// ListCommunityMembers queries AppView DB for members
420
+
func (s *communityService) ListCommunityMembers(ctx context.Context, communityIdentifier string, limit, offset int) ([]*Membership, error) {
421
+
communityDID, err := s.ResolveCommunityIdentifier(ctx, communityIdentifier)
426
+
if limit <= 0 || limit > 100 {
430
+
return s.repo.ListMembers(ctx, communityDID, limit, offset)
433
+
// ValidateHandle checks if a community handle is valid
434
+
func (s *communityService) ValidateHandle(handle string) error {
436
+
return NewValidationError("handle", "required")
439
+
if !communityHandleRegex.MatchString(handle) {
440
+
return ErrInvalidHandle
446
+
// ResolveCommunityIdentifier converts a handle or DID to a DID
447
+
func (s *communityService) ResolveCommunityIdentifier(ctx context.Context, identifier string) (string, error) {
448
+
if identifier == "" {
449
+
return "", ErrInvalidInput
452
+
// If it's already a DID, return it
453
+
if strings.HasPrefix(identifier, "did:") {
454
+
return identifier, nil
457
+
// If it's a handle, look it up in AppView DB
458
+
if strings.HasPrefix(identifier, "!") {
459
+
community, err := s.repo.GetByHandle(ctx, identifier)
463
+
return community.DID, nil
466
+
return "", NewValidationError("identifier", "must be a DID or handle")
469
+
// Validation helpers
471
+
func (s *communityService) validateCreateRequest(req CreateCommunityRequest) error {
472
+
if req.Name == "" {
473
+
return NewValidationError("name", "required")
476
+
if len(req.Name) > 64 {
477
+
return NewValidationError("name", "must be 64 characters or less")
480
+
// Name can only contain alphanumeric and hyphens
481
+
nameRegex := regexp.MustCompile(`^[a-zA-Z0-9]([a-zA-Z0-9-]{0,62}[a-zA-Z0-9])?$`)
482
+
if !nameRegex.MatchString(req.Name) {
483
+
return NewValidationError("name", "must contain only alphanumeric characters and hyphens")
486
+
if req.Description != "" && len(req.Description) > 3000 {
487
+
return NewValidationError("description", "must be 3000 characters or less")
490
+
// Visibility should already be set with default in CreateCommunity
491
+
if req.Visibility != "public" && req.Visibility != "unlisted" && req.Visibility != "private" {
492
+
return ErrInvalidVisibility
495
+
if req.CreatedByDID == "" {
496
+
return NewValidationError("createdByDid", "required")
499
+
if req.HostedByDID == "" {
500
+
return NewValidationError("hostedByDid", "required")
506
+
// PDS write-forward helpers
508
+
func (s *communityService) createRecordOnPDS(ctx context.Context, repoDID, collection, rkey string, record map[string]interface{}) (string, string, error) {
509
+
endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.createRecord", strings.TrimSuffix(s.pdsURL, "/"))
511
+
payload := map[string]interface{}{
513
+
"collection": collection,
518
+
payload["rkey"] = rkey
521
+
return s.callPDS(ctx, "POST", endpoint, payload)
524
+
func (s *communityService) putRecordOnPDS(ctx context.Context, repoDID, collection, rkey string, record map[string]interface{}) (string, string, error) {
525
+
endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.putRecord", strings.TrimSuffix(s.pdsURL, "/"))
527
+
payload := map[string]interface{}{
529
+
"collection": collection,
534
+
return s.callPDS(ctx, "POST", endpoint, payload)
537
+
func (s *communityService) deleteRecordOnPDS(ctx context.Context, repoDID, collection, rkey string) error {
538
+
endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.deleteRecord", strings.TrimSuffix(s.pdsURL, "/"))
540
+
payload := map[string]interface{}{
542
+
"collection": collection,
546
+
_, _, err := s.callPDS(ctx, "POST", endpoint, payload)
550
+
func (s *communityService) callPDS(ctx context.Context, method, endpoint string, payload map[string]interface{}) (string, string, error) {
551
+
jsonData, err := json.Marshal(payload)
553
+
return "", "", fmt.Errorf("failed to marshal payload: %w", err)
556
+
req, err := http.NewRequestWithContext(ctx, method, endpoint, bytes.NewBuffer(jsonData))
558
+
return "", "", fmt.Errorf("failed to create request: %w", err)
560
+
req.Header.Set("Content-Type", "application/json")
562
+
// Add authentication if we have an access token
563
+
if s.pdsAccessToken != "" {
564
+
req.Header.Set("Authorization", "Bearer "+s.pdsAccessToken)
567
+
client := &http.Client{Timeout: 10 * time.Second}
568
+
resp, err := client.Do(req)
570
+
return "", "", fmt.Errorf("failed to call PDS: %w", err)
572
+
defer resp.Body.Close()
574
+
body, err := io.ReadAll(resp.Body)
576
+
return "", "", fmt.Errorf("failed to read response: %w", err)
579
+
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
580
+
return "", "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
583
+
// Parse response to extract URI and CID
584
+
var result struct {
585
+
URI string `json:"uri"`
586
+
CID string `json:"cid"`
588
+
if err := json.Unmarshal(body, &result); err != nil {
589
+
// For delete operations, there might not be a response body
590
+
if method == "POST" && strings.Contains(endpoint, "deleteRecord") {
593
+
return "", "", fmt.Errorf("failed to parse PDS response: %w", err)
596
+
return result.URI, result.CID, nil
599
+
// Helper functions
601
+
func extractDomain(didOrURL string) string {
602
+
// For did:web:example.com -> example.com
603
+
if strings.HasPrefix(didOrURL, "did:web:") {
604
+
parts := strings.Split(didOrURL, ":")
605
+
if len(parts) >= 3 {
610
+
// For URLs, extract domain
611
+
if strings.Contains(didOrURL, "://") {
612
+
parts := strings.Split(didOrURL, "://")
613
+
if len(parts) >= 2 {
614
+
domain := strings.Split(parts[1], "/")[0]
615
+
domain = strings.Split(domain, ":")[0] // Remove port
623
+
func extractRKeyFromURI(uri string) string {
624
+
// at://did/collection/rkey -> rkey
625
+
parts := strings.Split(uri, "/")
626
+
if len(parts) >= 4 {
627
+
return parts[len(parts)-1]
bretton.dev