A community based topic aggregation platform built on atproto
bretton.dev
1package main
2
3import (
4 "Coves/internal/api/middleware"
5 "Coves/internal/api/routes"
6 "Coves/internal/atproto/auth"
7 "Coves/internal/atproto/identity"
8 "Coves/internal/atproto/jetstream"
9 "Coves/internal/core/aggregators"
10 "Coves/internal/core/communities"
11 "Coves/internal/core/communityFeeds"
12 "Coves/internal/core/discover"
13 "Coves/internal/core/posts"
14 "Coves/internal/core/timeline"
15 "Coves/internal/core/users"
16 "bytes"
17 "context"
18 "database/sql"
19 "encoding/json"
20 "fmt"
21 "io"
22 "log"
23 "net/http"
24 "os"
25 "strings"
26 "time"
27
28 "github.com/go-chi/chi/v5"
29 chiMiddleware "github.com/go-chi/chi/v5/middleware"
30 _ "github.com/lib/pq"
31 "github.com/pressly/goose/v3"
32
33 postgresRepo "Coves/internal/db/postgres"
34)
35
36func main() {
37 // Database configuration (AppView database)
38 dbURL := os.Getenv("DATABASE_URL")
39 if dbURL == "" {
40 // Use dev database from .env.dev
41 dbURL = "postgres://dev_user:dev_password@localhost:5435/coves_dev?sslmode=disable"
42 }
43
44 // Default PDS URL for this Coves instance (supports self-hosting)
45 defaultPDS := os.Getenv("PDS_URL")
46 if defaultPDS == "" {
47 defaultPDS = "http://localhost:3001" // Local dev PDS
48 }
49
50 // Cursor secret for HMAC signing (prevents cursor manipulation)
51 cursorSecret := os.Getenv("CURSOR_SECRET")
52 if cursorSecret == "" {
53 // Generate a random secret if not set (dev mode)
54 // IMPORTANT: In production, set CURSOR_SECRET to a strong random value
55 cursorSecret = "dev-cursor-secret-change-in-production"
56 log.Println("⚠️ WARNING: Using default cursor secret. Set CURSOR_SECRET env var in production!")
57 }
58
59 db, err := sql.Open("postgres", dbURL)
60 if err != nil {
61 log.Fatal("Failed to connect to database:", err)
62 }
63 defer func() {
64 if closeErr := db.Close(); closeErr != nil {
65 log.Printf("Failed to close database connection: %v", closeErr)
66 }
67 }()
68
69 if err = db.Ping(); err != nil {
70 log.Fatal("Failed to ping database:", err)
71 }
72
73 log.Println("Connected to AppView database")
74
75 // Run migrations
76 if err = goose.SetDialect("postgres"); err != nil {
77 log.Fatal("Failed to set goose dialect:", err)
78 }
79
80 if err = goose.Up(db, "internal/db/migrations"); err != nil {
81 log.Fatal("Failed to run migrations:", err)
82 }
83
84 log.Println("Migrations completed successfully")
85
86 r := chi.NewRouter()
87
88 r.Use(chiMiddleware.Logger)
89 r.Use(chiMiddleware.Recoverer)
90 r.Use(chiMiddleware.RequestID)
91
92 // Rate limiting: 100 requests per minute per IP
93 rateLimiter := middleware.NewRateLimiter(100, 1*time.Minute)
94 r.Use(rateLimiter.Middleware)
95
96 // Initialize identity resolver
97 // IMPORTANT: In dev mode, identity resolution MUST use the same local PLC
98 // directory as DID registration to ensure E2E tests work without hitting
99 // the production plc.directory
100 identityConfig := identity.DefaultConfig()
101
102 isDevEnv := os.Getenv("IS_DEV_ENV") == "true"
103 plcDirectoryURL := os.Getenv("PLC_DIRECTORY_URL")
104 if plcDirectoryURL == "" {
105 plcDirectoryURL = "https://plc.directory" // Default to production PLC
106 }
107
108 // In dev mode, use PLC_DIRECTORY_URL for identity resolution
109 // In prod mode, use IDENTITY_PLC_URL if set, otherwise PLC_DIRECTORY_URL
110 if isDevEnv {
111 identityConfig.PLCURL = plcDirectoryURL
112 log.Printf("🧪 DEV MODE: Identity resolver will use local PLC: %s", plcDirectoryURL)
113 } else {
114 // Production: Allow separate IDENTITY_PLC_URL for read operations
115 if identityPLCURL := os.Getenv("IDENTITY_PLC_URL"); identityPLCURL != "" {
116 identityConfig.PLCURL = identityPLCURL
117 } else {
118 identityConfig.PLCURL = plcDirectoryURL
119 }
120 log.Printf("✅ PRODUCTION MODE: Identity resolver using PLC: %s", identityConfig.PLCURL)
121 }
122
123 if cacheTTL := os.Getenv("IDENTITY_CACHE_TTL"); cacheTTL != "" {
124 if duration, parseErr := time.ParseDuration(cacheTTL); parseErr == nil {
125 identityConfig.CacheTTL = duration
126 }
127 }
128
129 identityResolver := identity.NewResolver(db, identityConfig)
130
131 // Initialize atProto auth middleware for JWT validation
132 // Phase 1: Set skipVerify=true to test JWT parsing only
133 // Phase 2: Set skipVerify=false to enable full signature verification
134 skipVerify := os.Getenv("AUTH_SKIP_VERIFY") == "true"
135 if skipVerify {
136 log.Println("⚠️ WARNING: JWT signature verification is DISABLED (Phase 1 testing)")
137 log.Println(" Set AUTH_SKIP_VERIFY=false for production")
138 }
139
140 jwksCacheTTL := 1 * time.Hour // Cache public keys for 1 hour
141 jwksFetcher := auth.NewCachedJWKSFetcher(jwksCacheTTL)
142 authMiddleware := middleware.NewAtProtoAuthMiddleware(jwksFetcher, skipVerify)
143 log.Println("✅ atProto auth middleware initialized")
144
145 // Initialize repositories and services
146 userRepo := postgresRepo.NewUserRepository(db)
147 userService := users.NewUserService(userRepo, identityResolver, defaultPDS)
148
149 communityRepo := postgresRepo.NewCommunityRepository(db)
150
151 // V2.0: PDS-managed DID generation
152 // Community DIDs and keys are generated entirely by the PDS
153 // No Coves-side DID generator needed (reserved for future V2.1 hybrid approach)
154
155 instanceDID := os.Getenv("INSTANCE_DID")
156 if instanceDID == "" {
157 instanceDID = "did:web:coves.social" // Default for development
158 }
159
160 // V2: Extract instance domain for community handles
161 // IMPORTANT: This MUST match the domain in INSTANCE_DID for security
162 // We cannot allow arbitrary domains to prevent impersonation attacks
163 // Example attack: !leagueoflegends@riotgames.com on a non-Riot instance
164 //
165 // SECURITY: did:web domain verification is implemented in the Jetstream consumer
166 // See: internal/atproto/jetstream/community_consumer.go - verifyHostedByClaim()
167 // Communities with mismatched hostedBy domains are rejected during indexing
168 var instanceDomain string
169 if strings.HasPrefix(instanceDID, "did:web:") {
170 // Extract domain from did:web (this is the authoritative source)
171 instanceDomain = strings.TrimPrefix(instanceDID, "did:web:")
172 } else {
173 // For non-web DIDs (e.g., did:plc), require explicit INSTANCE_DOMAIN
174 instanceDomain = os.Getenv("INSTANCE_DOMAIN")
175 if instanceDomain == "" {
176 log.Fatal("INSTANCE_DOMAIN must be set for non-web DIDs")
177 }
178 }
179
180 log.Printf("Instance domain: %s (extracted from DID: %s)", instanceDomain, instanceDID)
181
182 // V2.0: Initialize PDS account provisioner for communities (simplified)
183 // PDS handles all DID and key generation - no Coves-side cryptography needed
184 provisioner := communities.NewPDSAccountProvisioner(instanceDomain, defaultPDS)
185 log.Printf("✅ Community provisioner initialized (PDS-managed keys)")
186 log.Printf(" - Communities will be created at: %s", defaultPDS)
187 log.Printf(" - PDS will generate and manage all DIDs and keys")
188
189 // Initialize community service (no longer needs didGenerator directly)
190 communityService := communities.NewCommunityService(communityRepo, defaultPDS, instanceDID, instanceDomain, provisioner)
191
192 // Authenticate Coves instance with PDS to enable community record writes
193 // The instance needs a PDS account to write community records it owns
194 pdsHandle := os.Getenv("PDS_INSTANCE_HANDLE")
195 pdsPassword := os.Getenv("PDS_INSTANCE_PASSWORD")
196 if pdsHandle != "" && pdsPassword != "" {
197 log.Printf("Authenticating Coves instance (%s) with PDS...", instanceDID)
198 accessToken, authErr := authenticateWithPDS(defaultPDS, pdsHandle, pdsPassword)
199 if authErr != nil {
200 log.Printf("Warning: Failed to authenticate with PDS: %v", authErr)
201 log.Println("Community creation will fail until PDS authentication is configured")
202 } else {
203 if svc, ok := communityService.(interface{ SetPDSAccessToken(string) }); ok {
204 svc.SetPDSAccessToken(accessToken)
205 log.Println("✓ Coves instance authenticated with PDS")
206 }
207 }
208 } else {
209 log.Println("Note: PDS_INSTANCE_HANDLE and PDS_INSTANCE_PASSWORD not set")
210 log.Println("Community creation via write-forward is disabled")
211 }
212
213 // Start Jetstream consumer for read-forward user indexing
214 jetstreamURL := os.Getenv("JETSTREAM_URL")
215 if jetstreamURL == "" {
216 jetstreamURL = "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.actor.profile"
217 }
218
219 pdsFilter := os.Getenv("JETSTREAM_PDS_FILTER") // Optional: filter to specific PDS
220
221 userConsumer := jetstream.NewUserEventConsumer(userService, identityResolver, jetstreamURL, pdsFilter)
222 ctx := context.Background()
223 go func() {
224 if startErr := userConsumer.Start(ctx); startErr != nil {
225 log.Printf("Jetstream consumer stopped: %v", startErr)
226 }
227 }()
228
229 log.Printf("Started Jetstream user consumer: %s", jetstreamURL)
230
231 // Start Jetstream consumer for community events (profiles and subscriptions)
232 // This consumer indexes:
233 // 1. Community profiles (social.coves.community.profile) - in community's own repo
234 // 2. User subscriptions (social.coves.community.subscription) - in user's repo
235 communityJetstreamURL := os.Getenv("COMMUNITY_JETSTREAM_URL")
236 if communityJetstreamURL == "" {
237 // Local Jetstream for communities - filter to our instance's collections
238 // IMPORTANT: We listen to social.coves.community.subscription (not social.coves.community.subscribe)
239 // because subscriptions are RECORD TYPES in the communities namespace, not XRPC procedures
240 communityJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.profile&wantedCollections=social.coves.community.subscription"
241 }
242
243 // Initialize community event consumer with did:web verification
244 skipDIDWebVerification := os.Getenv("SKIP_DID_WEB_VERIFICATION") == "true"
245 if skipDIDWebVerification {
246 log.Println("⚠️ WARNING: did:web domain verification is DISABLED (dev mode)")
247 log.Println(" Set SKIP_DID_WEB_VERIFICATION=false for production")
248 }
249
250 communityEventConsumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, skipDIDWebVerification)
251 communityJetstreamConnector := jetstream.NewCommunityJetstreamConnector(communityEventConsumer, communityJetstreamURL)
252
253 go func() {
254 if startErr := communityJetstreamConnector.Start(ctx); startErr != nil {
255 log.Printf("Community Jetstream consumer stopped: %v", startErr)
256 }
257 }()
258
259 log.Printf("Started Jetstream community consumer: %s", communityJetstreamURL)
260 log.Println(" - Indexing: social.coves.community.profile (community profiles)")
261 log.Println(" - Indexing: social.coves.community.subscription (user subscriptions)")
262
263 // Start JWKS cache cleanup background job
264 go func() {
265 ticker := time.NewTicker(1 * time.Hour)
266 defer ticker.Stop()
267 for range ticker.C {
268 jwksFetcher.CleanupExpiredCache()
269 log.Println("JWKS cache cleanup completed")
270 }
271 }()
272
273 log.Println("Started JWKS cache cleanup background job (runs hourly)")
274
275 // Initialize aggregator service
276 aggregatorRepo := postgresRepo.NewAggregatorRepository(db)
277 aggregatorService := aggregators.NewAggregatorService(aggregatorRepo, communityService)
278 log.Println("✅ Aggregator service initialized")
279
280 // Initialize post service (with aggregator support)
281 postRepo := postgresRepo.NewPostRepository(db)
282 postService := posts.NewPostService(postRepo, communityService, aggregatorService, defaultPDS)
283
284 // Initialize vote repository (used by Jetstream consumer for indexing)
285 voteRepo := postgresRepo.NewVoteRepository(db)
286 log.Println("✅ Vote repository initialized (Jetstream indexing only)")
287
288 // Initialize feed service
289 feedRepo := postgresRepo.NewCommunityFeedRepository(db)
290 feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
291 log.Println("✅ Feed service initialized")
292
293 // Initialize timeline service (home feed from subscribed communities)
294 timelineRepo := postgresRepo.NewTimelineRepository(db, cursorSecret)
295 timelineService := timeline.NewTimelineService(timelineRepo)
296 log.Println("✅ Timeline service initialized")
297
298 // Initialize discover service (public feed from all communities)
299 discoverRepo := postgresRepo.NewDiscoverRepository(db, cursorSecret)
300 discoverService := discover.NewDiscoverService(discoverRepo)
301 log.Println("✅ Discover service initialized")
302
303 // Start Jetstream consumer for posts
304 // This consumer indexes posts created in community repositories via the firehose
305 // Currently handles only CREATE operations - UPDATE/DELETE deferred until those features exist
306 postJetstreamURL := os.Getenv("POST_JETSTREAM_URL")
307 if postJetstreamURL == "" {
308 // Listen to post record creation events
309 postJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.post.record"
310 }
311
312 postEventConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)
313 postJetstreamConnector := jetstream.NewPostJetstreamConnector(postEventConsumer, postJetstreamURL)
314
315 go func() {
316 if startErr := postJetstreamConnector.Start(ctx); startErr != nil {
317 log.Printf("Post Jetstream consumer stopped: %v", startErr)
318 }
319 }()
320
321 log.Printf("Started Jetstream post consumer: %s", postJetstreamURL)
322 log.Println(" - Indexing: social.coves.post.record CREATE operations")
323 log.Println(" - UPDATE/DELETE indexing deferred until those features are implemented")
324
325 // Start Jetstream consumer for aggregators
326 // This consumer indexes aggregator service declarations and authorization records
327 // Following Bluesky's pattern for feed generators and labelers
328 // NOTE: Uses the same Jetstream as communities, just filtering different collections
329 aggregatorJetstreamURL := communityJetstreamURL
330 // Override if specific URL needed for testing
331 if envURL := os.Getenv("AGGREGATOR_JETSTREAM_URL"); envURL != "" {
332 aggregatorJetstreamURL = envURL
333 } else if aggregatorJetstreamURL == "" {
334 // Fallback if community URL also not set
335 aggregatorJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.aggregator.service&wantedCollections=social.coves.aggregator.authorization"
336 }
337
338 aggregatorEventConsumer := jetstream.NewAggregatorEventConsumer(aggregatorRepo)
339 aggregatorJetstreamConnector := jetstream.NewAggregatorJetstreamConnector(aggregatorEventConsumer, aggregatorJetstreamURL)
340
341 go func() {
342 if startErr := aggregatorJetstreamConnector.Start(ctx); startErr != nil {
343 log.Printf("Aggregator Jetstream consumer stopped: %v", startErr)
344 }
345 }()
346
347 log.Printf("Started Jetstream aggregator consumer: %s", aggregatorJetstreamURL)
348 log.Println(" - Indexing: social.coves.aggregator.service (service declarations)")
349 log.Println(" - Indexing: social.coves.aggregator.authorization (authorization records)")
350
351 // Start Jetstream consumer for votes
352 // This consumer indexes votes from user repositories and updates post vote counts
353 voteJetstreamURL := os.Getenv("VOTE_JETSTREAM_URL")
354 if voteJetstreamURL == "" {
355 // Listen to vote record CREATE/DELETE events from user repositories
356 voteJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.interaction.vote"
357 }
358
359 voteEventConsumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db)
360 voteJetstreamConnector := jetstream.NewVoteJetstreamConnector(voteEventConsumer, voteJetstreamURL)
361
362 go func() {
363 if startErr := voteJetstreamConnector.Start(ctx); startErr != nil {
364 log.Printf("Vote Jetstream consumer stopped: %v", startErr)
365 }
366 }()
367
368 log.Printf("Started Jetstream vote consumer: %s", voteJetstreamURL)
369 log.Println(" - Indexing: social.coves.interaction.vote CREATE/DELETE operations")
370 log.Println(" - Updating: Post vote counts atomically")
371
372 // Register XRPC routes
373 routes.RegisterUserRoutes(r, userService)
374 routes.RegisterCommunityRoutes(r, communityService, authMiddleware)
375 log.Println("Community XRPC endpoints registered with OAuth authentication")
376
377 routes.RegisterPostRoutes(r, postService, authMiddleware)
378 log.Println("Post XRPC endpoints registered with OAuth authentication")
379
380 // Vote write endpoints removed - clients write directly to their PDS
381 // The AppView indexes votes from Jetstream (see vote consumer above)
382
383 routes.RegisterCommunityFeedRoutes(r, feedService)
384 log.Println("Feed XRPC endpoints registered (public, no auth required)")
385
386 routes.RegisterTimelineRoutes(r, timelineService, authMiddleware)
387 log.Println("Timeline XRPC endpoints registered (requires authentication)")
388
389 routes.RegisterDiscoverRoutes(r, discoverService)
390 log.Println("Discover XRPC endpoints registered (public, no auth required)")
391
392 routes.RegisterAggregatorRoutes(r, aggregatorService)
393 log.Println("Aggregator XRPC endpoints registered (query endpoints public)")
394
395 r.Get("/health", func(w http.ResponseWriter, r *http.Request) {
396 w.WriteHeader(http.StatusOK)
397 if _, err := w.Write([]byte("OK")); err != nil {
398 log.Printf("Failed to write health check response: %v", err)
399 }
400 })
401
402 port := os.Getenv("APPVIEW_PORT")
403 if port == "" {
404 port = "8081" // Match .env.dev default
405 }
406
407 fmt.Printf("Coves AppView starting on port %s\n", port)
408 fmt.Printf("Default PDS: %s\n", defaultPDS)
409 log.Fatal(http.ListenAndServe(":"+port, r))
410}
411
412// authenticateWithPDS creates a session on the PDS and returns an access token
413func authenticateWithPDS(pdsURL, handle, password string) (string, error) {
414 type CreateSessionRequest struct {
415 Identifier string `json:"identifier"`
416 Password string `json:"password"`
417 }
418
419 type CreateSessionResponse struct {
420 DID string `json:"did"`
421 Handle string `json:"handle"`
422 AccessJwt string `json:"accessJwt"`
423 }
424
425 reqBody, err := json.Marshal(CreateSessionRequest{
426 Identifier: handle,
427 Password: password,
428 })
429 if err != nil {
430 return "", fmt.Errorf("failed to marshal request: %w", err)
431 }
432
433 resp, err := http.Post(
434 pdsURL+"/xrpc/com.atproto.server.createSession",
435 "application/json",
436 bytes.NewReader(reqBody),
437 )
438 if err != nil {
439 return "", fmt.Errorf("failed to call PDS: %w", err)
440 }
441 defer func() {
442 if closeErr := resp.Body.Close(); closeErr != nil {
443 log.Printf("Failed to close response body: %v", closeErr)
444 }
445 }()
446
447 if resp.StatusCode != http.StatusOK {
448 body, readErr := io.ReadAll(resp.Body)
449 if readErr != nil {
450 return "", fmt.Errorf("PDS returned status %d and failed to read body: %w", resp.StatusCode, readErr)
451 }
452 return "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
453 }
454
455 var session CreateSessionResponse
456 if err := json.NewDecoder(resp.Body).Decode(&session); err != nil {
457 return "", fmt.Errorf("failed to decode response: %w", err)
458 }
459
460 return session.AccessJwt, nil
461}