diff --git a/crates/mailify-api/src/lib.rs b/crates/mailify-api/src/lib.rs index a8776cd..0e6ed17 100644 --- a/crates/mailify-api/src/lib.rs +++ b/crates/mailify-api/src/lib.rs @@ -8,8 +8,13 @@ use std::sync::Arc; use axum::{http::StatusCode, middleware, routing::get, Router}; use mailify_auth::{middleware::AuthLayer, require_jwt}; use tower_http::{ - cors::CorsLayer, limit::RequestBodyLimitLayer, timeout::TimeoutLayer, trace::TraceLayer, + cors::CorsLayer, + limit::RequestBodyLimitLayer, + timeout::TimeoutLayer, + trace::{DefaultMakeSpan, DefaultOnRequest, DefaultOnResponse, TraceLayer}, + LatencyUnit, }; +use tracing::Level; use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; @@ -29,6 +34,7 @@ pub fn build_router(state: AppState) -> Router { "/mail/send-custom", axum::routing::post(routes::mail::send_custom), ) + .route("/mail/jobs/:id", get(routes::mail::get_job_state)) .route("/templates", get(routes::templates::list)) .route( "/templates/:id/preview", @@ -52,7 +58,21 @@ pub fn build_router(state: AppState) -> Router { .merge(protected) .with_state(Arc::new(state)) .merge(swagger) - .layer(TraceLayer::new_for_http()) + .layer( + TraceLayer::new_for_http() + .make_span_with( + DefaultMakeSpan::new() + .level(Level::INFO) + .include_headers(false), + ) + .on_request(DefaultOnRequest::new().level(Level::INFO)) + .on_response( + DefaultOnResponse::new() + .level(Level::INFO) + .latency_unit(LatencyUnit::Millis) + .include_headers(false), + ), + ) .layer(CorsLayer::very_permissive()) .layer(RequestBodyLimitLayer::new(10 * 1024 * 1024)) .layer(TimeoutLayer::with_status_code( diff --git a/crates/mailify-api/src/main.rs b/crates/mailify-api/src/main.rs index 9ccbac1..0eb1b1c 100644 --- a/crates/mailify-api/src/main.rs +++ b/crates/mailify-api/src/main.rs @@ -1,7 +1,9 @@ use std::sync::Arc; use mailify_api::{build_router, AppState}; -use mailify_auth::JwtIssuer; +use mailify_auth::{ + generate_bootstrap_key, generate_jwt_secret, print_bootstrap_banner, JwtIssuer, +}; use mailify_config::{AppConfig, LogFormat}; use mailify_queue::{worker::WorkerDeps, QueueRuntime}; use mailify_smtp::SmtpSender; @@ -12,11 +14,13 @@ use tracing_subscriber::{prelude::*, EnvFilter}; #[tokio::main] async fn main() -> anyhow::Result<()> { - let cfg = AppConfig::load()?; + let mut cfg = AppConfig::load()?; init_tracing(&cfg); info!(version = env!("CARGO_PKG_VERSION"), "starting mailify"); + maybe_bootstrap_auth(&mut cfg); + info!(api_key_ids = ?cfg.auth.api_keys.keys().cloned().collect::>(), "loaded auth api_key ids"); let registry = Arc::new(TemplateRegistry::load_from_dir( @@ -80,15 +84,58 @@ async fn main() -> anyhow::Result<()> { Ok(()) } +const DEFAULT_JWT_SECRET: &str = "CHANGE_ME_IN_PRODUCTION"; +const BOOTSTRAP_KEY_ID: &str = "DEFAULT"; + +fn maybe_bootstrap_auth(cfg: &mut AppConfig) { + if !cfg.auth.bootstrap || !cfg.auth.api_keys.is_empty() { + return; + } + + let key = match generate_bootstrap_key(BOOTSTRAP_KEY_ID) { + Ok(k) => k, + Err(e) => { + error!(error = %e, "failed to generate bootstrap api key"); + return; + } + }; + + let jwt_override = if cfg.auth.jwt_secret == DEFAULT_JWT_SECRET { + let secret = generate_jwt_secret(); + cfg.auth.jwt_secret = secret.clone(); + Some(secret) + } else { + None + }; + + cfg.auth.api_keys.insert(key.id.clone(), key.hash.clone()); + + print_bootstrap_banner(&key, jwt_override.as_deref()); +} + fn init_tracing(cfg: &AppConfig) { - let filter = - EnvFilter::try_new(&cfg.observability.log_level).unwrap_or_else(|_| EnvFilter::new("info")); + // Precedence: RUST_LOG (standard) > cfg.observability.log_level > built-in fallback. + let filter = EnvFilter::try_from_default_env() + .or_else(|_| EnvFilter::try_new(&cfg.observability.log_level)) + .unwrap_or_else(|_| { + EnvFilter::new( + "info,mailify=debug,mailify_api=debug,mailify_queue=debug,tower_http=info", + ) + }); let registry = tracing_subscriber::registry().with(filter); match cfg.observability.log_format { LogFormat::Json => registry .with(tracing_subscriber::fmt::layer().json()) .init(), - LogFormat::Pretty => registry.with(tracing_subscriber::fmt::layer()).init(), + LogFormat::Pretty => registry + .with( + tracing_subscriber::fmt::layer() + .with_target(true) + .with_thread_ids(false) + .with_level(true) + .compact(), + ) + .init(), } } diff --git a/crates/mailify-api/src/openapi.rs b/crates/mailify-api/src/openapi.rs index f7a320a..9b865ae 100644 --- a/crates/mailify-api/src/openapi.rs +++ b/crates/mailify-api/src/openapi.rs @@ -42,6 +42,7 @@ impl Modify for SecurityAddon { routes::config::get_config, routes::mail::send_registered, routes::mail::send_custom, + routes::mail::get_job_state, ), components(schemas( // Domain types @@ -69,6 +70,7 @@ impl Modify for SecurityAddon { routes::mail::SendRegisteredRequest, routes::mail::SendCustomRequest, routes::mail::EnqueuedResponse, + routes::mail::JobStateResponse, )), tags( (name = "auth", description = "JWT issuance"), diff --git a/crates/mailify-api/src/routes/mail.rs b/crates/mailify-api/src/routes/mail.rs index 4f06bd8..e93eada 100644 --- a/crates/mailify-api/src/routes/mail.rs +++ b/crates/mailify-api/src/routes/mail.rs @@ -1,12 +1,18 @@ use std::sync::Arc; -use axum::{extract::State, Json}; +use axum::{ + extract::{Path, State}, + Json, +}; use mailify_core::{ email::{Attachment, EmailAddress}, priority::Priority, smtp_override::SmtpOverride, }; -use mailify_queue::job::{MailJob, MailJobKind}; +use mailify_queue::{ + job::{MailJob, MailJobKind}, + JobSnapshot, +}; use serde::{Deserialize, Serialize}; use serde_json::Value; use utoipa::ToSchema; @@ -66,8 +72,39 @@ pub struct SendCustomRequest { #[derive(Debug, Serialize, ToSchema)] pub struct EnqueuedResponse { - pub job_id: Uuid, + /// Queue task id (ULID). Use it with `GET /mail/jobs/{id}` to poll state. + pub job_id: String, + pub status: String, +} + +#[derive(Debug, Serialize, ToSchema)] +pub struct JobStateResponse { + pub task_id: String, + pub mail_id: Uuid, + /// One of: `pending`, `scheduled`, `running`, `done`, `failed`, `killed`. pub status: String, + pub attempts: usize, + pub max_attempts: i32, + pub last_error: Option, + pub run_at: chrono::DateTime, + pub lock_at: Option, + pub done_at: Option, +} + +impl From for JobStateResponse { + fn from(s: JobSnapshot) -> Self { + Self { + task_id: s.task_id, + mail_id: s.mail_id, + status: s.status, + attempts: s.attempts, + max_attempts: s.max_attempts, + last_error: s.last_error, + run_at: s.run_at, + lock_at: s.lock_at, + done_at: s.done_at, + } + } } /// Queue an email using a built-in template (see `GET /templates` for available ids). @@ -121,7 +158,7 @@ pub async fn send_registered( .map_err(|e| ApiError::Internal(e.to_string()))?; Ok(Json(EnqueuedResponse { job_id: id, - status: "queued".into(), + status: "pending".into(), })) } @@ -179,10 +216,37 @@ pub async fn send_custom( .map_err(|e| ApiError::Internal(e.to_string()))?; Ok(Json(EnqueuedResponse { job_id: id, - status: "queued".into(), + status: "pending".into(), })) } +/// Fetch the current state of a previously-enqueued job. +#[utoipa::path( + get, + path = "/mail/jobs/{id}", + tag = "mail", + params(("id" = String, Path, description = "Task id returned by /mail/send or /mail/send-custom")), + security(("bearer_jwt" = [])), + responses( + (status = 200, description = "Job state", body = JobStateResponse), + (status = 400, description = "Malformed id"), + (status = 401, description = "Missing or invalid JWT"), + (status = 404, description = "Unknown job id (or vacuumed)"), + ) +)] +pub async fn get_job_state( + State(state): State>, + Path(id): Path, +) -> Result, ApiError> { + let mut queue = state.queue.clone(); + match queue.fetch(&id).await { + Ok(Some(snap)) => Ok(Json(snap.into())), + Ok(None) => Err(ApiError::NotFound), + Err(mailify_queue::worker::QueueError::InvalidId(m)) => Err(ApiError::BadRequest(m)), + Err(e) => Err(ApiError::Internal(e.to_string())), + } +} + fn default_from( state: &AppState, supplied: Option, diff --git a/crates/mailify-api/tests/it_e2e.rs b/crates/mailify-api/tests/it_e2e.rs index ea7c8e4..66262e5 100644 --- a/crates/mailify-api/tests/it_e2e.rs +++ b/crates/mailify-api/tests/it_e2e.rs @@ -66,6 +66,7 @@ fn build_test_cfg(url: String) -> AppConfig { jwt_issuer: "mailify-e2e".into(), jwt_ttl_secs: 300, api_keys: Default::default(), + bootstrap: false, }, queue: QueueConfig { worker_concurrency: 2, @@ -166,7 +167,7 @@ async fn send_custom_via_http_delivers_to_mailpit() { .await .unwrap(); let resp: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); - assert_eq!(resp["status"], "queued"); + assert_eq!(resp["status"], "pending"); // Wait for mailpit delivery. let search_url = format!( diff --git a/crates/mailify-auth/src/bootstrap.rs b/crates/mailify-auth/src/bootstrap.rs new file mode 100644 index 0000000..681e5ff --- /dev/null +++ b/crates/mailify-auth/src/bootstrap.rs @@ -0,0 +1,93 @@ +//! First-boot bootstrap: when no API key is configured, generate an ephemeral one +//! and print the env line the operator needs to copy back into their environment. + +use rand_core::{OsRng, RngCore}; + +use crate::api_key::{hash_api_key, ApiKeyError}; + +/// Result of a bootstrap run. +pub struct BootstrapKey { + pub id: String, + pub plaintext: String, + pub hash: String, +} + +/// Generate a random url-safe plaintext key and return it alongside its argon2 hash. +pub fn generate_bootstrap_key(id: impl Into) -> Result { + let plaintext = random_token(32); + let hash = hash_api_key(&plaintext)?; + Ok(BootstrapKey { + id: id.into(), + plaintext, + hash, + }) +} + +/// Generate a random JWT secret (used when the default placeholder is still in place). +pub fn generate_jwt_secret() -> String { + random_token(48) +} + +fn random_token(byte_len: usize) -> String { + const ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"; + let mut bytes = vec![0u8; byte_len]; + OsRng.fill_bytes(&mut bytes); + bytes + .into_iter() + .map(|b| ALPHABET[(b as usize) % ALPHABET.len()] as char) + .collect() +} + +/// Print a human-readable banner to stderr so it survives JSON log mode. +pub fn print_bootstrap_banner(key: &BootstrapKey, jwt_secret_generated: Option<&str>) { + let mut lines = vec![ + "".to_string(), + "===================== MAILIFY BOOTSTRAP =====================".to_string(), + "No API key was configured. An ephemeral key has been generated".to_string(), + "for this session. It will NOT survive a restart unless you add".to_string(), + "the hash below to your environment (e.g. .env / compose env).".to_string(), + "".to_string(), + format!(" Key id: {}", key.id), + format!(" Plaintext: {}", key.plaintext), + "".to_string(), + " Paste into your env and restart to persist:".to_string(), + format!(" MAILIFY_AUTH__API_KEYS__{}={}", key.id, key.hash), + ]; + if let Some(secret) = jwt_secret_generated { + lines.push("".to_string()); + lines + .push(" JWT secret was the default placeholder — generated a random one.".to_string()); + lines.push(" Persist it too (otherwise issued tokens die on restart):".to_string()); + lines.push(format!(" MAILIFY_AUTH__JWT_SECRET={secret}")); + } + lines.push("=============================================================".to_string()); + lines.push("".to_string()); + eprintln!("{}", lines.join("\n")); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::api_key::verify_api_key; + + #[test] + fn bootstrap_key_verifies_against_its_hash() { + let key = generate_bootstrap_key("DEFAULT").unwrap(); + assert_eq!(key.id, "DEFAULT"); + assert!(verify_api_key(&key.plaintext, &key.hash).unwrap()); + } + + #[test] + fn random_tokens_are_unique_and_sized() { + let a = random_token(32); + let b = random_token(32); + assert_eq!(a.len(), 32); + assert_ne!(a, b); + } + + #[test] + fn jwt_secret_is_non_empty() { + let s = generate_jwt_secret(); + assert_eq!(s.len(), 48); + } +} diff --git a/crates/mailify-auth/src/lib.rs b/crates/mailify-auth/src/lib.rs index a281fce..3a6df11 100644 --- a/crates/mailify-auth/src/lib.rs +++ b/crates/mailify-auth/src/lib.rs @@ -1,7 +1,11 @@ pub mod api_key; +pub mod bootstrap; pub mod jwt; pub mod middleware; pub use api_key::{hash_api_key, verify_api_key, ApiKeyError}; +pub use bootstrap::{ + generate_bootstrap_key, generate_jwt_secret, print_bootstrap_banner, BootstrapKey, +}; pub use jwt::{Claims, JwtError, JwtIssuer}; pub use middleware::{require_jwt, AuthLayer}; diff --git a/crates/mailify-config/src/lib.rs b/crates/mailify-config/src/lib.rs index 8f44a7d..b5755cd 100644 --- a/crates/mailify-config/src/lib.rs +++ b/crates/mailify-config/src/lib.rs @@ -77,6 +77,16 @@ pub struct AuthConfig { /// API keys (argon2 hashes), key = identifier, value = hash. make hash-key KEY=CHANGE_ME_IN_PRODUCTION #[serde(default)] pub api_keys: std::collections::HashMap, + /// When true (default) and `api_keys` is empty at boot, the server generates + /// an ephemeral API key, prints the plaintext + the env line to paste, and keeps + /// the hash in memory for the current session. Set `false` to disable (e.g. when + /// intentionally running with no keys). + #[serde(default = "default_true")] + pub bootstrap: bool, +} + +fn default_true() -> bool { + true } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -143,6 +153,7 @@ impl Default for AppConfig { jwt_issuer: "mailify".into(), jwt_ttl_secs: 3600, api_keys: Default::default(), + bootstrap: true, }, queue: QueueConfig { worker_concurrency: 4, diff --git a/crates/mailify-queue/src/lib.rs b/crates/mailify-queue/src/lib.rs index 5df9b66..9c743fa 100644 --- a/crates/mailify-queue/src/lib.rs +++ b/crates/mailify-queue/src/lib.rs @@ -8,4 +8,4 @@ pub mod job; pub mod worker; pub use job::{MailJob, MailJobKind}; -pub use worker::{QueueHandle, QueueRuntime}; +pub use worker::{JobSnapshot, QueueHandle, QueueRuntime}; diff --git a/crates/mailify-queue/src/worker.rs b/crates/mailify-queue/src/worker.rs index d7e02be..14373f4 100644 --- a/crates/mailify-queue/src/worker.rs +++ b/crates/mailify-queue/src/worker.rs @@ -1,8 +1,11 @@ use std::sync::Arc; +use std::time::Duration; use apalis::layers::retry::RetryPolicy; use apalis::prelude::*; +use apalis_sql::context::SqlContext; use apalis_sql::postgres::{PgListen, PostgresStorage}; +use chrono::{DateTime, Utc}; use mailify_config::{AppConfig, QueueConfig}; use mailify_core::smtp_override::SmtpOverride; use mailify_smtp::{Envelope, SmtpSender}; @@ -20,24 +23,92 @@ pub enum QueueError { Migrate(String), #[error("apalis push error: {0}")] Push(String), + #[error("invalid task id: {0}")] + InvalidId(String), + #[error("apalis fetch error: {0}")] + Fetch(String), #[error("worker error: {0}")] Worker(String), } -/// Handle used by HTTP routes to enqueue jobs. +/// Point-in-time view of a queued job's state, as exposed to HTTP clients. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct JobSnapshot { + /// apalis task id (ULID) — the externally-visible job identifier. + pub task_id: String, + /// Our domain-level MailJob id (useful to correlate logs). + pub mail_id: uuid::Uuid, + /// One of: pending, scheduled, running, done, failed, killed. + pub status: String, + pub attempts: usize, + pub max_attempts: i32, + pub last_error: Option, + pub run_at: DateTime, + /// Unix seconds when the job was locked by a worker. + pub lock_at: Option, + /// Unix seconds when the job finished (either Done or Failed). + pub done_at: Option, +} + +/// Handle used by HTTP routes to enqueue and inspect jobs. #[derive(Clone)] pub struct QueueHandle { storage: PostgresStorage, } impl QueueHandle { - pub async fn push(&mut self, job: MailJob) -> Result { - let id = job.id; - self.storage + /// Enqueue a job and return its apalis task id (ULID string). + pub async fn push(&mut self, job: MailJob) -> Result { + let parts = self + .storage .push(job) .await .map_err(|e| QueueError::Push(e.to_string()))?; - Ok(id) + Ok(parts.task_id.to_string()) + } + + /// Fetch a snapshot of a job's current state. Returns `None` if the id is unknown + /// or the job has been vacuumed out of storage. + pub async fn fetch(&mut self, task_id: &str) -> Result, QueueError> { + let parsed: TaskId = task_id + .parse() + .map_err(|e| QueueError::InvalidId(format!("{e}")))?; + let res = self + .storage + .fetch_by_id(&parsed) + .await + .map_err(|e| QueueError::Fetch(e.to_string()))?; + Ok(res.map(|req| snapshot_from_request(task_id, &req))) + } +} + +fn snapshot_from_request( + task_id: &str, + req: &apalis::prelude::Request, +) -> JobSnapshot { + let ctx = &req.parts.context; + JobSnapshot { + task_id: task_id.to_string(), + mail_id: req.args.id, + status: state_label(ctx.status()).to_string(), + attempts: req.parts.attempt.current(), + max_attempts: ctx.max_attempts(), + last_error: ctx.last_error().clone(), + run_at: *ctx.run_at(), + lock_at: *ctx.lock_at(), + done_at: *ctx.done_at(), + } +} + +fn state_label(s: &apalis::prelude::State) -> &'static str { + use apalis::prelude::State; + match s { + State::Pending => "pending", + State::Scheduled => "scheduled", + State::Running => "running", + State::Done => "done", + State::Failed => "failed", + State::Killed => "killed", } } @@ -60,15 +131,36 @@ impl QueueRuntime { app_cfg: &AppConfig, deps: WorkerDeps, ) -> Result<(Self, QueueHandle), QueueError> { + let sanitized = redact_db_url(&app_cfg.database.url); + info!( + database.url = %sanitized, + database.max_connections = app_cfg.database.max_connections, + database.min_connections = app_cfg.database.min_connections, + "connecting to postgres", + ); + let pool = PgPoolOptions::new() .max_connections(app_cfg.database.max_connections) .min_connections(app_cfg.database.min_connections) + .acquire_timeout(Duration::from_secs(10)) .connect(&app_cfg.database.url) - .await?; - - PostgresStorage::setup(&pool) .await - .map_err(|e| QueueError::Migrate(e.to_string()))?; + .map_err(|e| { + error!(error = %e, database.url = %sanitized, "postgres connection failed"); + QueueError::Sqlx(e) + })?; + + sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| { + error!(error = %e, "postgres ping (SELECT 1) failed"); + QueueError::Sqlx(e) + })?; + info!("postgres reachable"); + + PostgresStorage::setup(&pool).await.map_err(|e| { + error!(error = %e, "apalis migrations failed"); + QueueError::Migrate(e.to_string()) + })?; + info!("apalis migrations applied"); let storage = PostgresStorage::new(pool.clone()); let handle = QueueHandle { @@ -189,3 +281,41 @@ async fn process(job: MailJob, deps: Arc) -> Result<(), String> { fn sender_from_override(ov: &SmtpOverride) -> Result { SmtpSender::from_override(ov) } + +/// Strip password from a `postgres://user:pass@host/db` URL so it is safe to log. +fn redact_db_url(url: &str) -> String { + match (url.find("://"), url.find('@')) { + (Some(scheme_end), Some(at)) if scheme_end + 3 < at => { + let (prefix, rest) = url.split_at(scheme_end + 3); + let (creds, host) = rest.split_at(at - (scheme_end + 3)); + let user = creds.split(':').next().unwrap_or(""); + if user.is_empty() { + format!("{prefix}***{host}") + } else { + format!("{prefix}{user}:***{host}") + } + } + _ => url.to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn redact_db_url_masks_password() { + assert_eq!( + redact_db_url("postgres://mailify:secret@localhost:5432/mailify"), + "postgres://mailify:***@localhost:5432/mailify" + ); + } + + #[test] + fn redact_db_url_without_creds_is_unchanged() { + assert_eq!( + redact_db_url("postgres://localhost/mailify"), + "postgres://localhost/mailify" + ); + } +} diff --git a/crates/mailify-queue/tests/it_postgres.rs b/crates/mailify-queue/tests/it_postgres.rs index 55b8e54..354095b 100644 --- a/crates/mailify-queue/tests/it_postgres.rs +++ b/crates/mailify-queue/tests/it_postgres.rs @@ -69,6 +69,7 @@ fn test_cfg(url: String) -> AppConfig { jwt_issuer: "mailify".into(), jwt_ttl_secs: 60, api_keys: Default::default(), + bootstrap: false, }, queue: QueueConfig { worker_concurrency: 2, diff --git a/docker/Dockerfile b/docker/Dockerfile index 787fbf5..cb53d4c 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -12,7 +12,7 @@ RUN bun run build # ─────────────────────────── 2. Rust builder ─────────────────────────── FROM rust:1.88-slim AS rs-builder RUN apt-get update && apt-get install -y --no-install-recommends \ - pkg-config libssl-dev ca-certificates && rm -rf /var/lib/apt/lists/* + pkg-config libssl-dev ca-certificates curl unzip build-essential && rm -rf /var/lib/apt/lists/* WORKDIR /w