diff --git a/docs/PRD_BACKLOG.md b/docs/PRD_BACKLOG.md index e3b107a..372a887 100644 --- a/docs/PRD_BACKLOG.md +++ b/docs/PRD_BACKLOG.md @@ -2,7 +2,7 @@ **Status:** Ongoing **Owner:** Platform Team -**Last Updated:** 2025-10-16 +**Last Updated:** 2025-10-17 ## Overview @@ -49,14 +49,34 @@ Miscellaneous platform improvements, bug fixes, and technical debt that don't fi --- -### Token Refresh Logic for Community Credentials -**Added:** 2025-10-11 | **Effort:** 1-2 days | **Priority:** ALPHA BLOCKER +### ✅ Token Refresh Logic for Community Credentials - COMPLETE +**Added:** 2025-10-11 | **Completed:** 2025-10-17 | **Effort:** 1.5 days | **Status:** ✅ DONE **Problem:** Community PDS access tokens expire (~2hrs). Updates fail until manual intervention. -**Solution:** Auto-refresh tokens before PDS operations. Parse JWT exp claim, use refresh token when expired, update DB. +**Solution Implemented:** +- ✅ Automatic token refresh before PDS operations (5-minute buffer before expiration) +- ✅ JWT expiration parsing without signature verification (`parseJWTExpiration`, `needsRefresh`) +- ✅ Token refresh using Indigo SDK (`atproto.ServerRefreshSession`) +- ✅ Password fallback when refresh tokens expire (~2 months) via `atproto.ServerCreateSession` +- ✅ Atomic credential updates (`UpdateCredentials` repository method) +- ✅ Concurrency-safe with per-community mutex locking +- ✅ Structured logging for monitoring (`[TOKEN-REFRESH]` events) +- ✅ Integration tests for token expiration detection and credential updates + +**Files Created:** +- [internal/core/communities/token_utils.go](../internal/core/communities/token_utils.go) - JWT parsing utilities +- [internal/core/communities/token_refresh.go](../internal/core/communities/token_refresh.go) - Refresh and re-auth logic +- [tests/integration/token_refresh_test.go](../tests/integration/token_refresh_test.go) - Integration tests + +**Files Modified:** +- [internal/core/communities/service.go](../internal/core/communities/service.go) - Added `ensureFreshToken` + concurrency control +- [internal/core/communities/interfaces.go](../internal/core/communities/interfaces.go) - Added `UpdateCredentials` interface +- [internal/db/postgres/community_repo.go](../internal/db/postgres/community_repo.go) - Implemented `UpdateCredentials` + +**Documentation:** See [IMPLEMENTATION_TOKEN_REFRESH.md](../docs/IMPLEMENTATION_TOKEN_REFRESH.md) for full details -**Code:** TODO in [communities/service.go:123](../internal/core/communities/service.go#L123) +**Impact:** ✅ Communities can now be updated 24+ hours after creation without manual intervention --- @@ -112,6 +132,56 @@ Miscellaneous platform improvements, bug fixes, and technical debt that don't fi --- +## 🔴 P1.5: Federation Blockers (Beta Launch) + +### Cross-PDS Write-Forward Support +**Added:** 2025-10-17 | **Effort:** 3-4 hours | **Priority:** FEDERATION BLOCKER (Beta) + +**Problem:** Current write-forward implementation assumes all users are on the same PDS as the Coves instance. This breaks federation when users from external PDSs try to interact with communities. + +**Current Behavior:** +- User on `pds.bsky.social` subscribes to community on `coves.social` +- Coves calls `s.pdsURL` (instance default: `http://localhost:3001`) +- Write goes to WRONG PDS → fails with 401/403 + +**Impact:** +- ✅ **Alpha**: Works fine (single PDS deployment) +- ❌ **Beta**: Breaks federation (users on different PDSs can't subscribe/interact) + +**Root Cause:** +- [service.go:736](../internal/core/communities/service.go#L736): `createRecordOnPDSAs` hardcodes `s.pdsURL` +- [service.go:753](../internal/core/communities/service.go#L753): `putRecordOnPDSAs` hardcodes `s.pdsURL` +- [service.go:767](../internal/core/communities/service.go#L767): `deleteRecordOnPDSAs` hardcodes `s.pdsURL` + +**Solution:** +1. Add identity resolver dependency to `CommunityService` +2. Before write-forward, resolve user's DID → extract PDS URL +3. Call user's actual PDS instead of `s.pdsURL` + +**Implementation:** +```go +// Before write-forward to user's repo: +userIdentity, err := s.identityResolver.ResolveDID(ctx, userDID) +if err != nil { + return fmt.Errorf("failed to resolve user PDS: %w", err) +} + +// Use user's actual PDS URL +endpoint := fmt.Sprintf("%s/xrpc/com.atproto.repo.createRecord", userIdentity.PDSURL) +``` + +**Files to Modify:** +- `internal/core/communities/service.go` - Add resolver, modify write-forward methods +- `cmd/server/main.go` - Pass identity resolver to community service constructor +- Tests - Add cross-PDS scenarios + +**Testing:** +- User on external PDS subscribes to community +- User on external PDS blocks community +- Community updates still work (communities ARE on instance PDS) + +--- + ## 🟢 P2: Nice-to-Have ### Remove Categories from Community Lexicon @@ -223,6 +293,34 @@ Document: did:plc choice, pgcrypto encryption, Jetstream vs firehose, write-forw ## Recent Completions +### ✅ Token Refresh for Community Credentials (2025-10-17) +**Completed:** Automatic token refresh prevents communities from breaking after 2 hours + +**Implementation:** +- ✅ JWT expiration parsing and refresh detection (5-minute buffer) +- ✅ Token refresh using Indigo SDK (`atproto.ServerRefreshSession`) +- ✅ Password fallback when refresh tokens expire (`atproto.ServerCreateSession`) +- ✅ Atomic credential updates in database (`UpdateCredentials`) +- ✅ Concurrency-safe with per-community mutex locking +- ✅ Structured logging for monitoring (`[TOKEN-REFRESH]` events) +- ✅ Integration tests for expiration detection and credential updates + +**Files Created:** +- [internal/core/communities/token_utils.go](../internal/core/communities/token_utils.go) +- [internal/core/communities/token_refresh.go](../internal/core/communities/token_refresh.go) +- [tests/integration/token_refresh_test.go](../tests/integration/token_refresh_test.go) + +**Files Modified:** +- [internal/core/communities/service.go](../internal/core/communities/service.go) - Added `ensureFreshToken` method +- [internal/core/communities/interfaces.go](../internal/core/communities/interfaces.go) - Added `UpdateCredentials` interface +- [internal/db/postgres/community_repo.go](../internal/db/postgres/community_repo.go) - Implemented `UpdateCredentials` + +**Documentation:** [IMPLEMENTATION_TOKEN_REFRESH.md](../docs/IMPLEMENTATION_TOKEN_REFRESH.md) + +**Impact:** Communities now work indefinitely without manual token management + +--- + ### ✅ OAuth Authentication for Community Actions (2025-10-16) **Completed:** Full OAuth JWT authentication flow for protected endpoints diff --git a/docs/PRD_COMMUNITIES.md b/docs/PRD_COMMUNITIES.md index f2d8a5d..ac1165e 100644 --- a/docs/PRD_COMMUNITIES.md +++ b/docs/PRD_COMMUNITIES.md @@ -2,7 +2,7 @@ **Status:** In Development **Owner:** Platform Team -**Last Updated:** 2025-10-16 +**Last Updated:** 2025-10-17 ## Overview @@ -33,12 +33,13 @@ Hosted By: did:web:coves.social (instance manages credentials) --- -## ✅ Completed Features (2025-10-10) +## ✅ Completed Features (Updated 2025-10-17) ### Core Infrastructure - [x] **V2 Architecture:** Communities own their own repositories - [x] **PDS Account Provisioning:** Automatic account creation for each community - [x] **Credential Management:** Secure storage of community PDS credentials +- [x] **Token Refresh:** Automatic refresh of expired access tokens (completed 2025-10-17) - [x] **Encryption at Rest:** PostgreSQL pgcrypto for sensitive credentials - [x] **Write-Forward Pattern:** Service → PDS → Firehose → AppView - [x] **Jetstream Consumer:** Real-time indexing from firehose @@ -47,8 +48,11 @@ Hosted By: did:web:coves.social (instance manages credentials) ### Security & Data Protection - [x] **Encrypted Credentials:** Access/refresh tokens encrypted in database - [x] **Credential Persistence:** PDS credentials survive server restarts +- [x] **Automatic Token Refresh:** Tokens refresh 5 minutes before expiration (completed 2025-10-17) +- [x] **Password Fallback:** Re-authentication when refresh tokens expire +- [x] **Concurrency Safety:** Per-community mutex prevents refresh race conditions - [x] **JSON Exclusion:** Credentials never exposed in API responses (`json:"-"` tags) -- [x] **Password Hashing:** bcrypt for PDS account passwords +- [x] **Password Encryption:** Encrypted (not hashed) for session creation fallback - [x] **Timeout Handling:** 30s timeout for write operations, 10s for reads ### Database Schema @@ -79,6 +83,7 @@ Hosted By: did:web:coves.social (instance manages credentials) ### Testing Coverage - [x] **Integration Tests:** Full CRUD operations - [x] **Credential Tests:** Persistence, encryption, decryption +- [x] **Token Refresh Tests:** JWT parsing, credential updates, concurrency (completed 2025-10-17) - [x] **V2 Validation Tests:** Rkey enforcement, self-ownership - [x] **Consumer Tests:** Firehose event processing - [x] **Repository Tests:** Database operations @@ -112,12 +117,15 @@ Hosted By: did:web:coves.social (instance manages credentials) ## ⚠️ Alpha Blockers (Must Complete Before Alpha Launch) ### Critical Missing Features -- [ ] **Community Blocking:** Users can block communities from their feeds - - Lexicon: ❌ Need new record type (extend `social.coves.actor.block` or create new) - - Service: ❌ No implementation (`BlockCommunity()` / `UnblockCommunity()`) - - Handler: ❌ No endpoints - - Repository: ❌ No methods - - **Impact:** Users have no way to hide unwanted communities +- [x] **Community Blocking:** ✅ COMPLETE - Users can block communities from their feeds + - ✅ Lexicon: `social.coves.community.block` record type implemented + - ✅ Service: `BlockCommunity()` / `UnblockCommunity()` / `GetBlockedCommunities()` / `IsBlocked()` + - ✅ Handlers: Block/unblock endpoints implemented + - ✅ Repository: Full blocking methods with indexing + - ✅ Jetstream Consumer: Real-time indexing of block events + - ✅ Integration tests: Comprehensive coverage + - **Completed:** 2025-10-16 + - **Impact:** Users can now hide unwanted communities from their feeds ### ✅ Critical Infrastructure - RESOLVED (2025-10-16) - [x] **✅ Subscription Indexing & ContentVisibility - COMPLETE** @@ -159,9 +167,14 @@ Hosted By: did:web:coves.social (instance manages credentials) - ✅ All E2E tests pass with real PDS authentication - **Completed:** 2025-10-16 -- [ ] **Token Refresh Logic:** Auto-refresh expired PDS access tokens - - **Impact:** Communities break after ~2 hours when tokens expire - - **See:** [PRD_BACKLOG.md P1 Priority](docs/PRD_BACKLOG.md#L31-L38) +- [x] **Token Refresh Logic:** ✅ COMPLETE - Auto-refresh expired PDS access tokens + - ✅ Automatic token refresh before PDS operations (5-minute buffer) + - ✅ Password fallback when refresh tokens expire (~2 months) + - ✅ Concurrency-safe with per-community mutex locking + - ✅ Atomic credential updates in database + - ✅ Integration tests and structured logging + - **Completed:** 2025-10-17 + - **See:** [IMPLEMENTATION_TOKEN_REFRESH.md](docs/IMPLEMENTATION_TOKEN_REFRESH.md) --- diff --git a/internal/core/communities/interfaces.go b/internal/core/communities/interfaces.go index 40545d5..8b66cdf 100644 --- a/internal/core/communities/interfaces.go +++ b/internal/core/communities/interfaces.go @@ -12,6 +12,9 @@ type Repository interface { Update(ctx context.Context, community *Community) (*Community, error) Delete(ctx context.Context, did string) error + // Credential Management (for token refresh) + UpdateCredentials(ctx context.Context, did, accessToken, refreshToken string) error + // Listing & Search List(ctx context.Context, req ListCommunitiesRequest) ([]*Community, int, error) // Returns communities + total count Search(ctx context.Context, req SearchCommunitiesRequest) ([]*Community, int, error) diff --git a/internal/core/communities/service.go b/internal/core/communities/service.go index 137523f..e9c20b3 100644 --- a/internal/core/communities/service.go +++ b/internal/core/communities/service.go @@ -12,6 +12,7 @@ import ( "net/http" "regexp" "strings" + "sync" "time" ) @@ -20,14 +21,31 @@ import ( 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])?$`) type communityService struct { - repo Repository - provisioner *PDSAccountProvisioner + // Interfaces and pointers first (better alignment) + repo Repository + provisioner *PDSAccountProvisioner + + // Token refresh concurrency control + // Each community gets its own mutex to prevent concurrent refresh attempts + refreshMutexes map[string]*sync.Mutex + + // Strings pdsURL string instanceDID string instanceDomain string pdsAccessToken string + + // Sync primitives last + mapMutex sync.RWMutex // Protects refreshMutexes map itself } +const ( + // Maximum recommended size for mutex cache (warning threshold, not hard limit) + // At 10,000 entries × 16 bytes = ~160KB memory (negligible overhead) + // Map can grow larger in production - even 100,000 entries = 1.6MB is acceptable + maxMutexCacheSize = 10000 +) + // NewCommunityService creates a new community service func NewCommunityService(repo Repository, pdsURL, instanceDID, instanceDomain string, provisioner *PDSAccountProvisioner) Service { // SECURITY: Basic validation that did:web domain matches configured instanceDomain @@ -50,6 +68,7 @@ func NewCommunityService(repo Repository, pdsURL, instanceDID, instanceDomain st instanceDID: instanceDID, instanceDomain: instanceDomain, provisioner: provisioner, + refreshMutexes: make(map[string]*sync.Mutex), } } @@ -235,6 +254,13 @@ func (s *communityService) UpdateCommunity(ctx context.Context, req UpdateCommun return nil, err } + // CRITICAL: Ensure fresh PDS access token before write operation + // Community PDS tokens expire every ~2 hours and must be refreshed + existing, err = s.ensureFreshToken(ctx, existing) + if err != nil { + return nil, fmt.Errorf("failed to ensure fresh credentials: %w", err) + } + // Authorization: verify user is the creator // TODO(Communities-Auth): Add moderator check when moderation system is implemented if existing.CreatedByDID != req.UpdatedByDID { @@ -349,6 +375,145 @@ func (s *communityService) UpdateCommunity(ctx context.Context, req UpdateCommun return &updated, nil } +// getOrCreateRefreshMutex returns a mutex for the given community DID +// Thread-safe with read-lock fast path for existing entries +// SAFETY: Does NOT evict entries to avoid race condition where: +// 1. Thread A holds mutex for community-123 +// 2. Thread B evicts community-123 from map +// 3. Thread C creates NEW mutex for community-123 +// 4. Now two threads can refresh community-123 concurrently (mutex defeated!) +func (s *communityService) getOrCreateRefreshMutex(did string) *sync.Mutex { + // Fast path: check if mutex already exists (read lock) + s.mapMutex.RLock() + mutex, exists := s.refreshMutexes[did] + s.mapMutex.RUnlock() + + if exists { + return mutex + } + + // Slow path: create new mutex (write lock) + s.mapMutex.Lock() + defer s.mapMutex.Unlock() + + // Double-check after acquiring write lock (another goroutine might have created it) + mutex, exists = s.refreshMutexes[did] + if exists { + return mutex + } + + // Create new mutex + mutex = &sync.Mutex{} + s.refreshMutexes[did] = mutex + + // SAFETY: No eviction to prevent race condition + // Map will grow beyond maxMutexCacheSize but this is safer than evicting in-use mutexes + if len(s.refreshMutexes) > maxMutexCacheSize { + memoryKB := len(s.refreshMutexes) * 16 / 1024 + 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", + len(s.refreshMutexes), maxMutexCacheSize, memoryKB) + } + + return mutex +} + +// ensureFreshToken checks if a community's access token needs refresh and updates if needed +// Returns updated community with fresh credentials (or original if no refresh needed) +// Thread-safe: Uses per-community mutex to prevent concurrent refresh attempts +func (s *communityService) ensureFreshToken(ctx context.Context, community *Community) (*Community, error) { + // Get or create mutex for this specific community DID + mutex := s.getOrCreateRefreshMutex(community.DID) + + // Lock for this specific community (allows other communities to refresh concurrently) + mutex.Lock() + defer mutex.Unlock() + + // Re-fetch community from DB (another goroutine might have already refreshed it) + fresh, err := s.repo.GetByDID(ctx, community.DID) + if err != nil { + return nil, fmt.Errorf("failed to re-fetch community: %w", err) + } + + // Check if token needs refresh (5-minute buffer before expiration) + needsRefresh, err := NeedsRefresh(fresh.PDSAccessToken) + if err != nil { + log.Printf("[TOKEN-REFRESH] Community: %s, Event: token_parse_failed, Error: %v", fresh.DID, err) + return nil, fmt.Errorf("failed to check token expiration: %w", err) + } + + if !needsRefresh { + // Token still valid, no refresh needed + return fresh, nil + } + + log.Printf("[TOKEN-REFRESH] Community: %s, Event: token_refresh_started, Message: Access token expiring soon", fresh.DID) + + // Attempt token refresh using refresh token + newAccessToken, newRefreshToken, err := refreshPDSToken(ctx, fresh.PDSURL, fresh.PDSAccessToken, fresh.PDSRefreshToken) + if err != nil { + // Check if refresh token expired (need password fallback) + if strings.Contains(err.Error(), "expired or invalid") { + log.Printf("[TOKEN-REFRESH] Community: %s, Event: refresh_token_expired, Message: Re-authenticating with password", fresh.DID) + + // Fallback: Re-authenticate with stored password + newAccessToken, newRefreshToken, err = reauthenticateWithPassword( + ctx, + fresh.PDSURL, + fresh.PDSEmail, + fresh.PDSPassword, // Retrieved decrypted from DB + ) + if err != nil { + log.Printf("[TOKEN-REFRESH] Community: %s, Event: password_auth_failed, Error: %v", fresh.DID, err) + return nil, fmt.Errorf("failed to re-authenticate community: %w", err) + } + + log.Printf("[TOKEN-REFRESH] Community: %s, Event: password_fallback_success, Message: Re-authenticated after refresh token expiry", fresh.DID) + } else { + log.Printf("[TOKEN-REFRESH] Community: %s, Event: refresh_failed, Error: %v", fresh.DID, err) + return nil, fmt.Errorf("failed to refresh token: %w", err) + } + } + + // CRITICAL: Update database with new tokens immediately + // Refresh tokens are SINGLE-USE - old one is now invalid + // Use retry logic to handle transient DB failures + const maxRetries = 3 + var updateErr error + for attempt := 0; attempt < maxRetries; attempt++ { + updateErr = s.repo.UpdateCredentials(ctx, fresh.DID, newAccessToken, newRefreshToken) + if updateErr == nil { + break // Success + } + + log.Printf("[TOKEN-REFRESH] Community: %s, Event: db_update_retry, Attempt: %d/%d, Error: %v", + fresh.DID, attempt+1, maxRetries, updateErr) + + if attempt < maxRetries-1 { + // Exponential backoff: 100ms, 200ms, 400ms + backoff := time.Duration(1<