A community based topic aggregation platform built on atproto
bretton.dev
1package main
2
3import (
4 "bytes"
5 "context"
6 "database/sql"
7 "encoding/json"
8 "fmt"
9 "io"
10 "log"
11 "net/http"
12 "os"
13 "strings"
14 "time"
15
16 "github.com/go-chi/chi/v5"
17 chiMiddleware "github.com/go-chi/chi/v5/middleware"
18 _ "github.com/lib/pq"
19 "github.com/pressly/goose/v3"
20
21 "Coves/internal/api/handlers/oauth"
22 "Coves/internal/api/middleware"
23 "Coves/internal/api/routes"
24 "Coves/internal/atproto/did"
25 "Coves/internal/atproto/identity"
26 "Coves/internal/atproto/jetstream"
27 "Coves/internal/core/communities"
28 oauthCore "Coves/internal/core/oauth"
29 "Coves/internal/core/users"
30 postgresRepo "Coves/internal/db/postgres"
31)
32
33func main() {
34 // Database configuration (AppView database)
35 dbURL := os.Getenv("DATABASE_URL")
36 if dbURL == "" {
37 // Use dev database from .env.dev
38 dbURL = "postgres://dev_user:dev_password@localhost:5433/coves_dev?sslmode=disable"
39 }
40
41 // Default PDS URL for this Coves instance (supports self-hosting)
42 defaultPDS := os.Getenv("PDS_URL")
43 if defaultPDS == "" {
44 defaultPDS = "http://localhost:3001" // Local dev PDS
45 }
46
47 db, err := sql.Open("postgres", dbURL)
48 if err != nil {
49 log.Fatal("Failed to connect to database:", err)
50 }
51 defer db.Close()
52
53 if err := db.Ping(); err != nil {
54 log.Fatal("Failed to ping database:", err)
55 }
56
57 log.Println("Connected to AppView database")
58
59 // Run migrations
60 if err := goose.SetDialect("postgres"); err != nil {
61 log.Fatal("Failed to set goose dialect:", err)
62 }
63
64 if err := goose.Up(db, "internal/db/migrations"); err != nil {
65 log.Fatal("Failed to run migrations:", err)
66 }
67
68 log.Println("Migrations completed successfully")
69
70 r := chi.NewRouter()
71
72 r.Use(chiMiddleware.Logger)
73 r.Use(chiMiddleware.Recoverer)
74 r.Use(chiMiddleware.RequestID)
75
76 // Rate limiting: 100 requests per minute per IP
77 rateLimiter := middleware.NewRateLimiter(100, 1*time.Minute)
78 r.Use(rateLimiter.Middleware)
79
80 // Initialize identity resolver
81 identityConfig := identity.DefaultConfig()
82 // Override from environment if set
83 if plcURL := os.Getenv("IDENTITY_PLC_URL"); plcURL != "" {
84 identityConfig.PLCURL = plcURL
85 }
86 if cacheTTL := os.Getenv("IDENTITY_CACHE_TTL"); cacheTTL != "" {
87 if duration, err := time.ParseDuration(cacheTTL); err == nil {
88 identityConfig.CacheTTL = duration
89 }
90 }
91
92 identityResolver := identity.NewResolver(db, identityConfig)
93 log.Println("Identity resolver initialized with PLC:", identityConfig.PLCURL)
94
95 // Initialize OAuth session store
96 sessionStore := oauthCore.NewPostgresSessionStore(db)
97 log.Println("OAuth session store initialized")
98
99 // Initialize repositories and services
100 userRepo := postgresRepo.NewUserRepository(db)
101 userService := users.NewUserService(userRepo, identityResolver, defaultPDS)
102
103 communityRepo := postgresRepo.NewCommunityRepository(db)
104
105 // Initialize DID generator for communities
106 // IS_DEV_ENV=true: Generate did:plc:xxx without registering to PLC directory
107 // IS_DEV_ENV=false: Generate did:plc:xxx and register with PLC_DIRECTORY_URL
108 isDevEnv := os.Getenv("IS_DEV_ENV") == "true"
109 plcDirectoryURL := os.Getenv("PLC_DIRECTORY_URL")
110 if plcDirectoryURL == "" {
111 plcDirectoryURL = "https://plc.directory" // Default to Bluesky's PLC
112 }
113 didGenerator := did.NewGenerator(isDevEnv, plcDirectoryURL)
114 log.Printf("DID generator initialized (dev_mode=%v, plc_url=%s)", isDevEnv, plcDirectoryURL)
115
116 instanceDID := os.Getenv("INSTANCE_DID")
117 if instanceDID == "" {
118 instanceDID = "did:web:coves.local" // Default for development
119 }
120
121 // V2: Extract instance domain for community handles
122 // IMPORTANT: This MUST match the domain in INSTANCE_DID for security
123 // We cannot allow arbitrary domains to prevent impersonation attacks
124 // Example attack: !leagueoflegends@riotgames.com on a non-Riot instance
125 var instanceDomain string
126 if strings.HasPrefix(instanceDID, "did:web:") {
127 // Extract domain from did:web (this is the authoritative source)
128 instanceDomain = strings.TrimPrefix(instanceDID, "did:web:")
129 } else {
130 // For non-web DIDs (e.g., did:plc), require explicit INSTANCE_DOMAIN
131 instanceDomain = os.Getenv("INSTANCE_DOMAIN")
132 if instanceDomain == "" {
133 log.Fatal("INSTANCE_DOMAIN must be set for non-web DIDs")
134 }
135 }
136
137 log.Printf("Instance domain: %s (extracted from DID: %s)", instanceDomain, instanceDID)
138
139 // V2: Initialize PDS account provisioner for communities
140 provisioner := communities.NewPDSAccountProvisioner(userService, instanceDomain, defaultPDS)
141
142 communityService := communities.NewCommunityService(communityRepo, didGenerator, defaultPDS, instanceDID, instanceDomain, provisioner)
143
144 // Authenticate Coves instance with PDS to enable community record writes
145 // The instance needs a PDS account to write community records it owns
146 pdsHandle := os.Getenv("PDS_INSTANCE_HANDLE")
147 pdsPassword := os.Getenv("PDS_INSTANCE_PASSWORD")
148 if pdsHandle != "" && pdsPassword != "" {
149 log.Printf("Authenticating Coves instance (%s) with PDS...", instanceDID)
150 accessToken, err := authenticateWithPDS(defaultPDS, pdsHandle, pdsPassword)
151 if err != nil {
152 log.Printf("Warning: Failed to authenticate with PDS: %v", err)
153 log.Println("Community creation will fail until PDS authentication is configured")
154 } else {
155 if svc, ok := communityService.(interface{ SetPDSAccessToken(string) }); ok {
156 svc.SetPDSAccessToken(accessToken)
157 log.Println("✓ Coves instance authenticated with PDS")
158 }
159 }
160 } else {
161 log.Println("Note: PDS_INSTANCE_HANDLE and PDS_INSTANCE_PASSWORD not set")
162 log.Println("Community creation via write-forward is disabled")
163 }
164
165 // Start Jetstream consumer for read-forward user indexing
166 jetstreamURL := os.Getenv("JETSTREAM_URL")
167 if jetstreamURL == "" {
168 jetstreamURL = "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.actor.profile"
169 }
170
171 pdsFilter := os.Getenv("JETSTREAM_PDS_FILTER") // Optional: filter to specific PDS
172
173 userConsumer := jetstream.NewUserEventConsumer(userService, identityResolver, jetstreamURL, pdsFilter)
174 ctx := context.Background()
175 go func() {
176 if err := userConsumer.Start(ctx); err != nil {
177 log.Printf("Jetstream consumer stopped: %v", err)
178 }
179 }()
180
181 log.Printf("Started Jetstream user consumer: %s", jetstreamURL)
182
183 // Note: Community indexing happens through the same Jetstream firehose
184 // The CommunityEventConsumer is used by handlers when processing community-related events
185 // For now, community records are created via write-forward to PDS, then indexed when
186 // they appear in the firehose. A dedicated consumer can be added later if needed.
187 log.Println("Community event consumer initialized (processes events from firehose)")
188
189 // Start OAuth cleanup background job
190 go func() {
191 ticker := time.NewTicker(1 * time.Hour)
192 defer ticker.Stop()
193 for range ticker.C {
194 if pgStore, ok := sessionStore.(*oauthCore.PostgresSessionStore); ok {
195 _ = pgStore.CleanupExpiredRequests(ctx)
196 _ = pgStore.CleanupExpiredSessions(ctx)
197 log.Println("OAuth cleanup completed")
198 }
199 }
200 }()
201
202 log.Println("Started OAuth cleanup background job (runs hourly)")
203
204 // Initialize OAuth cookie store (singleton)
205 cookieSecret, err := oauth.GetEnvBase64OrPlain("OAUTH_COOKIE_SECRET")
206 if err != nil {
207 log.Fatalf("Failed to load OAUTH_COOKIE_SECRET: %v", err)
208 }
209 if cookieSecret == "" {
210 log.Fatal("OAUTH_COOKIE_SECRET not configured")
211 }
212
213 if err := oauth.InitCookieStore(cookieSecret); err != nil {
214 log.Fatalf("Failed to initialize cookie store: %v", err)
215 }
216
217 // Initialize OAuth handlers
218 loginHandler := oauth.NewLoginHandler(identityResolver, sessionStore)
219 callbackHandler := oauth.NewCallbackHandler(sessionStore)
220 logoutHandler := oauth.NewLogoutHandler(sessionStore)
221
222 // OAuth routes (public endpoints)
223 r.Post("/oauth/login", loginHandler.HandleLogin)
224 r.Get("/oauth/callback", callbackHandler.HandleCallback)
225 r.Post("/oauth/logout", logoutHandler.HandleLogout)
226 r.Get("/oauth/client-metadata.json", oauth.HandleClientMetadata)
227 r.Get("/oauth/jwks.json", oauth.HandleJWKS)
228
229 log.Println("OAuth endpoints registered")
230
231 // Register XRPC routes
232 routes.RegisterUserRoutes(r, userService)
233 routes.RegisterCommunityRoutes(r, communityService)
234 log.Println("Community XRPC endpoints registered")
235
236 r.Get("/health", func(w http.ResponseWriter, r *http.Request) {
237 w.WriteHeader(http.StatusOK)
238 w.Write([]byte("OK"))
239 })
240
241 port := os.Getenv("APPVIEW_PORT")
242 if port == "" {
243 port = "8081" // Match .env.dev default
244 }
245
246 fmt.Printf("Coves AppView starting on port %s\n", port)
247 fmt.Printf("Default PDS: %s\n", defaultPDS)
248 log.Fatal(http.ListenAndServe(":"+port, r))
249}
250
251// authenticateWithPDS creates a session on the PDS and returns an access token
252func authenticateWithPDS(pdsURL, handle, password string) (string, error) {
253 type CreateSessionRequest struct {
254 Identifier string `json:"identifier"`
255 Password string `json:"password"`
256 }
257
258 type CreateSessionResponse struct {
259 DID string `json:"did"`
260 Handle string `json:"handle"`
261 AccessJwt string `json:"accessJwt"`
262 }
263
264 reqBody, err := json.Marshal(CreateSessionRequest{
265 Identifier: handle,
266 Password: password,
267 })
268 if err != nil {
269 return "", fmt.Errorf("failed to marshal request: %w", err)
270 }
271
272 resp, err := http.Post(
273 pdsURL+"/xrpc/com.atproto.server.createSession",
274 "application/json",
275 bytes.NewReader(reqBody),
276 )
277 if err != nil {
278 return "", fmt.Errorf("failed to call PDS: %w", err)
279 }
280 defer resp.Body.Close()
281
282 if resp.StatusCode != http.StatusOK {
283 body, _ := io.ReadAll(resp.Body)
284 return "", fmt.Errorf("PDS returned status %d: %s", resp.StatusCode, string(body))
285 }
286
287 var session CreateSessionResponse
288 if err := json.NewDecoder(resp.Body).Decode(&session); err != nil {
289 return "", fmt.Errorf("failed to decode response: %w", err)
290 }
291
292 return session.AccessJwt, nil
293}