bretton.dev / coves
A community based topic aggregation platform built on atproto
fork atom

feat(server): wire aggregator system into application

Integrate aggregator components into server initialization and
register XRPC endpoints.

Changes to cmd/server/main.go:
- Initialize aggregator repository
- Initialize aggregator service (with community service dependency)
- Update post service to include aggregator service
- Register aggregator XRPC routes (3 query endpoints)
- Start aggregator Jetstream consumer in background goroutine
- Add comprehensive startup logging

Server startup output:
✅ Aggregator service initialized
Started Jetstream aggregator consumer: ws://localhost:6008/subscribe?...
- Indexing: social.coves.aggregator.service (service declarations)
- Indexing: social.coves.aggregator.authorization (authorization records)
Aggregator XRPC endpoints registered (query endpoints public)

Architecture:
- Aggregator service depends on: aggregator repo, community service
- Post service depends on: aggregator service (for auth checks)
- Jetstream consumer runs independently, indexes to DB via repository
- XRPC handlers call service layer methods

Phase 1 complete:
✅ Aggregators can authenticate (via JWT)
✅ Aggregators can post to authorized communities
✅ Rate limiting enforced (10 posts/hour per community)
✅ Query endpoints available for discovery

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

bretton.dev 87fc8db0 9689ffe7

+37 -2
unified split
+37 -2
cmd/server/main.go
··· 6 6 "Coves/internal/atproto/auth" 7 7 "Coves/internal/atproto/identity" 8 8 "Coves/internal/atproto/jetstream" 9 + "Coves/internal/core/aggregators" 9 10 "Coves/internal/core/communities" 10 11 "Coves/internal/core/communityFeeds" 11 12 "Coves/internal/core/posts" ··· 260 261 261 262 log.Println("Started JWKS cache cleanup background job (runs hourly)") 262 263 263 - // Initialize post service 264 + // Initialize aggregator service 265 + aggregatorRepo := postgresRepo.NewAggregatorRepository(db) 266 + aggregatorService := aggregators.NewAggregatorService(aggregatorRepo, communityService) 267 + log.Println("✅ Aggregator service initialized") 268 + 269 + // Initialize post service (with aggregator support) 264 270 postRepo := postgresRepo.NewPostRepository(db) 265 - postService := posts.NewPostService(postRepo, communityService, defaultPDS) 271 + postService := posts.NewPostService(postRepo, communityService, aggregatorService, defaultPDS) 266 272 267 273 // Initialize feed service 268 274 feedRepo := postgresRepo.NewCommunityFeedRepository(db) ··· 291 297 log.Println(" - Indexing: social.coves.post.record CREATE operations") 292 298 log.Println(" - UPDATE/DELETE indexing deferred until those features are implemented") 293 299 300 + // Start Jetstream consumer for aggregators 301 + // This consumer indexes aggregator service declarations and authorization records 302 + // Following Bluesky's pattern for feed generators and labelers 303 + // NOTE: Uses the same Jetstream as communities, just filtering different collections 304 + aggregatorJetstreamURL := communityJetstreamURL 305 + // Override if specific URL needed for testing 306 + if envURL := os.Getenv("AGGREGATOR_JETSTREAM_URL"); envURL != "" { 307 + aggregatorJetstreamURL = envURL 308 + } else if aggregatorJetstreamURL == "" { 309 + // Fallback if community URL also not set 310 + aggregatorJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.aggregator.service&wantedCollections=social.coves.aggregator.authorization" 311 + } 312 + 313 + aggregatorEventConsumer := jetstream.NewAggregatorEventConsumer(aggregatorRepo) 314 + aggregatorJetstreamConnector := jetstream.NewAggregatorJetstreamConnector(aggregatorEventConsumer, aggregatorJetstreamURL) 315 + 316 + go func() { 317 + if startErr := aggregatorJetstreamConnector.Start(ctx); startErr != nil { 318 + log.Printf("Aggregator Jetstream consumer stopped: %v", startErr) 319 + } 320 + }() 321 + 322 + log.Printf("Started Jetstream aggregator consumer: %s", aggregatorJetstreamURL) 323 + log.Println(" - Indexing: social.coves.aggregator.service (service declarations)") 324 + log.Println(" - Indexing: social.coves.aggregator.authorization (authorization records)") 325 + 294 326 // Register XRPC routes 295 327 routes.RegisterUserRoutes(r, userService) 296 328 routes.RegisterCommunityRoutes(r, communityService, authMiddleware) ··· 301 333 302 334 routes.RegisterCommunityFeedRoutes(r, feedService) 303 335 log.Println("Feed XRPC endpoints registered (public, no auth required)") 336 + 337 + routes.RegisterAggregatorRoutes(r, aggregatorService) 338 + log.Println("Aggregator XRPC endpoints registered (query endpoints public)") 304 339 305 340 r.Get("/health", func(w http.ResponseWriter, r *http.Request) { 306 341 w.WriteHeader(http.StatusOK)