diff --git a/CLAUDE.md b/CLAUDE.md index 7ed59bf..a166e59 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,119 +1,112 @@ -Project: Coves Builder You are a distinguished developer actively building Coves, a forum-like atProto social media platform. Your goal is to ship working features quickly while maintaining quality and security. - -## Builder Mindset - -- Ship working code today, refactor tomorrow -- Security is built-in, not bolted-on -- Test-driven: write the test, then make it pass -- ASK QUESTIONS if you need context surrounding the product DONT ASSUME - -## No Stubs, No Shortcuts -- **NEVER** use `unimplemented!()`, `todo!()`, or stub implementations -- **NEVER** leave placeholder code or incomplete implementations -- **NEVER** skip functionality because it seems complex -- Every function must be fully implemented and working -- Every feature must be complete before moving on -- E2E tests must test REAL infrastructure - not mocks - -## Issue Tracking - -**This project uses [bd (beads)](https://github.com/steveyegge/beads) for ALL issue tracking.** - -- Use `bd` commands, NOT markdown TODOs or task lists -- Check `bd ready` for unblocked work -- Always commit `.beads/issues.jsonl` with code changes -- See [AGENTS.md](AGENTS.md) for full workflow details - -Quick commands: -- `bd ready --json` - Show ready work -- `bd create "Title" -t bug|feature|task -p 0-4 --json` - Create issue -- `bd update --status in_progress --json` - Claim work -- `bd close --reason "Done" --json` - Complete work -## Break Down Complex Tasks -- Large files or complex features should be broken into manageable chunks -- If a file is too large, discuss breaking it into smaller modules -- If a task seems overwhelming, ask the user how to break it down -- Work incrementally, but each increment must be complete and functional - -#### Human & LLM Readability Guidelines: -- Descriptive Naming: Use full words over abbreviations (e.g., CommunityGovernance not CommGov) - -## atProto Essentials for Coves - -### Architecture - -- **PDS is Self-Contained**: Uses internal SQLite + CAR files (in Docker volume) -- **PostgreSQL for AppView Only**: One database for Coves AppView indexing -- **Don't Touch PDS Internals**: PDS manages its own storage, we just read from firehose -- **Data Flow**: Client → PDS → Firehose → AppView → PostgreSQL - -### Always Consider: - -- [ ]  **Identity**: Every action needs DID verification -- [ ]  **Record Types**: Define custom lexicons (e.g., `social.coves.post`, `social.coves.community`) -- [ ]  **Is it federated-friendly?** (Can other PDSs interact with it?) -- [ ]  **Does the Lexicon make sense?** (Would it work for other forums?) -- [ ]  **AppView only indexes**: We don't write to CAR files, only read from firehose - -## Security-First Building - -### Every Feature MUST: - -- [ ]  **Validate all inputs** at the handler level -- [ ]  **Use parameterized queries** (never string concatenation) -- [ ]  **Check authorization** before any operation -- [ ]  **Limit resource access** (pagination, rate limits) -- [ ]  **Log security events** (failed auth, invalid inputs) -- [ ]  **Never log sensitive data** (passwords, tokens, PII) - -### Red Flags to Avoid: - -- `fmt.Sprintf` in SQL queries → Use parameterized queries -- Missing `context.Context` → Need it for timeouts/cancellation -- No input validation → Add it immediately -- Error messages with internal details → Wrap errors properly -- Unbounded queries → Add limits/pagination - -### "How should I structure this?" - -1. One domain, one package -2. Interfaces for testability -3. Services coordinate repos -4. Handlers only handle XRPC - -## Comprehensive Testing -- Write comprehensive unit tests for every module -- Aim for high test coverage (all major code paths) -- Test edge cases, error conditions, and boundary values -- Include doc tests for public APIs -- All tests must pass before considering a file "complete" -- Test both success and failure cases -## Pre-Production Advantages - -Since we're pre-production: - -- **Break things**: Delete and rebuild rather than complex migrations -- **Experiment**: Try approaches, keep what works -- **Simplify**: Remove unused code aggressively -- **But never compromise security basics** - -## Success Metrics - -Your code is ready when: - -- [ ]  Tests pass (including security tests) -- [ ]  Follows atProto patterns -- [ ]  Handles errors gracefully -- [ ]  Works end-to-end with auth - -## Quick Checks Before Committing - -1. **Will it work?** (Integration test proves it) -2. **Is it secure?** (Auth, validation, parameterized queries) -3. **Is it simple?** (Could you explain to a junior?) -4. **Is it complete?** (Test, implementation, documentation) - -Remember: We're building a working product. Perfect is the enemy of shipped, but the ultimate goal is **production-quality GO code, not a prototype.** - -Every line of code should be something you'd be proud to ship in a production system. Quality over speed. Completeness over convenience. \ No newline at end of file +Project: Coves PR Reviewer +You are a distinguished senior architect conducting a thorough code review for Coves, a forum-like atProto social media platform. + +## Review Mindset +- Be constructive but thorough - catch issues before they reach production +- Question assumptions and look for edge cases +- Prioritize security, performance, and maintainability concerns +- Suggest alternatives when identifying problems +- Ensure there is proper test coverage that adequately tests atproto write forward architecture + + +## Special Attention Areas for Coves +- **atProto architecture**: - Ensure architecture follows atProto recommendations with WRITE FORWARD ARCHITECTURE (Appview -> PDS -> Relay -> Appview -> App DB (if necessary)) +- Ensure HTTP Endpoints match the Lexicon data contract +- **Federation**: Check for proper DID resolution and identity verification + +## Review Checklist + +### 1. Architecture Compliance +**MUST VERIFY:** +- [ ] NO SQL queries in handlers (automatic rejection if found) +- [ ] Proper layer separation: Handler → Service → Repository → Database +- [ ] Services use repository interfaces, not concrete implementations +- [ ] Dependencies injected via constructors, not globals +- [ ] No database packages imported in handlers + +### 2. Security Review +**CHECK FOR:** +- SQL injection vulnerabilities (even with prepared statements, verify) +- Proper input validation and sanitization +- Authentication/authorization checks on all protected endpoints +- No sensitive data in logs or error messages +- Rate limiting on public endpoints +- CSRF protection where applicable +- Proper atProto identity verification + +### 3. Error Handling Audit +**VERIFY:** +- All errors are handled, not ignored +- Error wrapping provides context: `fmt.Errorf("service: %w", err)` +- Domain errors defined in core/errors/ +- HTTP status codes correctly map to error types +- No internal error details exposed to API consumers +- Nil pointer checks before dereferencing + +### 4. Performance Considerations +**LOOK FOR:** +- N+1 query problems +- Missing database indexes for frequently queried fields +- Unnecessary database round trips +- Large unbounded queries without pagination +- Memory leaks in goroutines +- Proper connection pool usage +- Efficient atProto federation calls + +### 5. Testing Coverage +**REQUIRE:** +- Unit tests for all new service methods +- Integration tests for new API endpoints +- Edge case coverage (empty inputs, max values, special characters) +- Error path testing +- Mock verification in unit tests +- No flaky tests (check for time dependencies, random values) + +### 6. Code Quality +**ASSESS:** +- Naming follows conventions (full words, not abbreviations) +- Functions do one thing well +- No code duplication (DRY principle) +- Consistent error handling patterns +- Proper use of Go idioms +- No commented-out code + +### 7. Breaking Changes +**IDENTIFY:** +- API contract changes +- Database schema modifications affecting existing data +- Changes to core interfaces +- Modified error codes or response formats + +### 8. Documentation +**ENSURE:** +- API endpoints have example requests/responses +- Complex business logic is explained +- Database migrations include rollback scripts +- README updated if setup process changes +- Swagger/OpenAPI specs updated if applicable + +## Review Process + +1. **First Pass - Automatic Rejections** + - SQL in handlers + - Missing tests + - Security vulnerabilities + - Broken layer separation + +2. **Second Pass - Deep Dive** + - Business logic correctness + - Edge case handling + - Performance implications + - Code maintainability + +3. **Third Pass - Suggestions** + - Better patterns or approaches + - Refactoring opportunities + - Future considerations + +Then provide detailed feedback organized by: 1. 🚨 **Critical Issues** (must fix) 2. ⚠️ **Important Issues** (should fix) 3. 💡 **Suggestions** (consider for improvement) 4. ✅ **Good Practices Observed** (reinforce positive patterns) + + +Remember: The goal is to ship quality code quickly. Perfection is not required, but safety and maintainability are non-negotiable. diff --git a/cmd/reindex-votes/main.go b/cmd/reindex-votes/main.go new file mode 100644 index 0000000..db0f360 --- /dev/null +++ b/cmd/reindex-votes/main.go @@ -0,0 +1,267 @@ +// cmd/reindex-votes/main.go +// Quick tool to reindex votes from PDS to AppView database +package main + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "log" + "net/http" + "net/url" + "os" + "strings" + "time" + + _ "github.com/lib/pq" +) + +type ListRecordsResponse struct { + Records []Record `json:"records"` + Cursor string `json:"cursor"` +} + +type Record struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value map[string]interface{} `json:"value"` +} + +func main() { + // Get config from env + dbURL := os.Getenv("DATABASE_URL") + if dbURL == "" { + dbURL = "postgres://dev_user:dev_password@localhost:5435/coves_dev?sslmode=disable" + } + pdsURL := os.Getenv("PDS_URL") + if pdsURL == "" { + pdsURL = "http://localhost:3001" + } + + log.Printf("Connecting to database...") + db, err := sql.Open("postgres", dbURL) + if err != nil { + log.Fatalf("Failed to connect to database: %v", err) + } + defer db.Close() + + ctx := context.Background() + + // Get all accounts directly from the PDS + log.Printf("Fetching accounts from PDS (%s)...", pdsURL) + dids, err := fetchAllAccountsFromPDS(pdsURL) + if err != nil { + log.Fatalf("Failed to fetch accounts from PDS: %v", err) + } + log.Printf("Found %d accounts on PDS to check for votes", len(dids)) + + // Reset vote counts first + log.Printf("Resetting all vote counts...") + if _, err := db.ExecContext(ctx, "DELETE FROM votes"); err != nil { + log.Fatalf("Failed to clear votes table: %v", err) + } + if _, err := db.ExecContext(ctx, "UPDATE posts SET upvote_count = 0, downvote_count = 0, score = 0"); err != nil { + log.Fatalf("Failed to reset post vote counts: %v", err) + } + if _, err := db.ExecContext(ctx, "UPDATE comments SET upvote_count = 0, downvote_count = 0, score = 0"); err != nil { + log.Fatalf("Failed to reset comment vote counts: %v", err) + } + + // For each user, fetch their votes from PDS + totalVotes := 0 + for _, did := range dids { + votes, err := fetchVotesFromPDS(pdsURL, did) + if err != nil { + log.Printf("Warning: failed to fetch votes for %s: %v", did, err) + continue + } + + if len(votes) == 0 { + continue + } + + log.Printf("Found %d votes for %s", len(votes), did) + + // Index each vote + for _, vote := range votes { + if err := indexVote(ctx, db, did, vote); err != nil { + log.Printf("Warning: failed to index vote %s: %v", vote.URI, err) + continue + } + totalVotes++ + } + } + + log.Printf("✓ Reindexed %d votes from PDS", totalVotes) +} + +// fetchAllAccountsFromPDS queries the PDS sync API to get all repo DIDs +func fetchAllAccountsFromPDS(pdsURL string) ([]string, error) { + // Use com.atproto.sync.listRepos to get all repos on this PDS + var allDIDs []string + cursor := "" + + for { + reqURL := fmt.Sprintf("%s/xrpc/com.atproto.sync.listRepos?limit=100", pdsURL) + if cursor != "" { + reqURL += "&cursor=" + url.QueryEscape(cursor) + } + + resp, err := http.Get(reqURL) + if err != nil { + return nil, fmt.Errorf("HTTP request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var result struct { + Repos []struct { + DID string `json:"did"` + } `json:"repos"` + Cursor string `json:"cursor"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + for _, repo := range result.Repos { + allDIDs = append(allDIDs, repo.DID) + } + + if result.Cursor == "" { + break + } + cursor = result.Cursor + } + + return allDIDs, nil +} + +func fetchVotesFromPDS(pdsURL, did string) ([]Record, error) { + var allRecords []Record + cursor := "" + collection := "social.coves.feed.vote" + + for { + reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100", + pdsURL, url.QueryEscape(did), url.QueryEscape(collection)) + if cursor != "" { + reqURL += "&cursor=" + url.QueryEscape(cursor) + } + + resp, err := http.Get(reqURL) + if err != nil { + return nil, fmt.Errorf("HTTP request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == 400 { + // User doesn't exist on this PDS or has no records - that's OK + return nil, nil + } + if resp.StatusCode != 200 { + return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode) + } + + var result ListRecordsResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + allRecords = append(allRecords, result.Records...) + + if result.Cursor == "" { + break + } + cursor = result.Cursor + } + + return allRecords, nil +} + +func indexVote(ctx context.Context, db *sql.DB, voterDID string, record Record) error { + // Extract vote data from record + subject, ok := record.Value["subject"].(map[string]interface{}) + if !ok { + return fmt.Errorf("missing subject") + } + subjectURI, _ := subject["uri"].(string) + subjectCID, _ := subject["cid"].(string) + direction, _ := record.Value["direction"].(string) + createdAtStr, _ := record.Value["createdAt"].(string) + + if subjectURI == "" || direction == "" { + return fmt.Errorf("invalid vote record: missing required fields") + } + + // Parse created_at + createdAt, err := time.Parse(time.RFC3339, createdAtStr) + if err != nil { + createdAt = time.Now() + } + + // Extract rkey from URI (at://did/collection/rkey) + parts := strings.Split(record.URI, "/") + if len(parts) < 5 { + return fmt.Errorf("invalid URI format: %s", record.URI) + } + rkey := parts[len(parts)-1] + + // Start transaction + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback() + + // Insert vote + _, err = tx.ExecContext(ctx, ` + INSERT INTO votes (uri, cid, rkey, voter_did, subject_uri, subject_cid, direction, created_at, indexed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW()) + ON CONFLICT (uri) DO NOTHING + `, record.URI, record.CID, rkey, voterDID, subjectURI, subjectCID, direction, createdAt) + if err != nil { + return fmt.Errorf("failed to insert vote: %w", err) + } + + // Update post/comment counts + collection := extractCollectionFromURI(subjectURI) + var updateQuery string + + switch collection { + case "social.coves.community.post": + if direction == "up" { + updateQuery = `UPDATE posts SET upvote_count = upvote_count + 1, score = upvote_count + 1 - downvote_count WHERE uri = $1 AND deleted_at IS NULL` + } else { + updateQuery = `UPDATE posts SET downvote_count = downvote_count + 1, score = upvote_count - (downvote_count + 1) WHERE uri = $1 AND deleted_at IS NULL` + } + case "social.coves.community.comment": + if direction == "up" { + updateQuery = `UPDATE comments SET upvote_count = upvote_count + 1, score = upvote_count + 1 - downvote_count WHERE uri = $1 AND deleted_at IS NULL` + } else { + updateQuery = `UPDATE comments SET downvote_count = downvote_count + 1, score = upvote_count - (downvote_count + 1) WHERE uri = $1 AND deleted_at IS NULL` + } + default: + // Unknown collection, just index the vote + return tx.Commit() + } + + if _, err := tx.ExecContext(ctx, updateQuery, subjectURI); err != nil { + return fmt.Errorf("failed to update vote counts: %w", err) + } + + return tx.Commit() +} + +func extractCollectionFromURI(uri string) string { + // at://did:plc:xxx/social.coves.community.post/rkey + parts := strings.Split(uri, "/") + if len(parts) >= 4 { + return parts[3] + } + return "" +} diff --git a/cmd/server/main.go b/cmd/server/main.go index a8b5ad0..9e5c6e3 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -396,12 +396,17 @@ func main() { commentRepo := postgresRepo.NewCommentRepository(db) log.Println("✅ Comment repository initialized (Jetstream indexing only)") + // Initialize vote cache (stores user votes from PDS to avoid eventual consistency issues) + // TTL of 10 minutes - cache is also updated on vote create/delete + voteCache := votes.NewVoteCache(10*time.Minute, nil) + log.Println("✅ Vote cache initialized (10 minute TTL)") + // Initialize vote service (for XRPC API endpoints) // Note: We don't validate subject existence - the vote goes to the user's PDS regardless. // The Jetstream consumer handles orphaned votes correctly by only updating counts for // non-deleted subjects. This avoids race conditions and eventual consistency issues. - voteService := votes.NewService(voteRepo, oauthClient, oauthStore, nil) - log.Println("✅ Vote service initialized (with OAuth authentication)") + voteService := votes.NewService(voteRepo, oauthClient, oauthStore, voteCache, nil) + log.Println("✅ Vote service initialized (with OAuth authentication and vote cache)") // Initialize comment service (for query API) // Requires user and community repos for proper author/community hydration per lexicon @@ -524,11 +529,11 @@ func main() { routes.RegisterVoteRoutes(r, voteService, authMiddleware) log.Println("Vote XRPC endpoints registered with OAuth authentication") - routes.RegisterCommunityFeedRoutes(r, feedService) - log.Println("Feed XRPC endpoints registered (public, no auth required)") + routes.RegisterCommunityFeedRoutes(r, feedService, voteService, authMiddleware) + log.Println("Feed XRPC endpoints registered (public with optional auth for viewer vote state)") - routes.RegisterTimelineRoutes(r, timelineService, authMiddleware) - log.Println("Timeline XRPC endpoints registered (requires authentication)") + routes.RegisterTimelineRoutes(r, timelineService, voteService, authMiddleware) + log.Println("Timeline XRPC endpoints registered (requires authentication, includes viewer vote state)") routes.RegisterDiscoverRoutes(r, discoverService) log.Println("Discover XRPC endpoints registered (public, no auth required)") diff --git a/internal/api/handlers/communityFeed/get_community.go b/internal/api/handlers/communityFeed/get_community.go index 015e1f2..e2effec 100644 --- a/internal/api/handlers/communityFeed/get_community.go +++ b/internal/api/handlers/communityFeed/get_community.go @@ -1,8 +1,10 @@ package communityFeed import ( + "Coves/internal/api/middleware" "Coves/internal/core/communityFeeds" "Coves/internal/core/posts" + "Coves/internal/core/votes" "encoding/json" "log" "net/http" @@ -11,13 +13,15 @@ import ( // GetCommunityHandler handles community feed retrieval type GetCommunityHandler struct { - service communityFeeds.Service + service communityFeeds.Service + voteService votes.Service } // NewGetCommunityHandler creates a new community feed handler -func NewGetCommunityHandler(service communityFeeds.Service) *GetCommunityHandler { +func NewGetCommunityHandler(service communityFeeds.Service, voteService votes.Service) *GetCommunityHandler { return &GetCommunityHandler{ - service: service, + service: service, + voteService: voteService, } } @@ -36,10 +40,6 @@ func (h *GetCommunityHandler) HandleGetCommunity(w http.ResponseWriter, r *http. return } - // Alpha: No viewer context needed for basic community sorting - // TODO(feed-generator): Extract viewer DID when implementing viewer-specific state - // (blocks, upvotes, saves) in feed generator skeleton - // Get community feed response, err := h.service.GetCommunityFeed(r.Context(), req) if err != nil { @@ -47,6 +47,42 @@ func (h *GetCommunityHandler) HandleGetCommunity(w http.ResponseWriter, r *http. return } + // Populate viewer vote state if authenticated and vote service available + if h.voteService != nil { + session := middleware.GetOAuthSession(r) + if session != nil { + userDID := middleware.GetUserDID(r) + // Ensure vote cache is populated from PDS + if err := h.voteService.EnsureCachePopulated(r.Context(), session); err != nil { + // Log but don't fail - viewer state is optional + log.Printf("Warning: failed to populate vote cache: %v", err) + } else { + // Collect post URIs to batch lookup + postURIs := make([]string, 0, len(response.Feed)) + for _, feedPost := range response.Feed { + if feedPost.Post != nil { + postURIs = append(postURIs, feedPost.Post.URI) + } + } + + // Get viewer votes for all posts + viewerVotes := h.voteService.GetViewerVotesForSubjects(userDID, postURIs) + + // Populate viewer state on each post + for _, feedPost := range response.Feed { + if feedPost.Post != nil { + if vote, exists := viewerVotes[feedPost.Post.URI]; exists { + feedPost.Post.Viewer = &posts.ViewerState{ + Vote: &vote.Direction, + VoteURI: &vote.URI, + } + } + } + } + } + } + } + // Transform blob refs to URLs for all posts for _, feedPost := range response.Feed { if feedPost.Post != nil { diff --git a/internal/api/handlers/timeline/get_timeline.go b/internal/api/handlers/timeline/get_timeline.go index e6b9675..56ffa14 100644 --- a/internal/api/handlers/timeline/get_timeline.go +++ b/internal/api/handlers/timeline/get_timeline.go @@ -4,6 +4,7 @@ import ( "Coves/internal/api/middleware" "Coves/internal/core/posts" "Coves/internal/core/timeline" + "Coves/internal/core/votes" "encoding/json" "log" "net/http" @@ -13,13 +14,15 @@ import ( // GetTimelineHandler handles timeline feed retrieval type GetTimelineHandler struct { - service timeline.Service + service timeline.Service + voteService votes.Service } // NewGetTimelineHandler creates a new timeline handler -func NewGetTimelineHandler(service timeline.Service) *GetTimelineHandler { +func NewGetTimelineHandler(service timeline.Service, voteService votes.Service) *GetTimelineHandler { return &GetTimelineHandler{ - service: service, + service: service, + voteService: voteService, } } @@ -53,6 +56,41 @@ func (h *GetTimelineHandler) HandleGetTimeline(w http.ResponseWriter, r *http.Re return } + // Populate viewer vote state if authenticated and vote service available + if h.voteService != nil { + session := middleware.GetOAuthSession(r) + if session != nil { + // Ensure vote cache is populated from PDS + if err := h.voteService.EnsureCachePopulated(r.Context(), session); err != nil { + // Log but don't fail - viewer state is optional + log.Printf("Warning: failed to populate vote cache: %v", err) + } else { + // Collect post URIs to batch lookup + postURIs := make([]string, 0, len(response.Feed)) + for _, feedPost := range response.Feed { + if feedPost.Post != nil { + postURIs = append(postURIs, feedPost.Post.URI) + } + } + + // Get viewer votes for all posts + viewerVotes := h.voteService.GetViewerVotesForSubjects(userDID, postURIs) + + // Populate viewer state on each post + for _, feedPost := range response.Feed { + if feedPost.Post != nil { + if vote, exists := viewerVotes[feedPost.Post.URI]; exists { + feedPost.Post.Viewer = &posts.ViewerState{ + Vote: &vote.Direction, + VoteURI: &vote.URI, + } + } + } + } + } + } + } + // Transform blob refs to URLs for all posts for _, feedPost := range response.Feed { if feedPost.Post != nil { diff --git a/internal/api/routes/communityFeed.go b/internal/api/routes/communityFeed.go index 651cb21..74acad8 100644 --- a/internal/api/routes/communityFeed.go +++ b/internal/api/routes/communityFeed.go @@ -2,7 +2,9 @@ package routes import ( "Coves/internal/api/handlers/communityFeed" + "Coves/internal/api/middleware" "Coves/internal/core/communityFeeds" + "Coves/internal/core/votes" "github.com/go-chi/chi/v5" ) @@ -11,13 +13,13 @@ import ( func RegisterCommunityFeedRoutes( r chi.Router, feedService communityFeeds.Service, + voteService votes.Service, + authMiddleware *middleware.OAuthAuthMiddleware, ) { // Create handlers - getCommunityHandler := communityFeed.NewGetCommunityHandler(feedService) + getCommunityHandler := communityFeed.NewGetCommunityHandler(feedService, voteService) // GET /xrpc/social.coves.communityFeed.getCommunity - // Public endpoint - basic community sorting only for Alpha - // TODO(feed-generator): Add OptionalAuth middleware when implementing viewer-specific state - // (blocks, upvotes, saves, etc.) in feed generator skeleton - r.Get("/xrpc/social.coves.communityFeed.getCommunity", getCommunityHandler.HandleGetCommunity) + // Public endpoint with optional auth for viewer-specific state (vote state) + r.With(authMiddleware.OptionalAuth).Get("/xrpc/social.coves.communityFeed.getCommunity", getCommunityHandler.HandleGetCommunity) } diff --git a/internal/api/routes/timeline.go b/internal/api/routes/timeline.go index 0583deb..dd74b0c 100644 --- a/internal/api/routes/timeline.go +++ b/internal/api/routes/timeline.go @@ -4,6 +4,7 @@ import ( "Coves/internal/api/handlers/timeline" "Coves/internal/api/middleware" timelineCore "Coves/internal/core/timeline" + "Coves/internal/core/votes" "github.com/go-chi/chi/v5" ) @@ -12,10 +13,11 @@ import ( func RegisterTimelineRoutes( r chi.Router, timelineService timelineCore.Service, + voteService votes.Service, authMiddleware *middleware.OAuthAuthMiddleware, ) { // Create handlers - getTimelineHandler := timeline.NewGetTimelineHandler(timelineService) + getTimelineHandler := timeline.NewGetTimelineHandler(timelineService, voteService) // GET /xrpc/social.coves.feed.getTimeline // Requires authentication - user must be logged in to see their timeline diff --git a/internal/core/votes/cache.go b/internal/core/votes/cache.go new file mode 100644 index 0000000..690b341 --- /dev/null +++ b/internal/core/votes/cache.go @@ -0,0 +1,221 @@ +package votes + +import ( + "context" + "fmt" + "log/slog" + "strings" + "sync" + "time" + + "Coves/internal/atproto/pds" +) + +// CachedVote represents a vote stored in the cache +type CachedVote struct { + Direction string // "up" or "down" + URI string // vote record URI (at://did/collection/rkey) + RKey string // record key +} + +// VoteCache provides an in-memory cache of user votes fetched from their PDS. +// This avoids eventual consistency issues with the AppView database. +type VoteCache struct { + mu sync.RWMutex + votes map[string]map[string]*CachedVote // userDID -> subjectURI -> vote + expiry map[string]time.Time // userDID -> expiry time + ttl time.Duration + logger *slog.Logger +} + +// NewVoteCache creates a new vote cache with the specified TTL +func NewVoteCache(ttl time.Duration, logger *slog.Logger) *VoteCache { + if logger == nil { + logger = slog.Default() + } + return &VoteCache{ + votes: make(map[string]map[string]*CachedVote), + expiry: make(map[string]time.Time), + ttl: ttl, + logger: logger, + } +} + +// GetVotesForUser returns all cached votes for a user. +// Returns nil if cache is empty or expired for this user. +func (c *VoteCache) GetVotesForUser(userDID string) map[string]*CachedVote { + c.mu.RLock() + defer c.mu.RUnlock() + + // Check if cache exists and is not expired + expiry, exists := c.expiry[userDID] + if !exists || time.Now().After(expiry) { + return nil + } + + return c.votes[userDID] +} + +// GetVote returns the cached vote for a specific subject, or nil if not found/expired +func (c *VoteCache) GetVote(userDID, subjectURI string) *CachedVote { + votes := c.GetVotesForUser(userDID) + if votes == nil { + return nil + } + return votes[subjectURI] +} + +// IsCached returns true if the user's votes are cached and not expired +func (c *VoteCache) IsCached(userDID string) bool { + c.mu.RLock() + defer c.mu.RUnlock() + + expiry, exists := c.expiry[userDID] + return exists && time.Now().Before(expiry) +} + +// SetVotesForUser replaces all cached votes for a user +func (c *VoteCache) SetVotesForUser(userDID string, votes map[string]*CachedVote) { + c.mu.Lock() + defer c.mu.Unlock() + + c.votes[userDID] = votes + c.expiry[userDID] = time.Now().Add(c.ttl) + + c.logger.Debug("vote cache updated", + "user", userDID, + "vote_count", len(votes), + "expires_at", c.expiry[userDID]) +} + +// SetVote adds or updates a single vote in the cache +func (c *VoteCache) SetVote(userDID, subjectURI string, vote *CachedVote) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.votes[userDID] == nil { + c.votes[userDID] = make(map[string]*CachedVote) + } + + c.votes[userDID][subjectURI] = vote + + // Always extend expiry on vote action - active users keep their cache fresh + c.expiry[userDID] = time.Now().Add(c.ttl) + + c.logger.Debug("vote cached", + "user", userDID, + "subject", subjectURI, + "direction", vote.Direction) +} + +// RemoveVote removes a vote from the cache (for toggle-off) +func (c *VoteCache) RemoveVote(userDID, subjectURI string) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.votes[userDID] != nil { + delete(c.votes[userDID], subjectURI) + + // Extend expiry on vote action - active users keep their cache fresh + c.expiry[userDID] = time.Now().Add(c.ttl) + + c.logger.Debug("vote removed from cache", + "user", userDID, + "subject", subjectURI) + } +} + +// Invalidate removes all cached votes for a user +func (c *VoteCache) Invalidate(userDID string) { + c.mu.Lock() + defer c.mu.Unlock() + + delete(c.votes, userDID) + delete(c.expiry, userDID) + + c.logger.Debug("vote cache invalidated", "user", userDID) +} + +// FetchAndCacheFromPDS fetches all votes from the user's PDS and caches them. +// This should be called on first authenticated request or when cache is expired. +func (c *VoteCache) FetchAndCacheFromPDS(ctx context.Context, pdsClient pds.Client) error { + userDID := pdsClient.DID() + + c.logger.Debug("fetching votes from PDS", + "user", userDID, + "pds", pdsClient.HostURL()) + + votes, err := c.fetchAllVotesFromPDS(ctx, pdsClient) + if err != nil { + return fmt.Errorf("failed to fetch votes from PDS: %w", err) + } + + c.SetVotesForUser(userDID, votes) + + c.logger.Info("vote cache populated from PDS", + "user", userDID, + "vote_count", len(votes)) + + return nil +} + +// fetchAllVotesFromPDS paginates through all vote records on the user's PDS +func (c *VoteCache) fetchAllVotesFromPDS(ctx context.Context, pdsClient pds.Client) (map[string]*CachedVote, error) { + votes := make(map[string]*CachedVote) + cursor := "" + const pageSize = 100 + const collection = "social.coves.feed.vote" + + for { + result, err := pdsClient.ListRecords(ctx, collection, pageSize, cursor) + if err != nil { + if pds.IsAuthError(err) { + return nil, ErrNotAuthorized + } + return nil, fmt.Errorf("listRecords failed: %w", err) + } + + for _, rec := range result.Records { + // Extract subject from record value + subject, ok := rec.Value["subject"].(map[string]any) + if !ok { + continue + } + + subjectURI, ok := subject["uri"].(string) + if !ok || subjectURI == "" { + continue + } + + direction, _ := rec.Value["direction"].(string) + if direction == "" { + continue + } + + // Extract rkey from URI + rkey := extractRKeyFromURI(rec.URI) + + votes[subjectURI] = &CachedVote{ + Direction: direction, + URI: rec.URI, + RKey: rkey, + } + } + + if result.Cursor == "" { + break + } + cursor = result.Cursor + } + + return votes, nil +} + +// extractRKeyFromURI extracts the rkey from an AT-URI (at://did/collection/rkey) +func extractRKeyFromURI(uri string) string { + parts := strings.Split(uri, "/") + if len(parts) >= 5 { + return parts[len(parts)-1] + } + return "" +} diff --git a/internal/core/votes/service.go b/internal/core/votes/service.go index 3a0b702..835e19f 100644 --- a/internal/core/votes/service.go +++ b/internal/core/votes/service.go @@ -44,6 +44,20 @@ type Service interface { // - Deletes the user's vote record from their PDS // - AppView will soft-delete via Jetstream consumer DeleteVote(ctx context.Context, session *oauthlib.ClientSessionData, req DeleteVoteRequest) error + + // EnsureCachePopulated fetches the user's votes from their PDS if not already cached. + // This should be called before rendering feeds to ensure vote state is available. + // If cache is already populated and not expired, this is a no-op. + EnsureCachePopulated(ctx context.Context, session *oauthlib.ClientSessionData) error + + // GetViewerVote returns the viewer's vote for a specific subject, or nil if not voted. + // Returns from cache if available, otherwise returns nil (caller should ensure cache is populated). + GetViewerVote(userDID, subjectURI string) *CachedVote + + // GetViewerVotesForSubjects returns the viewer's votes for multiple subjects. + // Returns a map of subjectURI -> CachedVote for subjects the user has voted on. + // This is efficient for batch lookups when rendering feeds. + GetViewerVotesForSubjects(userDID string, subjectURIs []string) map[string]*CachedVote } // CreateVoteRequest contains the parameters for creating a vote diff --git a/internal/core/votes/service_impl.go b/internal/core/votes/service_impl.go index 6a64e5b..e0a3003 100644 --- a/internal/core/votes/service_impl.go +++ b/internal/core/votes/service_impl.go @@ -30,10 +30,11 @@ type voteService struct { oauthStore oauth.ClientAuthStore logger *slog.Logger pdsClientFactory PDSClientFactory // Optional, for testing. If nil, uses OAuth. + cache *VoteCache // In-memory cache of user votes from PDS } // NewService creates a new vote service instance -func NewService(repo Repository, oauthClient *oauthclient.OAuthClient, oauthStore oauth.ClientAuthStore, logger *slog.Logger) Service { +func NewService(repo Repository, oauthClient *oauthclient.OAuthClient, oauthStore oauth.ClientAuthStore, cache *VoteCache, logger *slog.Logger) Service { if logger == nil { logger = slog.Default() } @@ -41,18 +42,20 @@ func NewService(repo Repository, oauthClient *oauthclient.OAuthClient, oauthStor repo: repo, oauthClient: oauthClient, oauthStore: oauthStore, + cache: cache, logger: logger, } } // NewServiceWithPDSFactory creates a vote service with a custom PDS client factory. // This is primarily for testing with password-based authentication. -func NewServiceWithPDSFactory(repo Repository, logger *slog.Logger, factory PDSClientFactory) Service { +func NewServiceWithPDSFactory(repo Repository, cache *VoteCache, logger *slog.Logger, factory PDSClientFactory) Service { if logger == nil { logger = slog.Default() } return &voteService{ repo: repo, + cache: cache, logger: logger, pdsClientFactory: factory, } @@ -150,6 +153,11 @@ func (s *voteService) CreateVote(ctx context.Context, session *oauth.ClientSessi "subject", req.Subject.URI, "direction", req.Direction) + // Update cache - remove the vote + if s.cache != nil { + s.cache.RemoveVote(session.AccountDID.String(), req.Subject.URI) + } + // Return empty response to indicate deletion return &CreateVoteResponse{ URI: "", @@ -197,6 +205,15 @@ func (s *voteService) CreateVote(ctx context.Context, session *oauth.ClientSessi "uri", uri, "cid", cid) + // Update cache - add the new vote + if s.cache != nil { + s.cache.SetVote(session.AccountDID.String(), req.Subject.URI, &CachedVote{ + Direction: req.Direction, + URI: uri, + RKey: extractRKeyFromURI(uri), + }) + } + return &CreateVoteResponse{ URI: uri, CID: cid, @@ -253,6 +270,11 @@ func (s *voteService) DeleteVote(ctx context.Context, session *oauth.ClientSessi "subject", req.Subject.URI, "uri", existing.URI) + // Update cache - remove the vote + if s.cache != nil { + s.cache.RemoveVote(session.AccountDID.String(), req.Subject.URI) + } + return nil } @@ -350,3 +372,63 @@ func (s *voteService) findExistingVote(ctx context.Context, pdsClient pds.Client // No vote found for this subject after checking all pages return nil, nil } + +// EnsureCachePopulated fetches the user's votes from their PDS if not already cached. +func (s *voteService) EnsureCachePopulated(ctx context.Context, session *oauth.ClientSessionData) error { + if s.cache == nil { + return nil // No cache configured + } + + // Check if already cached + if s.cache.IsCached(session.AccountDID.String()) { + return nil + } + + // Create PDS client for this session + pdsClient, err := s.getPDSClient(ctx, session) + if err != nil { + s.logger.Error("failed to create PDS client for cache population", + "error", err, + "user", session.AccountDID) + return fmt.Errorf("failed to create PDS client: %w", err) + } + + // Fetch and cache votes from PDS + if err := s.cache.FetchAndCacheFromPDS(ctx, pdsClient); err != nil { + s.logger.Error("failed to populate vote cache from PDS", + "error", err, + "user", session.AccountDID) + return fmt.Errorf("failed to populate vote cache: %w", err) + } + + return nil +} + +// GetViewerVote returns the viewer's vote for a specific subject, or nil if not voted. +func (s *voteService) GetViewerVote(userDID, subjectURI string) *CachedVote { + if s.cache == nil { + return nil + } + return s.cache.GetVote(userDID, subjectURI) +} + +// GetViewerVotesForSubjects returns the viewer's votes for multiple subjects. +func (s *voteService) GetViewerVotesForSubjects(userDID string, subjectURIs []string) map[string]*CachedVote { + result := make(map[string]*CachedVote) + if s.cache == nil { + return result + } + + allVotes := s.cache.GetVotesForUser(userDID) + if allVotes == nil { + return result + } + + for _, uri := range subjectURIs { + if vote, exists := allVotes[uri]; exists { + result[uri] = vote + } + } + + return result +} diff --git a/tests/integration/vote_e2e_test.go b/tests/integration/vote_e2e_test.go index d1c827a..49002b7 100644 --- a/tests/integration/vote_e2e_test.go +++ b/tests/integration/vote_e2e_test.go @@ -81,7 +81,7 @@ func TestVoteE2E_CreateUpvote(t *testing.T) { postRepo := postgres.NewPostRepository(db) // Setup services with password-based PDS client factory for E2E testing - voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, PasswordAuthPDSClientFactory()) + voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory()) // Create test user on PDS testUserHandle := fmt.Sprintf("voter-%d.local.coves.dev", time.Now().Unix()) @@ -304,7 +304,7 @@ func TestVoteE2E_ToggleSameDirection(t *testing.T) { voteRepo := postgres.NewVoteRepository(db) postRepo := postgres.NewPostRepository(db) - voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, PasswordAuthPDSClientFactory()) + voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory()) // Create test user testUserHandle := fmt.Sprintf("toggle-%d.local.coves.dev", time.Now().Unix()) @@ -473,7 +473,7 @@ func TestVoteE2E_ToggleDifferentDirection(t *testing.T) { voteRepo := postgres.NewVoteRepository(db) postRepo := postgres.NewPostRepository(db) - voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, PasswordAuthPDSClientFactory()) + voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory()) // Create test user testUserHandle := fmt.Sprintf("flip-%d.local.coves.dev", time.Now().Unix()) @@ -698,7 +698,7 @@ func TestVoteE2E_DeleteVote(t *testing.T) { voteRepo := postgres.NewVoteRepository(db) postRepo := postgres.NewPostRepository(db) - voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, PasswordAuthPDSClientFactory()) + voteService := votes.NewServiceWithPDSFactory(voteRepo, nil, nil, PasswordAuthPDSClientFactory()) // Create test user testUserHandle := fmt.Sprintf("delete-%d.local.coves.dev", time.Now().Unix())