fix(api): key rotation, migration registry, healthcheck, migrate subcommand

Dockerfile

@@ -46,24 +46,8 @@ EXPOSE 8080
 
 USER nobody
 
-ENTRYPOINT ["predictiq-api"]
-
-RUN cargo build --release --manifest-path services/api/Cargo.toml
-
-# ---------- Runtime stage ----------
-FROM debian:bookworm-slim AS runtime
-
-RUN apt-get update && apt-get install -y \
-    ca-certificates \
-    libssl3 \
-    && rm -rf /var/lib/apt/lists/*
-
-WORKDIR /app
-
-COPY --from=builder /usr/src/app/services/api/target/release/predictiq-api /usr/local/bin/predictiq-api
-
-EXPOSE 8080
-
-USER nobody
+# Built-in probe (no curl in the image); hits the lightweight liveness endpoint.
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
+    CMD ["predictiq-api", "healthcheck"]
 
 ENTRYPOINT ["predictiq-api"]

docs/deployment.md

@@ -359,6 +359,12 @@ In the deployment pipeline, add a snapshot step **before** the migration step an
     DATABASE_URL: ${{ secrets.DATABASE_URL }}
 ```
 
+### Running migrations from the API image
+
+The API image embeds every migration and exposes a `migrate` subcommand, so migrations can run as a one-off ECS task (`command: ["predictiq-api", "migrate"]`) with no `psql` or SQL files in the runtime image. Run it after the pre-migration snapshot and before rolling out the service.
+
+The container `HEALTHCHECK` runs `predictiq-api healthcheck`, which probes the lightweight `/health/live` endpoint (no `curl` required).
+
 ### Applying migrations
 
 Migrations must be applied **before** deploying a new API version that requires them.

services/api/src/handlers.rs

@@ -1848,57 +1848,48 @@ pub async fn rotate_api_key(
 ) -> Result<impl IntoResponse, ApiError> {
     use sha2::{Digest, Sha256};
 
-    let overlap_days = body.overlap_days.unwrap_or(7).max(1) as i64;
+    let overlap_days = body.overlap_days.unwrap_or(7).clamp(1, 90) as i64;
 
-    // Generate a new cryptographically random key (256-bit hex string).
+    // Generate a new key from 32 bytes of OS randomness (256-bit hex string).
     let new_raw_key = {
-        use std::fmt::Write;
-        let bytes = uuid::Uuid::new_v4().as_bytes().to_vec();
-        let bytes2 = uuid::Uuid::new_v4().as_bytes().to_vec();
-        let combined: Vec<u8> = bytes.into_iter().chain(bytes2).collect();
-        let mut s = String::with_capacity(combined.len() * 2);
-        for b in &combined {
-            write!(s, "{:02x}", b).unwrap();
-        }
-        s
+        use rand::RngCore;
+        let mut bytes = [0u8; 32];
+        rand::rngs::OsRng.fill_bytes(&mut bytes);
+        hex::encode(bytes)
     };
     let new_hash = hex::encode(Sha256::digest(new_raw_key.as_bytes()));
 
-    // Find any existing active key(s) with the same label and schedule their expiry.
-    let active_keys = state
-        .db
-        .api_key_list_active()
-        .await
-        .map_err(into_api_error)?;
-
-    let mut old_expires_at: Option<chrono::DateTime<chrono::Utc>> = None;
-    let mut old_label: Option<String> = None;
-
     let expires_at = chrono::Utc::now() + chrono::Duration::days(overlap_days);
 
-    for key in &active_keys {
-        if key.label == body.key_label {
-            state
-                .db
-                .api_key_set_expires(&key.key_hash, expires_at)
-                .await
-                .map_err(into_api_error)?;
-            old_expires_at = Some(expires_at);
-            old_label = Some(key.label.clone());
-            tracing::info!(
-                label = %key.label,
-                expires_at = %expires_at,
-                "API key scheduled for expiry (rotation overlap window)"
-            );
-        }
-    }
+    // Expire the old key(s) and insert the new one atomically so a failure
+    // never leaves the old key expiring with no replacement.
+    let mut tx = state.db.pool().begin().await.map_err(|e| into_api_error(e.into()))?;
 
-    // Insert the new key.
-    state
-        .db
-        .api_key_insert(&new_hash, &body.key_label)
+    let expired: Vec<String> = sqlx::query_scalar(
+        "UPDATE api_keys SET expires_at = $1
+         WHERE label = $2 AND revoked_at IS NULL
+           AND (expires_at IS NULL OR expires_at > NOW())
+         RETURNING label",
+    )
+    .bind(expires_at)
+    .bind(&body.key_label)
+    .fetch_all(&mut *tx)
+    .await
+    .map_err(|e| into_api_error(e.into()))?;
+
+    sqlx::query("INSERT INTO api_keys (key_hash, label) VALUES ($1, $2)")
+        .bind(&new_hash)
+        .bind(&body.key_label)
+        .execute(&mut *tx)
         .await
-        .map_err(into_api_error)?;
+        .map_err(|e| into_api_error(e.into()))?;
+
+    tx.commit().await.map_err(|e| into_api_error(e.into()))?;
+
+    let (old_expires_at, old_label) = match expired.into_iter().next() {
+        Some(label) => (Some(expires_at), Some(label)),
+        None => (None, None),
+    };
 
     tracing::info!(label = %body.key_label, "New API key issued");
 

services/api/src/main.rs

@@ -51,6 +51,27 @@ fn env_parse<T: std::str::FromStr>(key: &str, default: T) -> T {
 async fn main() -> anyhow::Result<()> {
     let config = Config::from_env();
 
+    match std::env::args().nth(1).as_deref() {
+        // Container HEALTHCHECK probe: GET /health/live on the bind address.
+        Some("healthcheck") => {
+            use tokio::io::{AsyncReadExt, AsyncWriteExt};
+            let mut conn = tokio::net::TcpStream::connect(("127.0.0.1", config.bind_addr.port())).await?;
+            conn.write_all(b"GET /health/live HTTP/1.0\r\nHost: localhost\r\n\r\n").await?;
+            let mut buf = Vec::new();
+            conn.read_to_end(&mut buf).await?;
+            let ok = buf.starts_with(b"HTTP/1.0 200") || buf.starts_with(b"HTTP/1.1 200");
+            std::process::exit(if ok { 0 } else { 1 });
+        }
+        // One-off migration task run before service rollout.
+        Some("migrate") => {
+            let pool = sqlx::PgPool::connect(&config.database_url).await?;
+            let applied = predictiq_api::migrations::MigrationRunner::new(&pool).run().await?;
+            println!("applied {applied} migration(s)");
+            return Ok(());
+        }
+        _ => {}
+    }
+
     tracing_config::init_tracing(
         "predictiq-api",
         env!("CARGO_PKG_VERSION"),

services/api/src/migrations.rs

@@ -21,6 +21,21 @@ use tracing::{info, warn};
 mod tests {
     use super::*;
 
+    #[test]
+    fn every_sql_file_is_registered() {
+        let dir = concat!(env!("CARGO_MANIFEST_DIR"), "/database/migrations");
+        let registered: std::collections::HashSet<&str> =
+            MIGRATIONS.iter().map(|m| m.name).collect();
+        for entry in std::fs::read_dir(dir).unwrap() {
+            let path = entry.unwrap().path();
+            if path.extension().and_then(|e| e.to_str()) != Some("sql") {
+                continue;
+            }
+            let stem = path.file_stem().unwrap().to_str().unwrap().to_string();
+            assert!(registered.contains(stem.as_str()), "migration {stem} is not in MIGRATIONS");
+        }
+    }
+
     #[test]
     fn warn_macro_is_in_scope() {
         // Verify that the warn! macro compiles and runs without panic.
@@ -105,11 +120,51 @@ const MIGRATIONS: &[Migration] = &[
         name: "008_create_email_tracking",
         sql: include_str!("../database/migrations/008_create_email_tracking.sql"),
     },
+    Migration {
+        version: "009",
+        name: "009_add_newsletter_indexes",
+        sql: include_str!("../database/migrations/009_add_newsletter_indexes.sql"),
+    },
+    Migration {
+        version: "010",
+        name: "010_add_soft_delete_newsletter",
+        sql: include_str!("../database/migrations/010_add_soft_delete_newsletter.sql"),
+    },
+    Migration {
+        version: "010b",
+        name: "010_create_audit_log",
+        sql: include_str!("../database/migrations/010_create_audit_log.sql"),
+    },
     Migration {
         version: "011",
         name: "011_create_markets",
         sql: include_str!("../database/migrations/011_create_markets.sql"),
     },
+    Migration {
+        version: "012",
+        name: "012_add_performance_indexes",
+        sql: include_str!("../database/migrations/012_add_performance_indexes.sql"),
+    },
+    Migration {
+        version: "013",
+        name: "013_add_email_queue_composite_index",
+        sql: include_str!("../database/migrations/013_add_email_queue_composite_index.sql"),
+    },
+    Migration {
+        version: "014",
+        name: "014_add_email_events_composite_index",
+        sql: include_str!("../database/migrations/014_add_email_events_composite_index.sql"),
+    },
+    Migration {
+        version: "015",
+        name: "015_add_newsletter_cleanup_index",
+        sql: include_str!("../database/migrations/015_add_newsletter_cleanup_index.sql"),
+    },
+    Migration {
+        version: "016",
+        name: "016_add_markets_deadline_index",
+        sql: include_str!("../database/migrations/016_add_markets_deadline_index.sql"),
+    },
     Migration {
         version: "017",
         name: "017_create_email_dead_letter_jobs",
@@ -125,6 +180,21 @@ const MIGRATIONS: &[Migration] = &[
         name: "017_add_analytics_events_market_id",
         sql: include_str!("../database/migrations/017_add_analytics_events_market_id.sql"),
     },
+    Migration {
+        version: "017d",
+        name: "017_create_api_keys",
+        sql: include_str!("../database/migrations/017_create_api_keys.sql"),
+    },
+    Migration {
+        version: "017e",
+        name: "017_create_unsubscribe_tokens",
+        sql: include_str!("../database/migrations/017_create_unsubscribe_tokens.sql"),
+    },
+    Migration {
+        version: "017f",
+        name: "017_create_watched_transactions",
+        sql: include_str!("../database/migrations/017_create_watched_transactions.sql"),
+    },
     Migration {
         version: "018",
         name: "018_add_email_tracking_email_status_index",
@@ -140,11 +210,21 @@ const MIGRATIONS: &[Migration] = &[
         name: "019_add_newsletter_confirmed_at_index",
         sql: include_str!("../database/migrations/019_add_newsletter_confirmed_at_index.sql"),
     },
+    Migration {
+        version: "019b",
+        name: "019_create_watched_transactions",
+        sql: include_str!("../database/migrations/019_create_watched_transactions.sql"),
+    },
     Migration {
         version: "020",
         name: "020_add_audit_log_actor_time_index",
         sql: include_str!("../database/migrations/020_add_audit_log_actor_time_index.sql"),
     },
+    Migration {
+        version: "021",
+        name: "021_markets_total_volume_numeric",
+        sql: include_str!("../database/migrations/021_markets_total_volume_numeric.sql"),
+    },
     Migration {
         version: "022",
         name: "022_add_audit_filter_time_indexes",