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 feed service
285 feedRepo := postgresRepo.NewCommunityFeedRepository(db)
286 feedService := communityFeeds.NewCommunityFeedService(feedRepo, communityService)
287 log.Println("✅ Feed service initialized")
288
289 // Initialize timeline service (home feed from subscribed communities)
290 timelineRepo := postgresRepo.NewTimelineRepository(db, cursorSecret)
291 timelineService := timeline.NewTimelineService(timelineRepo)
292 log.Println("✅ Timeline service initialized")
293
294 // Initialize discover service (public feed from all communities)
295 discoverRepo := postgresRepo.NewDiscoverRepository(db, cursorSecret)
296 discoverService := discover.NewDiscoverService(discoverRepo)
297 log.Println("✅ Discover service initialized")
298
299 // Start Jetstream consumer for posts
300 // This consumer indexes posts created in community repositories via the firehose
301 // Currently handles only CREATE operations - UPDATE/DELETE deferred until those features exist
302 postJetstreamURL := os.Getenv("POST_JETSTREAM_URL")
303 if postJetstreamURL == "" {
304 // Listen to post record creation events
305 postJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.post.record"
306 }
307
308 postEventConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService)
309 postJetstreamConnector := jetstream.NewPostJetstreamConnector(postEventConsumer, postJetstreamURL)
310
311 go func() {
312 if startErr := postJetstreamConnector.Start(ctx); startErr != nil {
313 log.Printf("Post Jetstream consumer stopped: %v", startErr)
314 }
315 }()
316
317 log.Printf("Started Jetstream post consumer: %s", postJetstreamURL)
318 log.Println(" - Indexing: social.coves.post.record CREATE operations")
319 log.Println(" - UPDATE/DELETE indexing deferred until those features are implemented")
320
321 // Start Jetstream consumer for aggregators
322 // This consumer indexes aggregator service declarations and authorization records
323 // Following Bluesky's pattern for feed generators and labelers
324 // NOTE: Uses the same Jetstream as communities, just filtering different collections
325 aggregatorJetstreamURL := communityJetstreamURL
326 // Override if specific URL needed for testing
327 if envURL := os.Getenv("AGGREGATOR_JETSTREAM_URL"); envURL != "" {
328 aggregatorJetstreamURL = envURL
329 } else if aggregatorJetstreamURL == "" {
330 // Fallback if community URL also not set
331 aggregatorJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.aggregator.service&wantedCollections=social.coves.aggregator.authorization"
332 }
333
334 aggregatorEventConsumer := jetstream.NewAggregatorEventConsumer(aggregatorRepo)
335 aggregatorJetstreamConnector := jetstream.NewAggregatorJetstreamConnector(aggregatorEventConsumer, aggregatorJetstreamURL)
336
337 go func() {
338 if startErr := aggregatorJetstreamConnector.Start(ctx); startErr != nil {
339 log.Printf("Aggregator Jetstream consumer stopped: %v", startErr)
340 }
341 }()
342
343 log.Printf("Started Jetstream aggregator consumer: %s", aggregatorJetstreamURL)
344 log.Println(" - Indexing: social.coves.aggregator.service (service declarations)")
345 log.Println(" - Indexing: social.coves.aggregator.authorization (authorization records)")
346
347 // Register XRPC routes
348 routes.RegisterUserRoutes(r, userService)
349 routes.RegisterCommunityRoutes(r, communityService, authMiddleware)
350 log.Println("Community XRPC endpoints registered with OAuth authentication")
351
352 routes.RegisterPostRoutes(r, postService, authMiddleware)
353 log.Println("Post XRPC endpoints registered with OAuth authentication")
354
355 routes.RegisterCommunityFeedRoutes(r, feedService)
356 log.Println("Feed XRPC endpoints registered (public, no auth required)")
357
358 routes.RegisterTimelineRoutes(r, timelineService, authMiddleware)
359 log.Println("Timeline XRPC endpoints registered (requires authentication)")
360
361 routes.RegisterDiscoverRoutes(r, discoverService)
362 log.Println("Discover XRPC endpoints registered (public, no auth required)")
363
364 routes.RegisterAggregatorRoutes(r, aggregatorService)
365 log.Println("Aggregator XRPC endpoints registered (query endpoints public)")
366
367 r.Get("/health", func(w http.ResponseWriter, r *http.Request) {
368 w.WriteHeader(http.StatusOK)
369 if _, err := w.Write([]byte("OK")); err != nil {
370 log.Printf("Failed to write health check response: %v", err)
371 }
372 })
373
374 port := os.Getenv("APPVIEW_PORT")
375 if port == "" {
376 port = "8081" // Match .env.dev default
377 }
378
379 fmt.Printf("Coves AppView starting on port %s\n", port)
380 fmt.Printf("Default PDS: %s\n", defaultPDS)
381 log.Fatal(http.ListenAndServe(":"+port, r))
382}
383
384// authenticateWithPDS creates a session on the PDS and returns an access token
385func authenticateWithPDS(pdsURL, handle, password string) (string, error) {
386 type CreateSessionRequest struct {
387 Identifier string `json:"identifier"`
388 Password string `json:"password"`
389 }
390
391 type CreateSessionResponse struct {
392 DID string `json:"did"`
393 Handle string `json:"handle"`
394 AccessJwt string `json:"accessJwt"`
395 }
396
397 reqBody, err := json.Marshal(CreateSessionRequest{
398 Identifier: handle,
399 Password: password,
400 })
401 if err != nil {
402 return "", fmt.Errorf("failed to marshal request: %w", err)
403 }
404
405 resp, err := http.Post(
406 pdsURL+"/xrpc/com.atproto.server.createSession",
407 "application/json",
408 bytes.NewReader(reqBody),
409 )
410 if err != nil {
411 return "", fmt.Errorf("failed to call PDS: %w", err)
412 }
413 defer func() {
414 if closeErr := resp.Body.Close(); closeErr != nil {
415 log.Printf("Failed to close response body: %v", closeErr)
416 }
417 }()
418
419 if resp.StatusCode != http.StatusOK {
420 body, readErr := io.ReadAll(resp.Body)
421 if readErr != nil {
422 return "", fmt.Errorf("PDS returned status %d and failed to read body: %w", resp.StatusCode, readErr)
423 }
424 return "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
425 }
426
427 var session CreateSessionResponse
428 if err := json.NewDecoder(resp.Body).Decode(&session); err != nil {
429 return "", fmt.Errorf("failed to decode response: %w", err)
430 }
431
432 return session.AccessJwt, nil
433}