boatramp-server 0.2.2

//! The background scheduler: the per-tick loop that drives each active
//! deployment's consumers, crons, and blob-change watchers, plus the shared
//! per-site limit/permit/error helpers the request path also uses.
//! `handlers`-gated; pulls the serve-pipeline scope in via `use super::*`.

use super::*;
use boatramp_core::project::ProjectRef;

/// How often the scheduler polls each active consumer for messages.
#[cfg(feature = "handlers")]
const SCHEDULER_TICK: Duration = Duration::from_millis(500);
/// Visibility-timeout lease per consumer delivery.
#[cfg(feature = "handlers")]
pub(super) const CONSUMER_LEASE: Duration = Duration::from_secs(30);
/// Deliveries before a message is dead-lettered.
#[cfg(feature = "handlers")]
pub(super) const CONSUMER_MAX_ATTEMPTS: u32 = 5;
/// Messages claimed per consumer per tick.
#[cfg(feature = "handlers")]
pub(super) const CONSUMER_BATCH: usize = 16;

/// Per-invocation limits from the site's caps only (consumers have no
/// per-component limit config), clamped to the engine ceiling downstream.
#[cfg(feature = "handlers")]
fn site_limits(
    site_handlers: &boatramp_core::config::HandlersSiteConfig,
) -> boatramp_handlers::Limits {
    let mut limits = boatramp_handlers::Limits::default();
    if let Some(mb) = site_handlers.max_memory_mb {
        limits.memory_bytes = (mb as usize).saturating_mul(1024 * 1024);
    }
    if let Some(ms) = site_handlers.max_timeout_ms {
        limits.timeout_ms = ms as u64;
    }
    limits
}

/// Current wall-clock decomposed into cron fields (+ a monotonic minute stamp
/// for once-per-minute dedup).
#[cfg(feature = "handlers")]
#[derive(Clone, Copy)]
pub(super) struct CronNow {
    pub(super) minute: u32,
    pub(super) hour: u32,
    pub(super) dom: u32,
    pub(super) month: u32,
    pub(super) dow: u32,
    pub(super) minute_stamp: i64,
}

#[cfg(feature = "handlers")]
impl CronNow {
    fn now() -> Self {
        use chrono::{Datelike, Timelike, Utc};
        let t = Utc::now();
        Self {
            minute: t.minute(),
            hour: t.hour(),
            dom: t.day(),
            month: t.month(),
            dow: t.weekday().num_days_from_sunday(),
            minute_stamp: t.timestamp().div_euclid(60),
        }
    }
}

/// Per-cron scheduler state: the minute we last fired in (dedup across the
/// sub-minute ticks) and whether a fire is still running (for `overlap: Skip`).
#[cfg(feature = "handlers")]
pub(super) struct CronEntry {
    last_minute: i64,
    pub(super) running: Arc<std::sync::atomic::AtomicBool>,
}

impl HandlerRuntime {
    /// Spawn the **background scheduler**: a loop that drives each *active*
    /// deployment's consumers and crons. "Active" = a site's
    /// current (production) deployment plus any site-configured background
    /// aliases; previews are never enumerated, so a preview deployment runs
    /// request handlers but **no background work**. Returns `None` when handlers
    /// are disabled (or no runtime). The caller aborts the handle on shutdown.
    #[cfg(feature = "handlers")]
    pub fn spawn_scheduler(&self, deploy: DeployStore) -> Option<tokio::task::JoinHandle<()>> {
        let inner = self.inner.clone()?;
        Some(tokio::spawn(async move {
            // Content-addressed component bytes never change, so cache them
            // across ticks (avoids re-reading the blob when a consumer is idle).
            let mut wasm_cache: std::collections::HashMap<String, Vec<u8>> =
                std::collections::HashMap::new();
            let mut cron_state: std::collections::HashMap<String, CronEntry> =
                std::collections::HashMap::new();
            // Live blob-change watchers, keyed by `<function>|<trigger id>`; each
            // owns a native watch stream + its drain task (FA-5).
            let mut blob_watchers: std::collections::HashMap<String, tokio::task::JoinHandle<()>> =
                std::collections::HashMap::new();
            let mut interval = tokio::time::interval(SCHEDULER_TICK);
            loop {
                interval.tick().await;
                // Spawned cron fires are detached (bounded by the invocation
                // timeout); their handles are dropped here.
                if let Err(err) = run_scheduler_tick(
                    &inner,
                    &deploy,
                    &mut wasm_cache,
                    &mut cron_state,
                    CronNow::now(),
                )
                .await
                {
                    tracing::warn!(%err, "scheduler tick failed");
                }
                // Reconcile blob-change watchers (FA-5): spawn one per live
                // `Blob` trigger, drop those whose trigger is gone.
                reconcile_blob_watchers(&inner, &deploy, &mut blob_watchers).await;
            }
        }))
    }
}

/// Reconcile the live blob-change watchers against the stored `Blob` triggers:
/// spawn a watcher for each new trigger, abort + drop watchers whose trigger was
/// removed. Leader-gated (shared-FS clusters would otherwise fire per-node).
#[cfg(feature = "handlers")]
async fn reconcile_blob_watchers(
    inner: &Arc<HandlerRuntimeInner>,
    deploy: &DeployStore,
    watchers: &mut std::collections::HashMap<String, tokio::task::JoinHandle<()>>,
) {
    use boatramp_core::function::TriggerKind;
    // Only the leader (or a single node) dispatches, matching cron/invoke.
    if inner.cron_leader_gate.get().is_some_and(|gate| !gate()) {
        for (_, handle) in watchers.drain() {
            handle.abort();
        }
        return;
    }
    // Reconcile every project's blob triggers; a same-named function in two
    // projects gets two independent watchers (the watch id is project-qualified).
    let projects = deploy.discover_projects().await.unwrap_or_default();
    let mut desired = std::collections::HashSet::new();
    for project_name in projects {
        let project = ProjectRef::new(&project_name);
        let functions = match deploy.list_stored_functions(project).await {
            Ok(f) => f,
            Err(_) => continue,
        };
        for function in functions {
            let triggers = deploy
                .list_triggers(project, &function.name)
                .await
                .unwrap_or_default();
            for trigger in triggers {
                let TriggerKind::Blob { prefix } = &trigger.kind else {
                    continue;
                };
                let watch_id = format!("{project_name}|{}|{}", function.name, trigger.id);
                desired.insert(watch_id.clone());
                if let std::collections::hash_map::Entry::Vacant(slot) = watchers.entry(watch_id) {
                    if let Some(handle) = spawn_blob_watcher(
                        inner.clone(),
                        deploy.clone(),
                        project_name.clone(),
                        function.clone(),
                        prefix,
                    )
                    .await
                    {
                        slot.insert(handle);
                    }
                }
            }
        }
    }
    // Drop watchers whose trigger no longer exists.
    watchers.retain(|id, handle| {
        if desired.contains(id) {
            true
        } else {
            handle.abort();
            false
        }
    });
}

/// Spawn a task that watches the function's blob prefix and enqueues an async
/// invocation on each change. Returns `None` if the backend can't watch (the
/// activation gate should already have refused, so this is defensive).
#[cfg(feature = "handlers")]
pub(super) async fn spawn_blob_watcher(
    inner: Arc<HandlerRuntimeInner>,
    deploy: DeployStore,
    project: String,
    function: boatramp_core::function::Function,
    prefix: &str,
) -> Option<tokio::task::JoinHandle<()>> {
    // The function's blobstore lives under its **project-qualified** namespace
    // (`hblob/fn/<name>/` for `default`, `hblob/{project}/fn/<name>/` otherwise);
    // the trigger prefix is relative to it (same key the provisioner + ledger
    // use, so a watcher and the ledger/provisioner agree).
    let project_ref = ProjectRef::new(&project);
    let storage_prefix = blob_storage_prefix(project_ref, &function.name, prefix);
    let mut stream = match inner.storage.watch(&storage_prefix).await {
        Ok(Some(stream)) => stream,
        Ok(None) => return None,
        Err(err) => {
            tracing::warn!(function = %function.name, %err, "starting blob watch failed");
            return None;
        }
    };
    let namespace = format!(
        "hblob/{}/",
        project_ref.qualified(&format!("fn/{}", function.name))
    );
    Some(tokio::spawn(async move {
        use futures::StreamExt;
        while let Some(change) = stream.next().await {
            enqueue_blob_invocation(
                &deploy,
                ProjectRef::new(&project),
                &function,
                &change,
                &namespace,
            )
            .await;
        }
    }))
}

/// Enqueue a durable async invocation for a blob change, with the changed key +
/// kind as the JSON request body (the function-relative key, `hblob/fn/<name>/`
/// stripped).
#[cfg(feature = "handlers")]
async fn enqueue_blob_invocation(
    deploy: &DeployStore,
    project: ProjectRef<'_>,
    function: &boatramp_core::function::Function,
    change: &boatramp_core::BlobChange,
    namespace: &str,
) {
    use boatramp_core::BlobChangeKind;
    let key = change.key.strip_prefix(namespace).unwrap_or(&change.key);
    let kind = match change.kind {
        BlobChangeKind::Created => "created",
        BlobChangeKind::Modified => "modified",
        BlobChangeKind::Removed => "removed",
    };
    let body = serde_json::json!({ "key": key, "kind": kind });
    let payload = serde_json::to_vec(&body).unwrap_or_default();
    let now = now_unix();
    let inv = boatramp_core::function::Invocation {
        id: new_invocation_id(),
        function: function.name.clone(),
        version: function.active.clone(),
        mode: boatramp_core::function::InvokeMode::Async,
        status: boatramp_core::function::InvocationStatus::Queued,
        idempotency_key: None,
        attempts: 0,
        request_b64: (!payload.is_empty()).then(|| b64_encode(&payload)),
        request_content_type: Some("application/json".to_string()),
        result: None,
        created: now,
        updated: now,
    };
    if let Err(err) = deploy.put_invocation(project, &inv).await {
        tracing::warn!(function = %function.name, %err, "enqueuing blob invocation failed");
    }
}

/// One scheduler pass: for every site, drive the consumers and crons of its
/// active deployments. Consumers are processed inline (claim+dispatch); crons
/// that are due are fired as detached tasks (loopback dispatch). Returns the
/// number of messages acked and the spawned cron-fire handles (for tests).
#[cfg(feature = "handlers")]
pub(super) async fn run_scheduler_tick(
    inner: &Arc<HandlerRuntimeInner>,
    deploy: &DeployStore,
    wasm_cache: &mut std::collections::HashMap<String, Vec<u8>>,
    cron_state: &mut std::collections::HashMap<String, CronEntry>,
    now: CronNow,
) -> Result<(usize, Vec<tokio::task::JoinHandle<()>>), DeployError> {
    use std::sync::atomic::Ordering;
    let mut acked = 0;
    let mut cron_handles = Vec::new();
    // Fan out over every project: a same-named site/function/workflow in two
    // projects is scheduled independently (its background jobs run under its own
    // tenant, never `default`). For a single default-only store this is one pass.
    let projects = deploy.discover_projects().await?;
    for project_name in &projects {
        let project = ProjectRef::new(project_name);
        for site in deploy.list_sites(project).await? {
            let Some(site_config) = deploy.get_site_config(project, &site).await? else {
                continue;
            };
            let Some(site_handlers) = site_config.handlers.as_ref().filter(|h| h.enabled) else {
                continue;
            };
            // Active deployments: the current one (production, namespace `{site}`)
            // plus each background alias (namespace `{site}/{alias}`). Never previews.
            let mut active: Vec<(String, String)> = Vec::new();
            if let Some(id) = deploy.current_id(project, &site).await? {
                active.push((id, site.clone()));
            }
            for alias in &site_handlers.background_aliases {
                if let Some(id) = deploy.get_alias(project, &site, alias).await? {
                    active.push((id, format!("{site}/{alias}")));
                }
            }
            for (deploy_id, scope) in active {
                let Some(manifest) = deploy.get_manifest(&deploy_id).await? else {
                    continue;
                };
                // --- consumers (only with a messaging backend) ---
                if let Some(messaging) = inner.messaging.clone() {
                    for consumer in &manifest.config.consumers {
                        let Some(entry) = manifest.files.get(&consumer.component) else {
                            tracing::warn!(site, component = %consumer.component, "consumer component missing");
                            continue;
                        };
                        // Cache the (content-addressed) component bytes by hash.
                        if !wasm_cache.contains_key(&entry.hash) {
                            match read_blob_bytes(deploy, &entry.hash).await {
                                Ok(bytes) => {
                                    wasm_cache.insert(entry.hash.clone(), bytes);
                                }
                                Err(err) => {
                                    tracing::warn!(site, %err, "reading consumer component failed");
                                    continue;
                                }
                            }
                        }
                        let wasm = &wasm_cache[&entry.hash];
                        let bindings = build_bindings(
                            inner,
                            project,
                            &site,
                            &scope,
                            None,
                            &consumer.imports,
                            site_handlers,
                            // Consumers have no deploy `env`; site secrets still apply.
                            &std::collections::BTreeMap::new(),
                            // Consumers do not get the invoke capability (no allowlist field).
                            &[],
                            0,
                        )
                        .await;
                        acked += dispatch_consumer_batch(
                            &inner.engine,
                            messaging.as_ref(),
                            &inner.metrics,
                            &site,
                            &format!("{scope}/{}", consumer.topic),
                            &format!("{scope}/"),
                            &entry.hash,
                            wasm,
                            &bindings,
                            site_limits(site_handlers),
                            CONSUMER_LEASE,
                            CONSUMER_MAX_ATTEMPTS,
                            CONSUMER_BATCH,
                        )
                        .await;
                    }
                }
                // --- crons (leader-only in cluster mode) ---
                // The gate fires crons on exactly one node; consumers above run on
                // every node (leased dispatch distributes them). `None` = single
                // node, always fires.
                let cron_enabled = inner.cron_leader_gate.get().is_none_or(|gate| gate());
                for (idx, cron) in manifest.config.crons.iter().enumerate() {
                    if !cron_enabled {
                        break;
                    }
                    let Ok(schedule) = boatramp_core::cron::CronSchedule::parse(&cron.schedule)
                    else {
                        continue;
                    };
                    if !schedule.fires_at(now.minute, now.hour, now.dom, now.month, now.dow) {
                        continue;
                    }
                    let key = format!("{project_name}|{scope}|cron|{idx}");
                    let entry = cron_state.entry(key).or_insert_with(|| CronEntry {
                        last_minute: -1,
                        running: Arc::new(std::sync::atomic::AtomicBool::new(false)),
                    });
                    if entry.last_minute == now.minute_stamp {
                        continue; // already fired this minute
                    }
                    if matches!(cron.overlap, boatramp_core::config::Overlap::Skip)
                        && entry.running.load(Ordering::Acquire)
                    {
                        tracing::info!(site, route = %cron.route, "cron skipped (previous run still in flight)");
                        continue;
                    }
                    entry.last_minute = now.minute_stamp;
                    let running = entry.running.clone();
                    running.store(true, Ordering::Release);
                    let (inner, deploy, manifest, project_owned, site, scope, site_handlers, cron) = (
                        inner.clone(),
                        deploy.clone(),
                        manifest.clone(),
                        project_name.clone(),
                        site.clone(),
                        scope.clone(),
                        site_handlers.clone(),
                        cron.clone(),
                    );
                    cron_handles.push(tokio::spawn(async move {
                        fire_cron(
                            &inner,
                            &deploy,
                            ProjectRef::new(&project_owned),
                            &manifest,
                            &site,
                            &scope,
                            &site_handlers,
                            &cron,
                        )
                        .await;
                        running.store(false, Ordering::Release);
                    }));
                }
            }
        }
    }
    // --- async function invocations (FA-3) ---
    // Drain each top-level function's queued invocations. Leader-gated like crons
    // (`None` = single node) so a durable async call runs exactly once
    // cluster-wide; the drain runs inline so the outcome is settled this tick.
    let invoke_enabled = inner.cron_leader_gate.get().is_none_or(|gate| gate());
    if invoke_enabled {
        // Same per-project fan-out as the site loop: each project's functions +
        // workflows drain under their own tenant.
        for project_name in &projects {
            let project = ProjectRef::new(project_name);
            for function in deploy.list_stored_functions(project).await? {
                // Fire due triggers first (a cron enqueues an invocation this
                // tick), then drain the queue so a just-enqueued call runs without
                // waiting.
                dispatch_function_triggers(inner, deploy, project, &function, &now).await;
                drain_function_invocations(inner, deploy, project, &function).await;
            }
            // --- workflow runs (FA-6), same leader gate ---
            for workflow in deploy.list_workflows(project).await? {
                drain_workflow_runs(inner, deploy, project, &workflow).await;
            }
        }
    }
    Ok((acked, cron_handles))
}

/// Fire one cron: dispatch the declared handler route in-process (loopback,
/// never a network hop) with a synthetic `GET`, scoped to the deployment's
/// namespace. The response is drained and discarded — a cron has no caller.
#[cfg(feature = "handlers")]
#[allow(clippy::too_many_arguments)]
async fn fire_cron(
    inner: &HandlerRuntimeInner,
    deploy: &DeployStore,
    project: ProjectRef<'_>,
    manifest: &Manifest,
    site: &str,
    scope: &str,
    site_handlers: &boatramp_core::config::HandlersSiteConfig,
    cron: &boatramp_core::config::CronConfig,
) {
    let Some(handler) = route::match_handler(&manifest.config.handlers, "GET", &cron.route) else {
        tracing::warn!(site, route = %cron.route, "cron route matches no GET handler");
        return;
    };
    let Some(entry) = manifest.files.get(&handler.component) else {
        return;
    };
    let wasm = match read_blob_bytes(deploy, &entry.hash).await {
        Ok(wasm) => wasm,
        Err(err) => {
            tracing::warn!(site, %err, "reading cron handler component failed");
            return;
        }
    };
    let bindings = build_bindings(
        inner,
        project,
        site,
        scope,
        None,
        &handler.imports,
        site_handlers,
        &handler.env,
        // A cron-triggered handler is also the root of a call chain (depth 0); its
        // invoke allowlist applies the same as on the HTTP path.
        &handler.invoke_targets,
        0,
    )
    .await;
    let limits = effective_limits(site_handlers, handler);
    let request = match axum::http::Request::builder()
        .method("GET")
        .uri(format!("http://localhost{}", cron.route))
        .header("x-boatramp-trigger", "cron")
        .body(boatramp_handlers::empty_body())
    {
        Ok(request) => request,
        Err(_) => return,
    };
    let start = std::time::Instant::now();
    let result = inner
        .engine
        .serve_with_limits(&entry.hash, &wasm, request, bindings, limits)
        .await;
    inner.metrics.observe(
        site,
        metrics::Trigger::Cron,
        &cron.route,
        &entry.hash,
        metrics::Outcome::from_result(&result),
        start.elapsed(),
    );
    match result {
        Ok(response) => {
            // Drive the (possibly streamed) body to completion so the guest's
            // side effects finish, then discard it.
            let _ = http_body_util::BodyExt::collect(response.into_body()).await;
            tracing::info!(site, route = %cron.route, "cron fired");
        }
        Err(err) => tracing::warn!(site, route = %cron.route, %err, "cron invocation failed"),
    }
}

/// `503` for a handler that cannot run (e.g. its component is missing).
#[cfg(feature = "handlers")]
pub(super) fn handler_unavailable() -> Response {
    (StatusCode::SERVICE_UNAVAILABLE, "handler unavailable\n").into_response()
}

/// The per-invocation limits for a handler: the site's caps and any per-handler
/// caps (the lower of the two for each dimension). Left at the engine default
/// where neither is set; the engine then clamps to its own ceiling.
#[cfg(feature = "handlers")]
pub(super) fn effective_limits(
    site_handlers: &boatramp_core::config::HandlersSiteConfig,
    handler: &boatramp_core::config::HandlerConfig,
) -> boatramp_handlers::Limits {
    let mut limits = boatramp_handlers::Limits::default();
    let handler_limits = handler.limits.as_ref();
    if let Some(mb) = [
        site_handlers.max_memory_mb,
        handler_limits.and_then(|l| l.memory_mb),
    ]
    .into_iter()
    .flatten()
    .min()
    {
        limits.memory_bytes = (mb as usize).saturating_mul(1024 * 1024);
    }
    if let Some(ms) = [
        site_handlers.max_timeout_ms,
        handler_limits.and_then(|l| l.timeout_ms),
    ]
    .into_iter()
    .flatten()
    .min()
    {
        limits.timeout_ms = ms as u64;
    }
    // CPU fuel cap: the smaller of the site ceiling and any per-handler budget
    // (a handler may only lower it). Absent on both → unmetered.
    limits.fuel = [site_handlers.max_fuel, handler_limits.and_then(|l| l.fuel)]
        .into_iter()
        .flatten()
        .min();
    limits
}

/// Acquire a permit from the site's concurrency semaphore (created on first use)
/// when the site sets `maxConcurrency`; `Ok(None)` if uncapped, `Err(())` when
/// the site is at its limit (the caller turns that into a 503).
#[cfg(feature = "handlers")]
pub(super) fn acquire_site_permit(
    inner: &HandlerRuntimeInner,
    site: &str,
    site_handlers: &boatramp_core::config::HandlersSiteConfig,
) -> Result<Option<tokio::sync::OwnedSemaphorePermit>, ()> {
    let Some(max) = site_handlers.max_concurrency else {
        return Ok(None);
    };
    let semaphore = {
        let mut map = inner.site_semaphores.lock().unwrap();
        map.entry(site.to_string())
            .or_insert_with(|| Arc::new(tokio::sync::Semaphore::new(max as usize)))
            .clone()
    };
    semaphore.try_acquire_owned().map(Some).map_err(|_| ())
}

/// Map a handler engine error to an HTTP status.
#[cfg(feature = "handlers")]
pub(super) fn handler_error_response(err: &boatramp_handlers::HandlerError) -> Response {
    use boatramp_handlers::HandlerError;
    let (status, body) = match err {
        HandlerError::Timeout => (StatusCode::GATEWAY_TIMEOUT, "handler timed out\n"),
        HandlerError::OutOfFuel => (
            StatusCode::GATEWAY_TIMEOUT,
            "handler exhausted its CPU budget\n",
        ),
        HandlerError::Overloaded => (
            StatusCode::SERVICE_UNAVAILABLE,
            "handler engine at capacity\n",
        ),
        HandlerError::Compile(_) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            "handler failed to compile\n",
        ),
        HandlerError::Trap(_) | HandlerError::NoResponse | HandlerError::Internal(_) => {
            (StatusCode::INTERNAL_SERVER_ERROR, "handler error\n")
        }
    };
    (status, body).into_response()
}