diff --git a/Cargo.toml b/Cargo.toml index ff2ff46..3adc854 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,4 +26,4 @@ url = "2" [dependencies.sqlx] version = "0.8" -features = ["postgres", "runtime-tokio-rustls", "json", "chrono"] +features = ["postgres", "runtime-tokio-rustls", "json", "chrono", "migrate"] diff --git a/Settings.toml.example b/Settings.toml.example index 994d6b9..fb464d1 100644 --- a/Settings.toml.example +++ b/Settings.toml.example @@ -2,7 +2,7 @@ address = "0.0.0.0" port = 8080 [database] -url = "postgresql://at_comments:at_comments@localhost:54329/at_comments" +url = "postgres://at_comments:at_comments@localhost:54329/at_comments" [poller] handle = "jay.bsky.team" emoji = "📝" diff --git a/migrations/20240101000000_create_posts.sql b/migrations/20240101000000_create_posts.sql new file mode 100644 index 0000000..4c4bc22 --- /dev/null +++ b/migrations/20240101000000_create_posts.sql @@ -0,0 +1,6 @@ +CREATE TABLE IF NOT EXISTS posts ( + id SERIAL PRIMARY KEY, + slug TEXT NOT NULL UNIQUE, + rkey TEXT NOT NULL, + time_us BIGINT NOT NULL +); diff --git a/migrations/20250101000000_change_time_us_to_bigint.sql b/migrations/20250101000000_change_time_us_to_bigint.sql new file mode 100644 index 0000000..6c01c2b --- /dev/null +++ b/migrations/20250101000000_change_time_us_to_bigint.sql @@ -0,0 +1,4 @@ +-- Change time_us from TEXT to BIGINT. +-- The existing values are stored as microsecond epoch strings and cast cleanly. +ALTER TABLE posts + ALTER COLUMN time_us TYPE BIGINT USING time_us::BIGINT; diff --git a/mise.toml b/mise.toml index c55ba14..c6c0f00 100644 --- a/mise.toml +++ b/mise.toml @@ -1,5 +1,6 @@ [tools] rust = "1.94.0" +"cargo:sqlx-cli" = "0.8" [env] _.file = ".env" @@ -45,21 +46,14 @@ exit 1 ''' [tasks.db-init] -description = "Create posts table (no migrations)" +description = "Apply database migrations via sqlx-cli" depends = ["db-up"] run = ''' set -euo pipefail - -docker exec -i "$POSTGRES_CONTAINER" psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" <<'SQL' -CREATE TABLE IF NOT EXISTS posts ( - id SERIAL PRIMARY KEY, - slug TEXT UNIQUE NOT NULL, - rkey TEXT NOT NULL, - time_us TEXT NOT NULL -); -SQL - -echo "Schema ready" +# sqlx-cli only accepts postgres://, not postgresql:// +DB_URL=$(echo "$ATC_DATABASE_URL" | sed 's|^postgresql://|postgres://|') +sqlx migrate run --database-url "$DB_URL" +echo "Migrations applied" ''' [tasks.db-seed] @@ -70,9 +64,9 @@ set -euo pipefail docker exec -i "$POSTGRES_CONTAINER" psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" <<'SQL' INSERT INTO posts (slug, rkey, time_us) VALUES - ('hello-world', '3lch7i6x4x22m', '1738411000000000'), - ('rust-and-rockets', '3lch7i6x4x22n', '1738414600000000'), - ('using-bluesky-comments', '3lch7i6x4x22o', '1738418200000000') + ('hello-world', '3lch7i6x4x22m', 1738411000000000), + ('rust-and-rockets', '3lch7i6x4x22n', 1738414600000000), + ('using-bluesky-comments', '3lch7i6x4x22o', 1738418200000000) ON CONFLICT (slug) DO UPDATE SET rkey = EXCLUDED.rkey, time_us = EXCLUDED.time_us; @@ -87,7 +81,7 @@ depends = ["db-seed"] run = ''' echo "Local Postgres is ready" echo "Run the app with: mise run dev" -echo "Quick test: curl http://127.0.0.1:4321/slug/hello-world" +echo "Quick test: curl http://127.0.0.1:${ATC_APP_PORT}/slug/hello-world" ''' [tasks.db-down] diff --git a/src/main.rs b/src/main.rs index 8901ef3..3a4294a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -17,10 +17,13 @@ use serde_json::{json, Value}; use sqlx::postgres::PgPoolOptions; use sqlx::Row; use std::net::SocketAddr; +use std::time::Duration; #[derive(Clone)] struct AppState { pool: sqlx::PgPool, + client: reqwest::Client, + poller_config: settings::PollerConfig, } #[tokio::main] @@ -34,17 +37,30 @@ async fn main() -> Result<(), Box> { // Create database pool let pool = PgPoolOptions::new() .max_connections(5) - .connect(&config.get::("database.url")?) + .connect(&config.database.url) .await?; log::info!("Connected to database"); - let app_state = AppState { pool: pool.clone() }; + sqlx::migrate!().run(&pool).await?; + log::info!("Database migrations applied"); + + // Create shared HTTP client with a request timeout + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build()?; + + let app_state = AppState { + pool: pool.clone(), + client: client.clone(), + poller_config: config.poller.clone(), + }; // Spawn background RSS poller task let poller_pool = pool.clone(); + let poller_config = config.poller.clone(); tokio::spawn(async move { - rss_poller::rss_polling_task(poller_pool).await; + rss_poller::rss_polling_task(client, poller_pool, poller_config).await; }); // Build router @@ -56,9 +72,7 @@ async fn main() -> Result<(), Box> { .with_state(app_state); // Bind and serve - let address = config.get::("app.address")?; - let port = config.get::("app.port")?; - let addr = format!("{}:{}", address, port).parse::()?; + let addr = format!("{}:{}", config.app.address, config.app.port).parse::()?; let listener = tokio::net::TcpListener::bind(&addr).await?; log::info!("Server listening on {}", addr); @@ -83,18 +97,21 @@ async fn index() -> &'static str { "## } -async fn slug_root() -> Json { - Json(json!({ - "status": "fail", - "data": {"slug": "A slug is required: /slug/"} - })) +async fn slug_root() -> impl IntoResponse { + ( + StatusCode::BAD_REQUEST, + Json(json!({ + "status": "fail", + "data": {"slug": "A slug is required: /slug/"} + })), + ) } async fn post_meta( State(state): State, Path(slug): Path, ) -> Result, AppError> { - let result = sqlx::query("SELECT id, slug, rkey, time_us FROM posts WHERE slug = $1") + let result = sqlx::query("SELECT slug, rkey, time_us FROM posts WHERE slug = $1") .bind(&slug) .fetch_one(&state.pool) .await; @@ -102,19 +119,18 @@ async fn post_meta( match result { Ok(row) => { let meta = models::Meta { - id: row.get(0), - slug: row.get(1), - rkey: row.get(2), - time_us: row.get(3), + slug: row.get(0), + rkey: row.get(1), + time_us: row.get(2), }; Ok(Json(json!({ "status": "success", "data": {"post": meta} }))) } - Err(_) => { + Err(sqlx::Error::RowNotFound) => { // Not in DB — check the live RSS feed - match rss_poller::lookup_slug_in_rss(&slug).await { + match rss_poller::lookup_slug_in_rss(&state.client, &slug, &state.poller_config).await { Some((rkey, time_us)) => { // Insert; ignore conflicts in case the background poller raced us let _ = sqlx::query( @@ -122,29 +138,28 @@ async fn post_meta( ) .bind(&slug) .bind(&rkey) - .bind(&time_us) + .bind(time_us) .execute(&state.pool) .await; - let meta = models::Meta { - id: 0, // Will be fetched from DB on next request - slug, - rkey, - time_us, - }; Ok(Json(json!({ "status": "success", - "data": {"post": meta} + "data": {"post": models::Meta { slug, rkey, time_us }} }))) } None => Err(AppError::NotFound), } } + Err(e) => { + log::error!("Database error looking up slug '{}': {}", slug, e); + Err(AppError::DatabaseError) + } } } enum AppError { NotFound, + DatabaseError, } impl IntoResponse for AppError { @@ -156,7 +171,6 @@ impl IntoResponse for AppError { header::CACHE_CONTROL, header::HeaderValue::from_static("no-store"), ); - // Return (StatusCode, HeaderMap, Body) so the header is sent. ( StatusCode::NOT_FOUND, headers, @@ -167,6 +181,14 @@ impl IntoResponse for AppError { ) .into_response() } + AppError::DatabaseError => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ + "status": "error", + "message": "An internal error occurred" + })), + ) + .into_response(), } } } diff --git a/src/models.rs b/src/models.rs index c397c92..a80e000 100644 --- a/src/models.rs +++ b/src/models.rs @@ -2,8 +2,7 @@ use serde::Serialize; #[derive(Serialize, Debug)] pub struct Meta { - pub id: i32, pub slug: String, pub rkey: String, - pub time_us: String, + pub time_us: i64, } diff --git a/src/rss_poller.rs b/src/rss_poller.rs index afbb3cd..64d2232 100644 --- a/src/rss_poller.rs +++ b/src/rss_poller.rs @@ -1,47 +1,21 @@ -use serde::Deserialize; +use crate::settings::PollerConfig; use tokio::time::{sleep, Duration}; use url::Url; -thread_local! { - static POLLER_CONFIG: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; -} - -#[derive(Deserialize, Debug, Clone)] -pub struct PollerConfig { - pub handle: String, - pub emoji: String, - pub domain: String, -} - -impl PollerConfig { - /// Load poller config from the application's config builder - pub fn from_config() -> Result { - let cfg = - crate::settings::build_config().map_err(|e| format!("Failed to load config: {}", e))?; - - let poster_handle = cfg - .get::("poller.handle") - .map_err(|_| "ATC_POLLER_HANDLE not set".to_string())?; - let target_emoji = cfg - .get::("poller.emoji") - .map_err(|_| "ATC_POLLER_EMOJI not set".to_string())?; - let blog_domain = cfg - .get::("poller.domain") - .map_err(|_| "ATC_POLLER_DOMAIN not set".to_string())?; - - Ok(PollerConfig { - handle: poster_handle, - emoji: target_emoji, - domain: blog_domain, - }) - } +/// A parsed post entry extracted from an RSS item. +struct PostEntry { + slug: String, + rkey: String, + time_us: i64, } /// Fetch and parse RSS feed from Bluesky profile -async fn fetch_rss(handle: &str) -> Result { +async fn fetch_rss(client: &reqwest::Client, handle: &str) -> Result { let url = format!("https://bsky.app/profile/{}/rss", handle); - let response = reqwest::get(&url) + let response = client + .get(&url) + .send() .await .map_err(|e| format!("Failed to fetch RSS: {}", e))?; @@ -99,16 +73,12 @@ fn find_blog_urls(description: &str, target_emoji: &str, blog_domain: &str) -> V .collect() } -/// Poll RSS feed and update database -async fn poll_rss(pool: &sqlx::Pool, config: &PollerConfig) -> Result<(), String> { - log::info!("Polling RSS feed for {}", config.handle); - - let channel = fetch_rss(&config.handle).await?; - - let mut processed = 0; +/// Parse all matching post entries from an RSS channel. +/// Returns one `PostEntry` per (slug, rkey) pair found. +fn parse_rss_items(channel: &rss::Channel, config: &PollerConfig) -> Vec { + let mut entries = Vec::new(); for item in channel.items() { - // Extract rkey from GUID (contains at:// URI) let guid = match item.guid() { Some(g) => g.value(), None => continue, @@ -122,50 +92,72 @@ async fn poll_rss(pool: &sqlx::Pool, config: &PollerConfig) -> R } }; - // Get post description/content let description = match item.description() { Some(d) => d, None => continue, }; - // Check for target emoji and blog URLs let urls = find_blog_urls(description, &config.emoji, &config.domain); - if urls.is_empty() { continue; } - // Extract timestamp from pub_date if available let time_us = item .pub_date() .and_then(|date_str| chrono::DateTime::parse_from_rfc2822(date_str).ok()) - .map(|dt| dt.timestamp_micros().to_string()) - .unwrap_or_else(|| chrono::Utc::now().timestamp_micros().to_string()); - - // Process each blog URL found - for url in urls { - if let Some(slug) = extract_slug_from_url(&url, &config.domain) { - let insert_result = sqlx::query( - "INSERT INTO posts (slug, rkey, time_us) VALUES ($1, $2, $3) ON CONFLICT (slug) DO NOTHING" - ) - .bind(&slug) - .bind(&rkey) - .bind(&time_us) - .execute(pool) - .await; - - match insert_result { - Ok(result) => { - if result.rows_affected() > 0 { - log::info!("Inserted new post: slug={}, rkey={}", slug, rkey); - processed += 1; - } - } - Err(e) => { - log::error!("Failed to insert post {}: {}", slug, e); - } + .map(|dt| dt.timestamp_micros()) + .unwrap_or_else(|| chrono::Utc::now().timestamp_micros()); + + for url in &urls { + if let Some(slug) = extract_slug_from_url(url, &config.domain) { + entries.push(PostEntry { + slug, + rkey: rkey.clone(), + time_us, + }); + } + } + } + + entries +} + +/// Poll RSS feed and update database +async fn poll_rss( + client: &reqwest::Client, + pool: &sqlx::Pool, + config: &PollerConfig, +) -> Result<(), String> { + log::info!("Polling RSS feed for {}", config.handle); + + let channel = fetch_rss(client, &config.handle).await?; + let entries = parse_rss_items(&channel, config); + let mut processed = 0; + + for entry in entries { + let insert_result = sqlx::query( + "INSERT INTO posts (slug, rkey, time_us) VALUES ($1, $2, $3) ON CONFLICT (slug) DO NOTHING" + ) + .bind(&entry.slug) + .bind(&entry.rkey) + .bind(entry.time_us) + .execute(pool) + .await; + + match insert_result { + Ok(result) => { + if result.rows_affected() > 0 { + log::info!( + "Inserted new post: slug={}, rkey={}", + entry.slug, + entry.rkey + ); + processed += 1; } } + Err(e) => { + log::error!("Failed to insert post {}: {}", entry.slug, e); + } } } @@ -175,16 +167,12 @@ async fn poll_rss(pool: &sqlx::Pool, config: &PollerConfig) -> R /// Look up a specific slug in the RSS feed on demand. /// Returns `(rkey, time_us)` if the slug is found, `None` otherwise. -pub async fn lookup_slug_in_rss(slug: &str) -> Option<(String, String)> { - let config = match PollerConfig::from_config() { - Ok(c) => c, - Err(e) => { - log::error!("Failed to load config: {}", e); - return None; - } - }; - - let channel = match fetch_rss(&config.handle).await { +pub async fn lookup_slug_in_rss( + client: &reqwest::Client, + slug: &str, + config: &PollerConfig, +) -> Option<(String, i64)> { + let channel = match fetch_rss(client, &config.handle).await { Ok(c) => c, Err(e) => { log::error!("Failed to fetch RSS for on-demand lookup: {}", e); @@ -192,53 +180,20 @@ pub async fn lookup_slug_in_rss(slug: &str) -> Option<(String, String)> { } }; - for item in channel.items() { - let guid = match item.guid() { - Some(g) => g.value(), - None => continue, - }; - - let rkey = match extract_rkey(guid) { - Some(r) => r, - None => continue, - }; - - let description = match item.description() { - Some(d) => d, - None => continue, - }; - - let urls = find_blog_urls(description, &config.emoji, &config.domain); + let entries = parse_rss_items(&channel, config); - for url in &urls { - match extract_slug_from_url(url, &config.domain) { - Some(found_slug) if found_slug == slug => { - let time_us = item - .pub_date() - .and_then(|date_str| chrono::DateTime::parse_from_rfc2822(date_str).ok()) - .map(|dt| dt.timestamp_micros().to_string()) - .unwrap_or_else(|| chrono::Utc::now().timestamp_micros().to_string()); - log::info!("On-demand lookup found slug={} rkey={}", slug, rkey); - return Some((rkey, time_us)); - } - _ => {} - } - } - } - - None + entries.into_iter().find(|e| e.slug == slug).map(|e| { + log::info!("On-demand lookup found slug={} rkey={}", slug, e.rkey); + (e.rkey, e.time_us) + }) } /// Background task that polls RSS every 15 minutes -pub async fn rss_polling_task(pool: sqlx::Pool) { - let config = match PollerConfig::from_config() { - Ok(c) => c, - Err(e) => { - log::error!("Failed to load config: {}", e); - return; - } - }; - +pub async fn rss_polling_task( + client: reqwest::Client, + pool: sqlx::Pool, + config: PollerConfig, +) { log::info!( "Starting RSS poller for {} (emoji: {}, domain: {})", config.handle, @@ -247,7 +202,7 @@ pub async fn rss_polling_task(pool: sqlx::Pool) { ); // Do initial poll immediately - if let Err(e) = poll_rss(&pool, &config).await { + if let Err(e) = poll_rss(&client, &pool, &config).await { log::error!("Initial poll failed: {}", e); } @@ -257,7 +212,7 @@ pub async fn rss_polling_task(pool: sqlx::Pool) { loop { sleep(interval).await; - if let Err(e) = poll_rss(&pool, &config).await { + if let Err(e) = poll_rss(&client, &pool, &config).await { log::error!("Poll failed: {}", e); } } diff --git a/src/settings.rs b/src/settings.rs index 21f7da6..5f439ea 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -1,33 +1,33 @@ -pub use config::{Config, ConfigError, Environment, File}; +use config::Config; +pub use config::{ConfigError, Environment, File}; use serde::Deserialize; -#[derive(Deserialize)] -#[allow(unused)] -struct DatabaseConfig { - url: String, +#[derive(Deserialize, Clone)] +pub struct DatabaseConfig { + pub url: String, } -#[derive(Deserialize)] -#[allow(unused)] -struct AppConfig { - address: String, - port: u16, + +#[derive(Deserialize, Clone)] +pub struct AppConfig { + pub address: String, + pub port: u16, } -#[derive(Deserialize)] -#[allow(unused)] -struct PollerConfig { - handle: String, - emoji: String, - domain: String, + +#[derive(Deserialize, Debug, Clone)] +pub struct PollerConfig { + pub handle: String, + pub emoji: String, + pub domain: String, } -#[derive(Deserialize)] -#[allow(unused)] + +#[derive(Deserialize, Clone)] pub struct Settings { - database: DatabaseConfig, - app: AppConfig, - poller: PollerConfig, + pub database: DatabaseConfig, + pub app: AppConfig, + pub poller: PollerConfig, } -pub fn build_config() -> Result { +pub fn build_config() -> Result { Config::builder() .add_source(File::with_name("Settings").required(false)) .add_source( @@ -36,5 +36,6 @@ pub fn build_config() -> Result { .separator("_") .prefix("ATC"), ) - .build() + .build()? + .try_deserialize() }