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

feat(server): integrate post creation endpoints and consumer

Wire up post creation feature in main server:

1. Initialize post service
- Create post repository (PostgreSQL)
- Create post service with community service integration
- Configure with default PDS URL

2. Register XRPC routes
- POST /xrpc/social.coves.post.create
- Requires OAuth authentication via middleware

3. Start Jetstream consumer for posts
- WebSocket URL: ws://localhost:6008/subscribe
- Collection filter: social.coves.post.record
- Runs in background goroutine
- Indexes CREATE operations (UPDATE/DELETE deferred)

Environment variables:
- POST_JETSTREAM_URL: Override default Jetstream endpoint

Initialization order ensures postRepo created before consumer.

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

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

bretton.dev 3db7903e 98fac089

+30
+30
cmd/server/main.go
··· 7 7 "Coves/internal/atproto/identity" 8 8 "Coves/internal/atproto/jetstream" 9 9 "Coves/internal/core/communities" 10 + "Coves/internal/core/posts" 10 11 "Coves/internal/core/users" 11 12 "bytes" 12 13 "context" ··· 258 259 259 260 log.Println("Started JWKS cache cleanup background job (runs hourly)") 260 261 262 + // Initialize post service 263 + postRepo := postgresRepo.NewPostRepository(db) 264 + postService := posts.NewPostService(postRepo, communityService, defaultPDS) 265 + 266 + // Start Jetstream consumer for posts 267 + // This consumer indexes posts created in community repositories via the firehose 268 + // Currently handles only CREATE operations - UPDATE/DELETE deferred until those features exist 269 + postJetstreamURL := os.Getenv("POST_JETSTREAM_URL") 270 + if postJetstreamURL == "" { 271 + // Listen to post record creation events 272 + postJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.post.record" 273 + } 274 + 275 + postEventConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService) 276 + postJetstreamConnector := jetstream.NewPostJetstreamConnector(postEventConsumer, postJetstreamURL) 277 + 278 + go func() { 279 + if startErr := postJetstreamConnector.Start(ctx); startErr != nil { 280 + log.Printf("Post Jetstream consumer stopped: %v", startErr) 281 + } 282 + }() 283 + 284 + log.Printf("Started Jetstream post consumer: %s", postJetstreamURL) 285 + log.Println(" - Indexing: social.coves.post.record CREATE operations") 286 + log.Println(" - UPDATE/DELETE indexing deferred until those features are implemented") 287 + 261 288 // Register XRPC routes 262 289 routes.RegisterUserRoutes(r, userService) 263 290 routes.RegisterCommunityRoutes(r, communityService, authMiddleware) 264 291 log.Println("Community XRPC endpoints registered with OAuth authentication") 292 + 293 + routes.RegisterPostRoutes(r, postService, authMiddleware) 294 + log.Println("Post XRPC endpoints registered with OAuth authentication") 265 295 266 296 r.Get("/health", func(w http.ResponseWriter, r *http.Request) { 267 297 w.WriteHeader(http.StatusOK)