Something went wrong. Try again.
Tea journaling on ATProto (alpha)
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614package main
import ( "bufio" "context" "flag" "fmt" "log/slog" "net/http" "os" "os/signal" "path/filepath" "strconv" "strings" "syscall" "time"
"tangled.org/arabica.social/arabica/internal/atproto" "tangled.org/arabica.social/arabica/internal/backup" "tangled.org/arabica.social/arabica/internal/database/boltstore" "tangled.org/arabica.social/arabica/internal/database/sqlitestore" "tangled.org/arabica.social/arabica/internal/email" "tangled.org/arabica.social/arabica/internal/feed" "tangled.org/arabica.social/arabica/internal/firehose" "tangled.org/arabica.social/arabica/internal/handlers" "tangled.org/arabica.social/arabica/internal/logging" "tangled.org/arabica.social/arabica/internal/metrics" "tangled.org/arabica.social/arabica/internal/moderation" "tangled.org/arabica.social/arabica/internal/routing" "tangled.org/arabica.social/arabica/internal/tracing"
"github.com/prometheus/client_golang/prometheus/promhttp" "github.com/rs/zerolog" "github.com/rs/zerolog/log" "go.opentelemetry.io/otel/attribute")
func main() { // Parse command-line flags knownDIDsFile := flag.String("known-dids", "", "Path to file containing DIDs to backfill on startup (one per line)") flag.Parse()
// Configure zerolog // Set log level from environment (default: info) logLevel := os.Getenv("LOG_LEVEL") switch logLevel { case "debug": zerolog.SetGlobalLevel(zerolog.DebugLevel) case "info", "": zerolog.SetGlobalLevel(zerolog.InfoLevel) case "warn": zerolog.SetGlobalLevel(zerolog.WarnLevel) case "error": zerolog.SetGlobalLevel(zerolog.ErrorLevel) default: zerolog.SetGlobalLevel(zerolog.InfoLevel) }
// Use pretty console logging in development, JSON in production if os.Getenv("LOG_FORMAT") == "json" { // Production: JSON logs log.Logger = zerolog.New(os.Stdout).With().Timestamp().Logger() } else { // Development: pretty console logs log.Logger = log.Output(zerolog.ConsoleWriter{ Out: os.Stdout, TimeFormat: time.RFC3339, }) }
// Bridge slog to zerolog so third-party libraries (e.g. indigo OAuth) // produce output consistent with the rest of the application. slog.SetDefault(slog.New(logging.NewZerologHandler(log.Logger)))
log.Info().Msg("Starting Arabica Coffee Tracker")
app := newArabicaApp() log.Info(). Str("app", app.Name). Int("descriptors", len(app.Descriptors)). Msg("Constructed app config")
// Initialize OpenTelemetry tracing tp, err := tracing.Init(context.Background()) if err != nil { log.Warn().Err(err).Msg("Failed to initialize tracing, continuing without it") } else { defer func() { if err := tp.Shutdown(context.Background()); err != nil { log.Error().Err(err).Msg("Error shutting down tracer provider") } }() log.Info().Msg("OpenTelemetry tracing initialized") }
// Get port from env or use default port := os.Getenv("PORT") if port == "" { port = "18910" }
// Get public root URL for reverse proxy deployments // This allows the server to be accessed via a different URL than it's running on // e.g., SERVER_PUBLIC_URL=https://arabica.example.com when behind a reverse proxy publicURL := os.Getenv("SERVER_PUBLIC_URL")
// Initialize BoltDB store for persistent sessions and feed registry dbPath := os.Getenv("ARABICA_DB_PATH") if dbPath == "" { // Default to XDG data directory or home directory for development // This avoids issues when running from read-only locations (e.g., nix run) dataDir := os.Getenv("XDG_DATA_HOME") if dataDir == "" { home, err := os.UserHomeDir() if err != nil { log.Fatal().Err(err).Msg("Failed to get home directory") } dataDir = filepath.Join(home, ".local", "share") } dbPath = filepath.Join(dataDir, "arabica", "arabica.db") }
store, err := boltstore.Open(boltstore.Options{ Path: dbPath, }) if err != nil { log.Fatal().Err(err).Str("path", dbPath).Msg("Failed to open database") } defer store.Close()
log.Info().Str("path", dbPath).Msg("Database opened")
// Get specialized stores sessionStore := store.SessionStore()
// Initialize OAuth manager with persistent session store // For local development, localhost URLs trigger special localhost mode in indigo clientID := os.Getenv("OAUTH_CLIENT_ID") redirectURI := os.Getenv("OAUTH_REDIRECT_URI")
if clientID == "" && redirectURI == "" { // Use public URL if set, otherwise localhost defaults for development if publicURL != "" { redirectURI = publicURL + "/oauth/callback" clientID = publicURL + "/.well-known/oauth-client-metadata.json" log.Info(). Str("public_url", publicURL). Msg("Using public URL for OAuth (reverse proxy mode)") } else { redirectURI = fmt.Sprintf("http://127.0.0.1:%s/oauth/callback", port) clientID = "" // Empty triggers localhost mode log.Info().Msg("Using localhost OAuth mode (for development)") } }
oauthManager, err := atproto.NewOAuthManager(clientID, redirectURI, app.OAuthScopes(), sessionStore) if err != nil { log.Fatal().Err(err).Msg("Failed to initialize OAuth") }
// feedRegistry and feedService are initialised after feedIndex opens below
// Setup context for graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel()
// Handle shutdown signals sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
// Initialize firehose consumer // Determine feed index path feedIndexPath := os.Getenv("ARABICA_FEED_INDEX_PATH") if feedIndexPath == "" { dataDir := os.Getenv("XDG_DATA_HOME") if dataDir == "" { home, err := os.UserHomeDir() if err != nil { log.Fatal().Err(err).Msg("Failed to get home directory for feed index") } dataDir = filepath.Join(home, ".local", "share") } feedIndexPath = filepath.Join(dataDir, "arabica", "feed-index.sqlite") }
// Create firehose config firehoseConfig := firehose.DefaultConfig() firehoseConfig.IndexPath = feedIndexPath firehoseConfig.WantedCollections = app.NSIDs()
// Parse profile cache TTL from env if set if ttlStr := os.Getenv("ARABICA_PROFILE_CACHE_TTL"); ttlStr != "" { if ttl, err := time.ParseDuration(ttlStr); err == nil { firehoseConfig.ProfileCacheTTL = int64(ttl.Seconds()) } }
// Create feed index feedIndex, err := firehose.NewFeedIndex(feedIndexPath, time.Duration(firehoseConfig.ProfileCacheTTL)*time.Second) if err != nil { log.Fatal().Err(err).Str("path", feedIndexPath).Msg("Failed to create feed index") }
log.Info().Str("path", feedIndexPath).Msg("Feed index opened")
// One-time seed: copy DIDs from the legacy BoltDB feed_registry bucket into // SQLite registered_dids. INSERT OR IGNORE makes this a no-op after first run. if legacyDIDs := store.LegacyFeedDIDs(); len(legacyDIDs) > 0 { added, err := feedIndex.SeedRegisteredDIDs(legacyDIDs) if err != nil { log.Warn().Err(err).Msg("Failed to seed registered DIDs from legacy feed registry") } else if added > 0 { log.Info().Int("seeded", added).Msg("Seeded registered_dids from legacy BoltDB feed registry") } }
// Initialise feed registry backed by SQLite registered_dids. // This preserves the distinction between registered users (explicit logins) // and known_dids (any user whose records have been indexed via firehose). feedRegistry := feed.NewPersistentRegistry(feedIndex) feedService := feed.NewService(feedRegistry)
log.Info().Int("registered_users", feedRegistry.Count()).Msg("Feed service initialised")
// Create and start consumer firehoseConsumer := firehose.NewConsumer(firehoseConfig, feedIndex) firehoseConsumer.Start(ctx)
// Create and start profile watcher (separate Jetstream connection filtered // to app.bsky.actor.profile events for known Arabica users only) profileWatcher := firehose.NewProfileWatcher(firehoseConfig, feedIndex) profileWatcher.Start(ctx)
// Wire up the feed service to use the firehose index adapter := firehose.NewFeedIndexAdapter(feedIndex) feedService.SetFirehoseIndex(adapter)
// Wire up moderation filtering for the feed moderationStore := sqlitestore.NewModerationStore(feedIndex.DB()) feedService.SetModerationFilter(moderationStore)
log.Info().Msg("Firehose consumer started")
// Start metrics collector for periodic gauge updates metrics.StartCollector(ctx, metrics.StatsSource{ KnownDIDCount: feedIndex.KnownDIDCount, RegisteredCount: feedRegistry.Count, RecordCount: feedIndex.RecordCount, PendingJoinCount: func() int { joinStore := store.JoinStore() if reqs, err := joinStore.ListRequests(context.Background()); err == nil { return len(reqs) } return 0 }, LikeCount: feedIndex.TotalLikeCount, CommentCount: feedIndex.TotalCommentCount, RecordCountByCollection: feedIndex.RecordCountByCollection, FirehoseConnected: firehoseConsumer.IsConnected, BoltStats: store.Stats, SQLiteStats: feedIndex.DB().Stats, }, 60*time.Second)
// Log known DIDs from database (DIDs discovered via firehose) if knownDIDsFromDB, err := feedIndex.GetKnownDIDs(context.Background()); err == nil { if len(knownDIDsFromDB) > 0 { log.Info(). Int("count", len(knownDIDsFromDB)). Strs("dids", knownDIDsFromDB). Msg("Known DIDs from firehose index") } else { log.Info().Msg("No known DIDs in firehose index yet (will populate as events arrive)") } } else { log.Warn().Err(err).Msg("Failed to retrieve known DIDs from firehose index") }
// Backfill registered users and known DIDs in background go func() { time.Sleep(5 * time.Second) // Wait for initial connection
// Create a root span so all backfill work is grouped under one trace backfillCtx, backfillSpan := tracing.HandlerSpan(ctx, "backfill.startup") defer backfillSpan.End()
// Collect all DIDs to backfill collectCtx, collectSpan := tracing.HandlerSpan(backfillCtx, "backfill.collect_dids") didsToBackfill := make(map[string]struct{})
// Add registered users for _, did := range feedRegistry.List() { didsToBackfill[did] = struct{}{} }
// Add DIDs from known-dids file if provided if *knownDIDsFile != "" { knownDIDs, err := loadKnownDIDs(*knownDIDsFile) if err != nil { log.Warn().Err(err).Str("file", *knownDIDsFile).Msg("Failed to load known DIDs file") } else { for _, did := range knownDIDs { didsToBackfill[did] = struct{}{} } log.Info(). Int("count", len(knownDIDs)). Str("file", *knownDIDsFile). Strs("dids", knownDIDs). Msg("Loaded known DIDs from file") } } collectSpan.SetAttributes(attribute.Int("backfill.total_dids", len(didsToBackfill))) collectSpan.End()
// Pre-load already-backfilled DIDs to avoid N+1 SELECT per DID filterCtx, filterSpan := tracing.HandlerSpan(collectCtx, "backfill.filter_backfilled") alreadyBackfilled, err := firehoseConsumer.BackfilledDIDs(filterCtx) if err != nil { log.Warn().Err(err).Msg("Failed to load backfilled DIDs, will check individually") alreadyBackfilled = make(map[string]struct{}) }
// Remove already-backfilled DIDs for did := range alreadyBackfilled { delete(didsToBackfill, did) } filterSpan.SetAttributes( attribute.Int("backfill.already_backfilled", len(alreadyBackfilled)), attribute.Int("backfill.remaining", len(didsToBackfill)), ) filterSpan.End()
// Backfill remaining DIDs _, execSpan := tracing.HandlerSpan(filterCtx, "backfill.execute", attribute.Int("backfill.count", len(didsToBackfill)), ) successCount := 0 for did := range didsToBackfill { if err := firehoseConsumer.BackfillDID(backfillCtx, did); err != nil { log.Warn().Err(err).Str("did", did).Msg("Failed to backfill user") } else { successCount++ } } execSpan.SetAttributes( attribute.Int("backfill.success", successCount), attribute.Int("backfill.failed", len(didsToBackfill)-successCount), ) execSpan.End()
log.Info(). Int("skipped", len(alreadyBackfilled)). Int("backfilled", successCount). Int("failed", len(didsToBackfill)-successCount). Msg("Backfill complete") }()
// Register users in the feed when they authenticate // This ensures users are added to the feed even if they had an existing session oauthManager.SetOnAuthSuccess(func(did string) { feedRegistry.Register(did) profileWatcher.Watch(did) // Backfill the user's records (BackfillUser creates its own span // only when there is actual work to do, avoiding empty traces for // already-backfilled users) go func() { if err := firehoseConsumer.BackfillDID(context.Background(), did); err != nil { log.Warn().Err(err).Str("did", did).Msg("Failed to backfill new user") } }() })
if clientID == "" { log.Info(). Str("mode", "localhost development"). Str("redirect_uri", redirectURI). Msg("OAuth configured") } else { log.Info(). Str("mode", "public"). Str("client_id", clientID). Str("redirect_uri", redirectURI). Msg("OAuth configured") }
// Initialize atproto client atprotoClient := atproto.NewClient(oauthManager) log.Info().Msg("ATProto client initialized")
// Initialize session cache for in-memory caching of user data sessionCache := atproto.NewSessionCache() stopCacheCleanup := sessionCache.StartCleanupRoutine(10 * time.Minute) defer stopCacheCleanup() log.Info().Msg("Session cache initialized with background cleanup")
// Determine if we should use secure cookies (default: false for development) // Set SECURE_COOKIES=true in production with HTTPS secureCookies := os.Getenv("SECURE_COOKIES") == "true"
// Initialize handlers with all dependencies via constructor injection h := handlers.NewHandler( oauthManager, atprotoClient, sessionCache, feedService, feedRegistry, handlers.Config{ SecureCookies: secureCookies, PublicURL: publicURL, }, )
// Wire up the feed index for like functionality h.SetFeedIndex(feedIndex)
// Wire up the witness cache so reads are served from the firehose index // before falling back to the PDS h.SetWitnessCache(feedIndex)
// Wire the per-app brand into the handler so layout/header/footer // templates render with the right display name and tagline. h.SetBrand(app.Brand)
// Initialize moderation service and wire up to handler moderatorsConfigPath := os.Getenv("ARABICA_MODERATORS_CONFIG") moderationSvc, err := moderation.NewService(moderatorsConfigPath) if err != nil { log.Warn().Err(err).Msg("Failed to initialize moderation service, moderation disabled") } else { h.SetModeration(moderationSvc, moderationStore) }
// Periodic cleanup of expired moderation labels go func() { ticker := time.NewTicker(1 * time.Hour) defer ticker.Stop() for { select { case <-ticker.C: if n, err := moderationStore.CleanExpiredLabels(ctx); err != nil { log.Error().Err(err).Msg("Failed to clean expired labels") } else if n > 0 { log.Info().Int("count", n).Msg("Cleaned expired moderation labels") } case <-ctx.Done(): return } } }()
// Initialize automated backups backupDir := os.Getenv("ARABICA_BACKUP_DIR") if backupDir == "" { dataDir := os.Getenv("XDG_DATA_HOME") if dataDir == "" { home, _ := os.UserHomeDir() dataDir = filepath.Join(home, ".local", "share") } backupDir = filepath.Join(dataDir, "arabica", "backups") } backupDest, err := backup.NewLocalDestination(backupDir) if err != nil { log.Warn().Err(err).Msg("Failed to create backup destination, backups disabled") } else { backupSvc := backup.NewService(backup.Config{ ScheduleHour: 11, // 11:00 UTC = 3:00 AM PST Retain: 2, Dest: backupDest, }) backupSvc.AddSource(backup.NewSQLiteSource("feed-index", feedIndex.DB())) backupSvc.Start(ctx) log.Info().Str("dir", backupDir).Msg("Automated backups enabled") }
// Initialize join request handling smtpPort := 587 if portStr := os.Getenv("SMTP_PORT"); portStr != "" { if p, err := strconv.Atoi(portStr); err == nil { smtpPort = p } } emailSender := email.NewSender(email.Config{ Host: os.Getenv("SMTP_HOST"), Port: smtpPort, User: os.Getenv("SMTP_USER"), Pass: os.Getenv("SMTP_PASS"), From: os.Getenv("SMTP_FROM"), AdminEmail: os.Getenv("ADMIN_EMAIL"), }) joinStore := store.JoinStore() pdsAdminURL := os.Getenv("PDS_ADMIN_URL") pdsAdminToken := os.Getenv("PDS_ADMIN_PASSWORD") h.SetJoin(emailSender, joinStore, pdsAdminURL, pdsAdminToken) if emailSender.Enabled() { log.Info().Str("host", os.Getenv("SMTP_HOST")).Msg("Email notifications enabled for join requests") } else { log.Info().Msg("Email notifications disabled (SMTP_HOST not set), join requests will be saved to database only") }
// Setup router with middleware handler := routing.SetupRouter(routing.Config{ App: app, Handlers: h, OAuthManager: oauthManager, Logger: log.Logger, ModerationService: moderationSvc, FirehoseConsumer: firehoseConsumer, })
// Start internal metrics server on localhost only (not publicly accessible) metricsPort := os.Getenv("METRICS_PORT") if metricsPort == "" { metricsPort = "9101" } metricsMux := http.NewServeMux() metricsMux.Handle("GET /metrics", promhttp.Handler()) metricsServer := &http.Server{ Addr: "127.0.0.1:" + metricsPort, Handler: metricsMux, } go func() { log.Info(). Str("address", "127.0.0.1:"+metricsPort). Msg("Starting metrics server (localhost only)") if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Error().Err(err).Msg("Metrics server failed to start") } }()
// Create HTTP server server := &http.Server{ Addr: "0.0.0.0:" + port, Handler: handler, }
// Start HTTP server in goroutine go func() { log.Info(). Str("address", "0.0.0.0:"+port). Str("url", "http://localhost:"+port). Bool("secure_cookies", secureCookies). Str("database", dbPath). Msg("Starting HTTP server")
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatal().Err(err).Msg("Server failed to start") } }()
// Wait for shutdown signal <-sigCh log.Info().Msg("Shutdown signal received")
// Stop firehose consumer first log.Info().Msg("Stopping firehose consumer...") firehoseConsumer.Stop()
// Graceful shutdown of HTTP server and metrics server shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) defer shutdownCancel()
if err := metricsServer.Shutdown(shutdownCtx); err != nil { log.Error().Err(err).Msg("Metrics server shutdown error") }
if err := server.Shutdown(shutdownCtx); err != nil { log.Error().Err(err).Msg("HTTP server shutdown error") }
log.Info().Msg("Server stopped")}
// loadKnownDIDs reads a file containing DIDs (one per line) and returns them as a slice.// Lines starting with # are treated as comments and ignored.// Empty lines and whitespace are trimmed.func loadKnownDIDs(filePath string) ([]string, error) { file, err := os.Open(filePath) if err != nil { return nil, fmt.Errorf("failed to open file: %w", err) } defer file.Close()
var dids []string scanner := bufio.NewScanner(file) lineNum := 0
for scanner.Scan() { lineNum++ line := strings.TrimSpace(scanner.Text())
// Skip empty lines and comments if line == "" || strings.HasPrefix(line, "#") { continue }
// Basic DID validation (must start with "did:") if !strings.HasPrefix(line, "did:") { log.Warn(). Str("file", filePath). Int("line", lineNum). Str("content", line). Msg("Skipping invalid DID (must start with 'did:')") continue }
dids = append(dids, line) }
if err := scanner.Err(); err != nil { return nil, fmt.Errorf("error reading file: %w", err) }
return dids, nil}