maarcadetweet/crates/appview/src/main.rs

use anyhow::Result;
use at_identity::DidHandleResolver;
use at_shared::config::AppConfig;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::info;
use tracing_subscriber::EnvFilter;

mod firehose;
mod handle_sync;
mod indexer;
mod ingest;
mod routes;
mod state;

use state::AppState;

#[tokio::main]
async fn main() -> Result<()> {
    // Load `.env` from the working directory (and upwards) if present.
    // Nothing else in the process reads it, so without this
    // `cp .env.example .env && cargo run` fails with `missing env:
    // PDS_HOST`. Real environment variables always win over the file.
    let _ = dotenvy::dotenv();

    // Install a rustls crypto provider before any TLS connection. `ring`
    // is the only one we currently support; using `aws_lc_rs` would
    // require a non-default feature on rustls.
    let _ = rustls::crypto::ring::default_provider().install_default();

    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
        .init();

    let cfg = AppConfig::from_env()?;
    let db = sqlx::PgPool::connect(&cfg.database_url_appview).await?;
    sqlx::migrate!("../../migrations/appview").run(&db).await?;

    // Read the last persisted cursor so we resume after restart instead of
    // missing events that landed in the gap between (a) the last value we
    // wrote and (b) Jetstream's default backfill window.
    let start_cursor = indexer::cursor_get(&db).await.unwrap_or(0);
    if start_cursor > 0 {
        info!(cursor = start_cursor, "resuming Jetstream from last persisted cursor");
    }

    // Bounded channel for "cursor wants to advance" signals. We push one
    // tick per ~100 events from the consumer thread; the flush task drains
    // the channel and writes a single batched UPDATE.
    let (cursor_tx, cursor_rx) = mpsc::channel::<i64>(32);

    let stats = firehose::Stats::new();

    // Spawn the cursor-flush task. It lives for the whole process — when
    // the receiver end drops (which only happens at shutdown), the task
    // does a final flush and exits.
    let _cursor_task = firehose::spawn_cursor_flush(db.clone(), cursor_rx, stats.clone());

    {
        let mut jetstream = at_firehose::JetstreamConsumer::new(
            cfg.jetstream_url.clone(),
            cfg.jetstream_collections.clone(),
        )
        .with_connected_flag(stats.jetstream_connected_arc())
        .with_max_backoff_secs(30);
        if start_cursor > 0 {
            jetstream = jetstream.with_cursor(start_cursor);
        }
        let handler = firehose::IndexHandler::new(db.clone(), stats.clone(), cursor_tx.clone());
        tokio::spawn(async move {
            if let Err(e) = jetstream
                .run(move |ev| {
                    let h = handler.clone();
                    async move { h.handle(ev).await }
                })
                .await
            {
                tracing::error!("jetstream terminated: {e:#}");
            }
        });
    }

    let state = AppState::new(cfg.clone(), db.clone(), stats.clone());

    // Back-fill the `handle` column on posts that the Jetstream
    // indexer inserted with an empty placeholder. The worker dispatches
    // by DID method: `did:plc:` → PLC directory, `did:web:` → a
    // WebResolver that fetches the host's `.well-known/did.json`.
    // Anything else (e.g. `did:key:`) is silently skipped.
    let plc: Arc<dyn DidHandleResolver> = Arc::new(at_identity::PlcClient::new(
        cfg.plc_directory_url.clone(),
    ));
    let web: Arc<dyn DidHandleResolver> = Arc::new(at_identity::WebResolver::new());
    // PDS-local handle resolver. The handle_sync worker consults it
    // first, before the public PLC/web resolvers, so `did:key:`
    // users (and any other DID the operator hosts on this PDS) get their
    // local handle without a round trip to plc.directory. A 404 from
    // the PDS falls through to the public resolvers.
    //
    // Use the cluster-internal URL when configured (e.g. `http://pds:3000`
    // inside docker compose) — `pds_public_url` may not be reachable
    // from inside the cluster when TLS / DNS is set up for outside
    // clients only.
    let pds_base_url = cfg
        .pds_internal_url
        .clone()
        .unwrap_or_else(|| cfg.pds_public_url.clone());
    let pds_resolver: Arc<dyn at_identity::DidHandleResolver> = Arc::new(
        at_identity::pds_handle::PdsHandleResolver::new(pds_base_url),
    );

    let handle_sync = handle_sync::HandleSyncWorker {
        db: db.clone(),
        pds_resolver,
        plc_resolver: plc,
        web_resolver: web,
        interval_secs: cfg.appview_handle_sync_interval_secs,
    };
    tokio::spawn(async move {
        handle_sync.run_forever().await;
    });

    let app = routes::router(state);
    let addr: SocketAddr = format!("{}:{}", cfg.appview_host, cfg.appview_port).parse()?;
    info!("appview listening on http://{addr}");
    let listener = tokio::net::TcpListener::bind(addr).await?;
    axum::serve(listener, app).await?;
    Ok(())
}