feat: add remote, with rss, cache, discord, and zulip

remote/rss-service/src/main.rs

@@ -0,0 +1,236 @@
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use axum::extract::{Query, State};
+use axum::http::{HeaderMap, StatusCode, header};
+use axum::response::IntoResponse;
+use axum::routing::{get, post};
+use axum::{Json, Router};
+use rusqlite::Connection;
+use serde::Deserialize;
+use tokio::sync::Mutex;
+use tracing::{error, info};
+
+struct AppState {
+    db: Arc<Mutex<Connection>>,
+    webhook_secret: String,
+    site_url: String,
+}
+
+#[derive(Deserialize)]
+struct WebhookPayload {
+    status: String,
+    timestamp: u64,
+}
+
+#[derive(Deserialize)]
+struct FeedQuery {
+    limit: Option<u32>,
+}
+
+fn unix_to_rfc3339(ts: u64) -> String {
+    let dt = time::OffsetDateTime::from_unix_timestamp(ts as i64).unwrap_or(time::OffsetDateTime::UNIX_EPOCH);
+    dt.format(&time::format_description::well_known::Rfc3339).unwrap_or_else(|_| "1970-01-01T00:00:00Z".to_string())
+}
+
+fn status_description(status: &str) -> &str {
+    match status {
+        "open" => "The door at Noisebridge is open.",
+        "closed" => "The door at Noisebridge is closed.",
+        "offline" => "The Noisebridge Pi is offline.",
+        _ => "Unknown status.",
+    }
+}
+
+fn status_title(status: &str) -> &str {
+    match status {
+        "open" => "Door is open",
+        "closed" => "Door is closed",
+        "offline" => "Pi is offline",
+        _ => "Unknown",
+    }
+}
+
+fn validate_bearer(headers: &HeaderMap, expected: &str) -> bool {
+    headers
+        .get("authorization")
+        .and_then(|v| v.to_str().ok())
+        .map(|v| v.strip_prefix("Bearer ").unwrap_or("") == expected)
+        .unwrap_or(false)
+}
+
+fn init_db(path: &str) -> Result<Connection> {
+    let conn = Connection::open(path).context("failed to open SQLite database")?;
+    conn.execute_batch(
+        "CREATE TABLE IF NOT EXISTS events (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            status TEXT NOT NULL,
+            timestamp INTEGER NOT NULL,
+            received_at INTEGER NOT NULL
+        );",
+    )
+    .context("failed to initialize database schema")?;
+    Ok(conn)
+}
+
+fn unix_now() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_secs()
+}
+
+async fn post_webhook(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    Json(body): Json<WebhookPayload>,
+) -> StatusCode {
+    if !validate_bearer(&headers, &state.webhook_secret) {
+        return StatusCode::UNAUTHORIZED;
+    }
+
+    let now = unix_now();
+    let conn = state.db.lock().await;
+    match conn.execute(
+        "INSERT INTO events (status, timestamp, received_at) VALUES (?1, ?2, ?3)",
+        rusqlite::params![body.status, body.timestamp, now],
+    ) {
+        Ok(_) => {
+            info!(status = %body.status, timestamp = body.timestamp, "event recorded");
+            StatusCode::OK
+        }
+        Err(e) => {
+            error!(error = %e, "failed to insert event");
+            StatusCode::INTERNAL_SERVER_ERROR
+        }
+    }
+}
+
+async fn get_feed(
+    State(state): State<Arc<AppState>>,
+    Query(query): Query<FeedQuery>,
+) -> impl IntoResponse {
+    let limit = query.limit.unwrap_or(50);
+    let conn = state.db.lock().await;
+
+    let mut stmt = match conn.prepare(
+        "SELECT status, timestamp FROM events ORDER BY id DESC LIMIT ?1",
+    ) {
+        Ok(s) => s,
+        Err(e) => {
+            error!(error = %e, "failed to prepare query");
+            return (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response();
+        }
+    };
+
+    let entries: Vec<(String, u64)> = match stmt
+        .query_map(rusqlite::params![limit], |row| {
+            Ok((row.get::<_, String>(0)?, row.get::<_, u64>(1)?))
+        }) {
+        Ok(rows) => rows.filter_map(|r| r.ok()).collect(),
+        Err(e) => {
+            error!(error = %e, "failed to query events");
+            return (StatusCode::INTERNAL_SERVER_ERROR, "internal error").into_response();
+        }
+    };
+
+    let updated = entries
+        .first()
+        .map(|(_, ts)| unix_to_rfc3339(*ts))
+        .unwrap_or_else(|| unix_to_rfc3339(unix_now()));
+
+    let mut xml = format!(
+        r#"<?xml version="1.0" encoding="utf-8"?>
+<feed xmlns="http://www.w3.org/2005/Atom">
+  <title>Noisebell Door Status</title>
+  <link href="{site_url}/feed" rel="self"/>
+  <link href="{site_url}" rel="alternate"/>
+  <id>urn:noisebell:door-status</id>
+  <updated>{updated}</updated>
+"#,
+        site_url = state.site_url,
+        updated = updated,
+    );
+
+    for (status, timestamp) in &entries {
+        let ts_rfc = unix_to_rfc3339(*timestamp);
+        xml.push_str(&format!(
+            r#"  <entry>
+    <title>{title}</title>
+    <id>urn:noisebell:event:{timestamp}</id>
+    <updated>{ts}</updated>
+    <content type="text">{description}</content>
+  </entry>
+"#,
+            title = status_title(status),
+            timestamp = timestamp,
+            ts = ts_rfc,
+            description = status_description(status),
+        ));
+    }
+
+    xml.push_str("</feed>\n");
+
+    (
+        StatusCode::OK,
+        [(header::CONTENT_TYPE, "application/atom+xml; charset=utf-8")],
+        xml,
+    )
+        .into_response()
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let port: u16 = std::env::var("NOISEBELL_RSS_PORT")
+        .unwrap_or_else(|_| "3002".into())
+        .parse()
+        .context("NOISEBELL_RSS_PORT must be a valid u16")?;
+
+    let webhook_secret = std::env::var("NOISEBELL_RSS_WEBHOOK_SECRET")
+        .context("NOISEBELL_RSS_WEBHOOK_SECRET is required")?;
+
+    let data_dir =
+        std::env::var("NOISEBELL_RSS_DATA_DIR").unwrap_or_else(|_| "/var/lib/noisebell-rss".into());
+
+    let site_url = std::env::var("NOISEBELL_RSS_SITE_URL")
+        .unwrap_or_else(|_| format!("https://rss.noisebell.extremist.software"));
+
+    info!(port, "starting noisebell-rss");
+
+    let db_path = format!("{data_dir}/rss.db");
+    let conn = init_db(&db_path)?;
+    let db = Arc::new(Mutex::new(conn));
+
+    let app_state = Arc::new(AppState {
+        db,
+        webhook_secret,
+        site_url,
+    });
+
+    let app = Router::new()
+        .route("/webhook", post(post_webhook))
+        .route("/feed", get(get_feed))
+        .with_state(app_state);
+
+    let listener = tokio::net::TcpListener::bind(("0.0.0.0", port))
+        .await
+        .context(format!("failed to bind to 0.0.0.0:{port}"))?;
+
+    info!(port, "listening");
+
+    let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+        .context("failed to register SIGTERM handler")?;
+    axum::serve(listener, app)
+        .with_graceful_shutdown(async move {
+            sigterm.recv().await;
+        })
+        .await
+        .context("server error")?;
+
+    info!("shutdown complete");
+    Ok(())
+}