diff --git a/Dockerfile b/Dockerfile index c7ce44a5..a65e7290 100644 --- a/Dockerfile +++ b/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"] diff --git a/docs/deployment.md b/docs/deployment.md index 79e06904..d12c9d06 100644 --- a/docs/deployment.md +++ b/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. diff --git a/services/api/src/handlers.rs b/services/api/src/handlers.rs index 314bc8a3..92eefe08 100644 --- a/services/api/src/handlers.rs +++ b/services/api/src/handlers.rs @@ -1848,57 +1848,48 @@ pub async fn rotate_api_key( ) -> Result { 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 = 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> = None; - let mut old_label: Option = 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 = 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"); diff --git a/services/api/src/main.rs b/services/api/src/main.rs index b6af3562..6fdc3b30 100644 --- a/services/api/src/main.rs +++ b/services/api/src/main.rs @@ -51,6 +51,27 @@ fn env_parse(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"), diff --git a/services/api/src/migrations.rs b/services/api/src/migrations.rs index da55dd94..86505c98 100644 --- a/services/api/src/migrations.rs +++ b/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",