PiCloud/crates/manager-core/src/repo.rs

use std::collections::BTreeMap;

use async_trait::async_trait;
use picloud_orchestrator_core::{ResolverError, ScriptResolver};
use picloud_shared::{
    AdminUserId, AppId, ExecutionLog, ExecutionSource, ExecutionStatus, GroupId, RequestId, Script,
    ScriptId, ScriptKind, ScriptSandbox,
};
use sqlx::PgPool;

#[derive(Debug, thiserror::Error)]
pub enum ScriptRepositoryError {
    #[error("database error: {0}")]
    Db(#[from] sqlx::Error),

    #[error("not found: {0}")]
    NotFound(ScriptId),

    #[error("conflict: {0}")]
    Conflict(String),
}

/// CRUD over the `scripts` table.
#[async_trait]
pub trait ScriptRepository: Send + Sync {
    async fn get(&self, id: ScriptId) -> Result<Option<Script>, ScriptRepositoryError>;
    /// v1.1.9. Resolve `(app_id, name) → Script`. Backs `invoke("name", …)`
    /// — the SDK lets a caller name a target script by its `scripts.name`
    /// instead of plumbing UUIDs. Returns `None` if no script in the app
    /// has that name. Names are unique per `(app_id, name)` already
    /// (existing migration constraint).
    async fn get_by_name(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Script>, ScriptRepositoryError>;
    /// Phase 4: resolve `name → Script` down `app_id`'s ownership chain —
    /// the app itself, then its ancestor groups nearest-first. Nearest owner
    /// wins (an app's own script shadows an inherited group one of the same
    /// name: CoW). Returns `None` if no script in the chain has that name.
    /// Backs inherited `invoke("name")` and declarative name→id binding.
    async fn get_by_name_inherited(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Script>, ScriptRepositoryError>;
    /// Phase 4: is `script_id` invocable in `app_id`'s context? True iff the
    /// script is owned by `app_id` directly, OR owned by a group on `app_id`'s
    /// ancestor chain. The cross-app isolation predicate generalized to the
    /// hierarchy — the runtime backstop for trigger/queue/invoke dispatch and
    /// trigger-bind validation. A missing script returns `false` (fail closed).
    async fn is_invocable_by_app(
        &self,
        script_id: ScriptId,
        app_id: AppId,
    ) -> Result<bool, ScriptRepositoryError>;
    /// Every script across all apps. Mostly for tests and admin
    /// "global" views; the dashboard reaches scripts via `list_for_app`.
    async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError>;
    async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Script>, ScriptRepositoryError>;
    /// Phase 4: every script owned directly by `group_id` (not inherited —
    /// just the group's own rows). Backs the group-script admin list.
    async fn list_for_group(&self, group_id: GroupId)
        -> Result<Vec<Script>, ScriptRepositoryError>;
    /// Every script in any app the user is a member of. Drives
    /// `GET /admin/scripts` for `member` instance-role callers so the
    /// API never returns scripts they shouldn't see — even before the
    /// per-handler capability check fires.
    async fn list_for_user(
        &self,
        user_id: AdminUserId,
    ) -> Result<Vec<Script>, ScriptRepositoryError>;
    async fn create(&self, input: NewScript) -> Result<Script, ScriptRepositoryError>;
    async fn update(
        &self,
        id: ScriptId,
        patch: ScriptPatch,
    ) -> Result<Script, ScriptRepositoryError>;
    async fn delete(&self, id: ScriptId) -> Result<(), ScriptRepositoryError>;

    /// v1.1.3: how many routes reference this script. Used by the
    /// API layer to refuse `endpoint → module` kind changes when the
    /// script is still bound to user-facing entry points.
    async fn count_routes_for_script(
        &self,
        script_id: ScriptId,
    ) -> Result<i64, ScriptRepositoryError>;

    /// v1.1.3: how many triggers (kv / docs / dead-letter) target
    /// this script. Same purpose as `count_routes_for_script`.
    async fn count_triggers_for_script(
        &self,
        script_id: ScriptId,
    ) -> Result<i64, ScriptRepositoryError>;

    /// v1.1.3: list module dependencies of this script — the rows in
    /// `script_imports` where `importer_script_id = script_id`. Used
    /// by tests and (eventually) a dashboard "Imports" panel.
    async fn list_imports(&self, script_id: ScriptId)
        -> Result<Vec<Script>, ScriptRepositoryError>;
}

/// F-Q-011: blanket impl for `Arc<T: ScriptRepository>`. Lets callers
/// pass a shared `Arc<PostgresScriptRepository>` directly anywhere a
/// `dyn ScriptRepository` is wanted — replacing the hand-delegated
/// `PostgresScriptRepoHandle` newtype that previously had to be
/// rewritten in lockstep every time the trait gained a method.
#[async_trait::async_trait]
impl<T: ScriptRepository + ?Sized> ScriptRepository for std::sync::Arc<T> {
    async fn get(&self, id: ScriptId) -> Result<Option<Script>, ScriptRepositoryError> {
        (**self).get(id).await
    }
    async fn get_by_name(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Script>, ScriptRepositoryError> {
        (**self).get_by_name(app_id, name).await
    }
    async fn get_by_name_inherited(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Script>, ScriptRepositoryError> {
        (**self).get_by_name_inherited(app_id, name).await
    }
    async fn is_invocable_by_app(
        &self,
        script_id: ScriptId,
        app_id: AppId,
    ) -> Result<bool, ScriptRepositoryError> {
        (**self).is_invocable_by_app(script_id, app_id).await
    }
    async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
        (**self).list().await
    }
    async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Script>, ScriptRepositoryError> {
        (**self).list_for_app(app_id).await
    }
    async fn list_for_group(
        &self,
        group_id: GroupId,
    ) -> Result<Vec<Script>, ScriptRepositoryError> {
        (**self).list_for_group(group_id).await
    }
    async fn list_for_user(
        &self,
        user_id: AdminUserId,
    ) -> Result<Vec<Script>, ScriptRepositoryError> {
        (**self).list_for_user(user_id).await
    }
    async fn create(&self, input: NewScript) -> Result<Script, ScriptRepositoryError> {
        (**self).create(input).await
    }
    async fn update(
        &self,
        id: ScriptId,
        patch: ScriptPatch,
    ) -> Result<Script, ScriptRepositoryError> {
        (**self).update(id, patch).await
    }
    async fn delete(&self, id: ScriptId) -> Result<(), ScriptRepositoryError> {
        (**self).delete(id).await
    }
    async fn count_routes_for_script(
        &self,
        script_id: ScriptId,
    ) -> Result<i64, ScriptRepositoryError> {
        (**self).count_routes_for_script(script_id).await
    }
    async fn count_triggers_for_script(
        &self,
        script_id: ScriptId,
    ) -> Result<i64, ScriptRepositoryError> {
        (**self).count_triggers_for_script(script_id).await
    }
    async fn list_imports(
        &self,
        script_id: ScriptId,
    ) -> Result<Vec<Script>, ScriptRepositoryError> {
        (**self).list_imports(script_id).await
    }
}

/// Inbound shape for create. Defaults match the migration's CHECK
/// constraints; the repo enforces them in the DB regardless.
#[derive(Debug, Clone)]
pub struct NewScript {
    /// App owner (the common case). Exactly one of `app_id`/`group_id` must
    /// be `Some` — the DB CHECK is the backstop, but callers should uphold
    /// it. App-owned creation continues to pass `Some(app_id)`, `group_id:
    /// None`; group-owned creation (Phase 4) inverts that.
    pub app_id: Option<AppId>,
    /// Group owner (Phase 4). See [`NewScript::app_id`].
    pub group_id: Option<GroupId>,
    pub name: String,
    pub description: Option<String>,
    pub source: String,
    /// Defaults to `Endpoint` if absent. `Module` scripts cannot be
    /// bound to routes or used as trigger targets.
    pub kind: ScriptKind,
    pub timeout_seconds: Option<i32>,
    pub memory_limit_mb: Option<i32>,
    /// Sandbox overrides; `None` means store an empty object (use
    /// platform defaults at exec time).
    pub sandbox: Option<ScriptSandbox>,
    /// Three-state lifecycle (§4.3). Create active by default.
    pub enabled: bool,
    /// v1.1.3: literal-path `import "<name>"` declarations extracted
    /// from the source. The repo writes these into `script_imports`
    /// transactionally with the script row. Empty when validation
    /// found no imports (the common case for endpoints today).
    pub imports: Vec<String>,
}

/// Inbound shape for update. `None` fields are left untouched.
#[derive(Debug, Clone, Default)]
pub struct ScriptPatch {
    pub name: Option<String>,
    pub description: Option<Option<String>>,
    pub source: Option<String>,
    pub timeout_seconds: Option<i32>,
    pub memory_limit_mb: Option<i32>,
    /// `Some(sandbox)` replaces the stored overrides wholesale (including
    /// `Some(empty)` to clear them); `None` leaves them untouched.
    pub sandbox: Option<ScriptSandbox>,
    /// `Some(new_kind)` changes the script's role; the API layer
    /// rejects unsafe transitions (e.g. endpoint→module when routes
    /// or triggers reference the script).
    pub kind: Option<ScriptKind>,
    /// `Some(v)` toggles the three-state lifecycle flag; `None` leaves it.
    pub enabled: Option<bool>,
    /// v1.1.3: when `source` is also `Some`, the repo replaces the
    /// `script_imports` edges for this script with these names.
    /// `None` keeps the existing edges untouched (a name/description
    /// edit alone shouldn't touch the dep graph).
    pub imports: Option<Vec<String>>,
}

pub struct PostgresScriptRepository {
    pool: PgPool,
}

impl PostgresScriptRepository {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    #[must_use]
    pub fn pool(&self) -> &PgPool {
        &self.pool
    }
}

/// Columns selected from `scripts` everywhere — kept in one constant so
/// adding `kind` (v1.1.3) and future columns can't accidentally skip
/// one query.
const SCRIPT_SELECT_COLS: &str = "id, app_id, group_id, name, description, version, source, kind, \
                                  timeout_seconds, memory_limit_mb, sandbox, enabled, \
                                  created_at, updated_at";

#[async_trait]
impl ScriptRepository for PostgresScriptRepository {
    async fn get(&self, id: ScriptId) -> Result<Option<Script>, ScriptRepositoryError> {
        let row = sqlx::query_as::<_, ScriptRow>(&format!(
            "SELECT {SCRIPT_SELECT_COLS} FROM scripts WHERE id = $1"
        ))
        .bind(id.into_inner())
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(Into::into))
    }

    async fn get_by_name(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Script>, ScriptRepositoryError> {
        let row = sqlx::query_as::<_, ScriptRow>(&format!(
            "SELECT {SCRIPT_SELECT_COLS} FROM scripts WHERE app_id = $1 AND name = $2"
        ))
        .bind(app_id.into_inner())
        .bind(name)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(Into::into))
    }

    async fn get_by_name_inherited(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<Option<Script>, ScriptRepositoryError> {
        // Walk app → ancestor groups (CHAIN_LEVELS_CTE binds $1 = app_id),
        // join scripts owned at each level, and take the nearest (lowest
        // depth) row named `name`. Case-insensitive to match the per-owner
        // `LOWER(name)` unique indexes. Mirrors the vars/secrets resolvers.
        let cols = SCRIPT_SELECT_COLS
            .split(", ")
            .map(|c| format!("s.{c}"))
            .collect::<Vec<_>>()
            .join(", ");
        let row = sqlx::query_as::<_, ScriptRow>(&format!(
            "{cte} SELECT {cols} FROM scripts s \
             JOIN chain c ON (s.app_id = c.app_owner OR s.group_id = c.group_owner) \
             WHERE LOWER(s.name) = LOWER($2) \
             ORDER BY c.depth ASC \
             LIMIT 1",
            cte = crate::config_resolver::CHAIN_LEVELS_CTE,
        ))
        .bind(app_id.into_inner())
        .bind(name)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(Into::into))
    }

    async fn is_invocable_by_app(
        &self,
        script_id: ScriptId,
        app_id: AppId,
    ) -> Result<bool, ScriptRepositoryError> {
        // EXISTS over the same chain: the script's owner (app or group) must
        // appear on `app_id`'s app→ancestor-group chain. CHAIN_LEVELS_CTE
        // binds $1 = app_id; $2 = script_id.
        let exists: (bool,) = sqlx::query_as(&format!(
            "{cte} SELECT EXISTS ( \
                 SELECT 1 FROM scripts s \
                 JOIN chain c ON (s.app_id = c.app_owner OR s.group_id = c.group_owner) \
                 WHERE s.id = $2 \
             )",
            cte = crate::config_resolver::CHAIN_LEVELS_CTE,
        ))
        .bind(app_id.into_inner())
        .bind(script_id.into_inner())
        .fetch_one(&self.pool)
        .await?;
        Ok(exists.0)
    }

    async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
        let rows = sqlx::query_as::<_, ScriptRow>(&format!(
            "SELECT {SCRIPT_SELECT_COLS} FROM scripts ORDER BY name"
        ))
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Script>, ScriptRepositoryError> {
        let rows = sqlx::query_as::<_, ScriptRow>(&format!(
            "SELECT {SCRIPT_SELECT_COLS} FROM scripts WHERE app_id = $1 ORDER BY name"
        ))
        .bind(app_id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn list_for_group(
        &self,
        group_id: GroupId,
    ) -> Result<Vec<Script>, ScriptRepositoryError> {
        let rows = sqlx::query_as::<_, ScriptRow>(&format!(
            "SELECT {SCRIPT_SELECT_COLS} FROM scripts WHERE group_id = $1 ORDER BY name"
        ))
        .bind(group_id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn list_for_user(
        &self,
        user_id: AdminUserId,
    ) -> Result<Vec<Script>, ScriptRepositoryError> {
        let cols = SCRIPT_SELECT_COLS
            .split(", ")
            .map(|c| format!("s.{c}"))
            .collect::<Vec<_>>()
            .join(", ");
        let rows = sqlx::query_as::<_, ScriptRow>(&format!(
            "SELECT {cols} FROM scripts s \
             JOIN app_members m ON m.app_id = s.app_id \
             WHERE m.user_id = $1 \
             ORDER BY s.name"
        ))
        .bind(user_id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn create(&self, input: NewScript) -> Result<Script, ScriptRepositoryError> {
        let mut tx = self.pool.begin().await?;
        let script = insert_script_tx(&mut tx, &input).await?;
        tx.commit().await?;
        Ok(script)
    }

    async fn update(
        &self,
        id: ScriptId,
        patch: ScriptPatch,
    ) -> Result<Script, ScriptRepositoryError> {
        let mut tx = self.pool.begin().await?;
        let script = update_script_tx(&mut tx, id, &patch).await?;
        tx.commit().await?;
        Ok(script)
    }

    async fn delete(&self, id: ScriptId) -> Result<(), ScriptRepositoryError> {
        let res = sqlx::query("DELETE FROM scripts WHERE id = $1")
            .bind(id.into_inner())
            .execute(&self.pool)
            .await?;
        if res.rows_affected() == 0 {
            return Err(ScriptRepositoryError::NotFound(id));
        }
        Ok(())
    }

    async fn count_routes_for_script(
        &self,
        script_id: ScriptId,
    ) -> Result<i64, ScriptRepositoryError> {
        let n: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM routes WHERE script_id = $1")
            .bind(script_id.into_inner())
            .fetch_one(&self.pool)
            .await?;
        Ok(n.0)
    }

    async fn count_triggers_for_script(
        &self,
        script_id: ScriptId,
    ) -> Result<i64, ScriptRepositoryError> {
        let n: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM triggers WHERE script_id = $1")
            .bind(script_id.into_inner())
            .fetch_one(&self.pool)
            .await?;
        Ok(n.0)
    }

    async fn list_imports(
        &self,
        script_id: ScriptId,
    ) -> Result<Vec<Script>, ScriptRepositoryError> {
        let cols = SCRIPT_SELECT_COLS
            .split(", ")
            .map(|c| format!("s.{c}"))
            .collect::<Vec<_>>()
            .join(", ");
        let rows = sqlx::query_as::<_, ScriptRow>(&format!(
            "SELECT {cols} FROM scripts s \
             JOIN script_imports i ON i.imported_script_id = s.id \
             WHERE i.importer_script_id = $1 \
             ORDER BY s.name"
        ))
        .bind(script_id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(Into::into).collect())
    }
}

/// Replace the `script_imports` edges for `importer` with rows derived
/// from `import_names`. Names that don't resolve to a `kind = 'module'`
/// script in the same app are silently skipped (best-effort dep graph).
async fn replace_imports_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    importer: ScriptId,
    app_id: AppId,
    import_names: &[String],
) -> Result<(), ScriptRepositoryError> {
    sqlx::query("DELETE FROM script_imports WHERE importer_script_id = $1")
        .bind(importer.into_inner())
        .execute(&mut **tx)
        .await?;
    if import_names.is_empty() {
        return Ok(());
    }
    // Insert with ON CONFLICT DO NOTHING in case the source declares
    // `import "x"` twice — the dep graph stores each pair at most once.
    sqlx::query(
        "INSERT INTO script_imports (app_id, importer_script_id, imported_script_id) \
         SELECT $1, $2, s.id \
         FROM scripts s \
         WHERE s.app_id = $1 \
           AND s.kind = 'module' \
           AND s.id <> $2 \
           AND s.name = ANY($3) \
         ON CONFLICT DO NOTHING",
    )
    .bind(app_id.into_inner())
    .bind(importer.into_inner())
    .bind(import_names)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// Insert a script within an existing transaction — the declarative
/// `apply` engine composes scripts + routes + triggers into one tx.
/// Mirrors `create` minus the `begin`/`commit`.
pub(crate) async fn insert_script_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    input: &NewScript,
) -> Result<Script, ScriptRepositoryError> {
    let sandbox_json = serde_json::to_value(input.sandbox.unwrap_or_default())
        .unwrap_or_else(|_| serde_json::json!({}));
    let res = sqlx::query_as::<_, ScriptRow>(&format!(
        "INSERT INTO scripts ( \
            app_id, group_id, name, description, source, kind, \
            timeout_seconds, memory_limit_mb, sandbox, enabled \
         ) VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, 30), COALESCE($8, 256), $9, $10) \
         RETURNING {SCRIPT_SELECT_COLS}"
    ))
    .bind(input.app_id.map(AppId::into_inner))
    .bind(input.group_id.map(GroupId::into_inner))
    .bind(&input.name)
    .bind(input.description.as_deref())
    .bind(&input.source)
    .bind(input.kind.as_str())
    .bind(input.timeout_seconds)
    .bind(input.memory_limit_mb)
    .bind(sandbox_json)
    .bind(input.enabled)
    .fetch_one(&mut **tx)
    .await;
    let script: Script = match res {
        Ok(row) => row.into(),
        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
            return Err(ScriptRepositoryError::Conflict(format!(
                "a script named {:?} already exists in this owner",
                input.name
            )));
        }
        Err(e) => return Err(e.into()),
    };
    // Module imports are resolved within the owning app's script set. Group
    // scripts must be self-contained in Phase 4-lite (the origin-aware import
    // resolver is Phase 4b), so only app-owned scripts wire up import edges.
    if let Some(app_id) = script.app_id {
        replace_imports_tx(tx, script.id, app_id, &input.imports).await?;
    }
    Ok(script)
}

/// Update a script within an existing transaction. Mirrors `update`
/// minus the `begin`/`commit`.
pub(crate) async fn update_script_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    id: ScriptId,
    patch: &ScriptPatch,
) -> Result<Script, ScriptRepositoryError> {
    let sandbox_json = patch
        .sandbox
        .as_ref()
        .map(|s| serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({})));
    let res = sqlx::query_as::<_, ScriptRow>(&format!(
        "UPDATE scripts SET \
            name = COALESCE($2, name), \
            description = CASE WHEN $3::bool THEN $4 ELSE description END, \
            source = COALESCE($5, source), \
            timeout_seconds = COALESCE($6, timeout_seconds), \
            memory_limit_mb = COALESCE($7, memory_limit_mb), \
            sandbox = COALESCE($8, sandbox), \
            kind = COALESCE($9, kind), \
            enabled = COALESCE($10, enabled), \
            version = version + 1, \
            updated_at = NOW() \
         WHERE id = $1 \
         RETURNING {SCRIPT_SELECT_COLS}"
    ))
    .bind(id.into_inner())
    .bind(patch.name.as_deref())
    .bind(patch.description.is_some())
    .bind(patch.description.as_ref().and_then(|d| d.as_deref()))
    .bind(patch.source.as_deref())
    .bind(patch.timeout_seconds)
    .bind(patch.memory_limit_mb)
    .bind(sandbox_json)
    .bind(patch.kind.map(ScriptKind::as_str))
    .bind(patch.enabled)
    .fetch_optional(&mut **tx)
    .await;
    let script: Script = match res {
        Ok(Some(row)) => row.into(),
        Ok(None) => return Err(ScriptRepositoryError::NotFound(id)),
        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
            return Err(ScriptRepositoryError::Conflict(
                "a script with that name already exists in this owner".into(),
            ));
        }
        Err(e) => return Err(e.into()),
    };
    if let Some(imports) = patch.imports.as_deref() {
        if let Some(app_id) = script.app_id {
            replace_imports_tx(tx, script.id, app_id, imports).await?;
        }
    }
    Ok(script)
}

/// Delete a script within an existing transaction (its routes/triggers
/// cascade via their FKs). Mirrors `delete` minus the pool.
pub(crate) async fn delete_script_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    id: ScriptId,
) -> Result<(), ScriptRepositoryError> {
    let res = sqlx::query("DELETE FROM scripts WHERE id = $1")
        .bind(id.into_inner())
        .execute(&mut **tx)
        .await?;
    if res.rows_affected() == 0 {
        return Err(ScriptRepositoryError::NotFound(id));
    }
    Ok(())
}

/// Row shape mirroring the `scripts` table for sqlx FromRow.
#[derive(sqlx::FromRow)]
struct ScriptRow {
    id: uuid::Uuid,
    /// Polymorphic owner (Phase 4): exactly one of `app_id`/`group_id` is
    /// non-NULL (DB CHECK). App-owned rows keep `app_id` set as before.
    app_id: Option<uuid::Uuid>,
    group_id: Option<uuid::Uuid>,
    name: String,
    description: Option<String>,
    version: i32,
    source: String,
    /// v1.1.3: 'endpoint' | 'module'. Stored as TEXT with a CHECK
    /// constraint so we don't need a Postgres enum (avoiding the
    /// migration churn of adding values later).
    kind: String,
    timeout_seconds: i32,
    memory_limit_mb: i32,
    sandbox: serde_json::Value,
    enabled: bool,
    created_at: chrono::DateTime<chrono::Utc>,
    updated_at: chrono::DateTime<chrono::Utc>,
}

impl From<ScriptRow> for Script {
    fn from(r: ScriptRow) -> Self {
        // Tolerate stale rows whose sandbox column predates a future
        // schema migration: unknown fields are rejected by serde, so
        // fall back to an empty ScriptSandbox rather than poisoning a
        // list response.
        let sandbox = serde_json::from_value(r.sandbox).unwrap_or_default();
        // Defensive: if a row's `kind` somehow falls outside the CHECK
        // constraint, treat it as Endpoint (the safe default — won't
        // grant a row import-target status it doesn't have).
        let kind = ScriptKind::parse_str(&r.kind).unwrap_or(ScriptKind::Endpoint);
        Self {
            id: r.id.into(),
            app_id: r.app_id.map(Into::into),
            group_id: r.group_id.map(Into::into),
            name: r.name,
            description: r.description,
            version: r.version,
            source: r.source,
            kind,
            timeout_seconds: u32::try_from(r.timeout_seconds).unwrap_or(30),
            memory_limit_mb: u32::try_from(r.memory_limit_mb).unwrap_or(256),
            sandbox,
            enabled: r.enabled,
            created_at: r.created_at,
            updated_at: r.updated_at,
        }
    }
}

/// Adapts a `ScriptRepository` into the `ScriptResolver` trait the
/// orchestrator depends on. Keeps orchestrator-core unaware of how
/// scripts are stored.
pub struct RepoResolver<R: ScriptRepository> {
    repo: R,
}

impl<R: ScriptRepository> RepoResolver<R> {
    pub fn new(repo: R) -> Self {
        Self { repo }
    }
}

#[async_trait]
impl<R: ScriptRepository> ScriptResolver for RepoResolver<R> {
    async fn resolve(&self, id: ScriptId) -> Result<Option<Script>, ResolverError> {
        self.repo
            .get(id)
            .await
            .map_err(|e| ResolverError::Backend(e.to_string()))
    }
}

// ----------------------------------------------------------------------------
// Execution log repository (read side)
// ----------------------------------------------------------------------------

/// Read-side access to the `execution_logs` table. Writes go through
/// `PostgresExecutionLogSink` so the read and write paths can diverge
/// in cluster mode without disturbing this trait.
/// Opaque cursor encoding `(created_at, id)` for keyset pagination.
/// Format: `<rfc3339>_<uuid>` — readable in URLs and stable across
/// server restarts (no opaque internal IDs).
#[derive(Debug, Clone)]
pub struct ExecutionLogCursor {
    pub created_at: chrono::DateTime<chrono::Utc>,
    pub id: uuid::Uuid,
}

impl ExecutionLogCursor {
    /// Encode as `<rfc3339>_<uuid>`.
    #[must_use]
    pub fn encode(&self) -> String {
        format!("{}_{}", self.created_at.to_rfc3339(), self.id)
    }

    /// Decode from the wire format. Returns `None` on any parse error;
    /// the caller treats that as "no cursor" (start of the page).
    #[must_use]
    pub fn decode(s: &str) -> Option<Self> {
        let (ts, id) = s.rsplit_once('_')?;
        let created_at = chrono::DateTime::parse_from_rfc3339(ts).ok()?.to_utc();
        let id = uuid::Uuid::parse_str(id).ok()?;
        Some(Self { created_at, id })
    }
}

/// One `(key, count)` row of a metrics breakdown (by status or by source).
#[derive(Debug, Clone, serde::Serialize)]
pub struct MetricsCount {
    pub key: String,
    pub count: i64,
}

/// One hour-bucket of the metrics time series.
#[derive(Debug, Clone, serde::Serialize)]
pub struct MetricsBucket {
    pub ts: chrono::DateTime<chrono::Utc>,
    pub total: i64,
    pub errors: i64,
}

/// Aggregate execution metrics for one app over a trailing window — the read
/// model behind the observability dashboard. Sourced entirely from
/// `execution_logs` (no hot-path instrumentation): counts + error rate +
/// latency percentiles + an hourly series.
#[derive(Debug, Clone, serde::Serialize)]
pub struct MetricsSummary {
    pub window_hours: i32,
    pub total: i64,
    pub errors: i64,
    /// Errors / total in `[0, 1]` (0 when there are no executions).
    pub error_rate: f64,
    pub avg_duration_ms: f64,
    pub p50_duration_ms: f64,
    pub p95_duration_ms: f64,
    pub by_status: Vec<MetricsCount>,
    pub by_source: Vec<MetricsCount>,
    pub series: Vec<MetricsBucket>,
}

#[async_trait]
pub trait ExecutionLogRepository: Send + Sync {
    /// F-P-005: keyset cursor on `(created_at, id)` — `cursor` is the
    /// boundary of the previous page (or `None` for the first page).
    /// Avoids the OFFSET tail-latency cliff on deep pagination.
    async fn list_for_script(
        &self,
        script_id: ScriptId,
        limit: i64,
        cursor: Option<ExecutionLogCursor>,
        source: Option<ExecutionSource>,
    ) -> Result<Vec<ExecutionLog>, ScriptRepositoryError>;

    /// A2: aggregate metrics for `app_id` over the trailing `window_hours`.
    /// Read-only rollup over `execution_logs`.
    async fn summarize_for_app(
        &self,
        app_id: AppId,
        window_hours: i32,
    ) -> Result<MetricsSummary, ScriptRepositoryError>;
}

pub struct PostgresExecutionLogRepository {
    pool: PgPool,
}

impl PostgresExecutionLogRepository {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

#[async_trait]
impl ExecutionLogRepository for PostgresExecutionLogRepository {
    async fn list_for_script(
        &self,
        script_id: ScriptId,
        limit: i64,
        cursor: Option<ExecutionLogCursor>,
        source: Option<ExecutionSource>,
    ) -> Result<Vec<ExecutionLog>, ScriptRepositoryError> {
        // The optional `source` filter is folded into one bind via
        // `$N::text IS NULL OR source = $N` so we don't fan out into four
        // query strings. `None` → the predicate is always true (no filter).
        let source = source.map(ExecutionSource::as_str);
        let rows = match cursor {
            Some(c) => {
                sqlx::query_as::<_, ExecutionLogRow>(
                    "SELECT id, app_id, script_id, request_id, \
                            request_path, request_headers, request_body, \
                            response_code, response_body, \
                            logs, duration_ms, status, source, created_at \
                     FROM execution_logs \
                     WHERE script_id = $1 \
                       AND (created_at, id) < ($2, $3) \
                       AND ($4::text IS NULL OR source = $4) \
                     ORDER BY created_at DESC, id DESC \
                     LIMIT $5",
                )
                .bind(script_id.into_inner())
                .bind(c.created_at)
                .bind(c.id)
                .bind(source)
                .bind(limit)
                .fetch_all(&self.pool)
                .await?
            }
            None => {
                sqlx::query_as::<_, ExecutionLogRow>(
                    "SELECT id, app_id, script_id, request_id, \
                            request_path, request_headers, request_body, \
                            response_code, response_body, \
                            logs, duration_ms, status, source, created_at \
                     FROM execution_logs \
                     WHERE script_id = $1 \
                       AND ($2::text IS NULL OR source = $2) \
                     ORDER BY created_at DESC, id DESC \
                     LIMIT $3",
                )
                .bind(script_id.into_inner())
                .bind(source)
                .bind(limit)
                .fetch_all(&self.pool)
                .await?
            }
        };

        Ok(rows.into_iter().map(Into::into).collect())
    }

    async fn summarize_for_app(
        &self,
        app_id: AppId,
        window_hours: i32,
    ) -> Result<MetricsSummary, ScriptRepositoryError> {
        // Clamp the window to something sane (1 hour … 90 days) so a crafted
        // `?window=` can't ask for an unbounded scan.
        let window_hours = window_hours.clamp(1, 24 * 90);
        let app = app_id.into_inner();

        // Headline aggregate: totals, error count, latency mean + percentiles.
        // `<> 'success'` counts every non-success terminal state as an error.
        let agg = sqlx::query_as::<_, AggRow>(
            "SELECT count(*)::int8 AS total, \
                    count(*) FILTER (WHERE status <> 'success')::int8 AS errors, \
                    COALESCE(avg(duration_ms), 0)::float8 AS avg_ms, \
                    COALESCE(percentile_cont(0.5) WITHIN GROUP (ORDER BY duration_ms), 0)::float8 AS p50, \
                    COALESCE(percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms), 0)::float8 AS p95 \
             FROM execution_logs \
             WHERE app_id = $1 AND created_at >= now() - ($2 * interval '1 hour')",
        )
        .bind(app)
        .bind(window_hours)
        .fetch_one(&self.pool)
        .await?;

        let by_status = sqlx::query_as::<_, CountRow>(
            "SELECT status AS key, count(*)::int8 AS count \
             FROM execution_logs \
             WHERE app_id = $1 AND created_at >= now() - ($2 * interval '1 hour') \
             GROUP BY status ORDER BY count DESC",
        )
        .bind(app)
        .bind(window_hours)
        .fetch_all(&self.pool)
        .await?;

        let by_source = sqlx::query_as::<_, CountRow>(
            "SELECT source AS key, count(*)::int8 AS count \
             FROM execution_logs \
             WHERE app_id = $1 AND created_at >= now() - ($2 * interval '1 hour') \
             GROUP BY source ORDER BY count DESC",
        )
        .bind(app)
        .bind(window_hours)
        .fetch_all(&self.pool)
        .await?;

        let series = sqlx::query_as::<_, BucketRow>(
            "SELECT date_trunc('hour', created_at) AS ts, count(*)::int8 AS total, \
                    count(*) FILTER (WHERE status <> 'success')::int8 AS errors \
             FROM execution_logs \
             WHERE app_id = $1 AND created_at >= now() - ($2 * interval '1 hour') \
             GROUP BY ts ORDER BY ts",
        )
        .bind(app)
        .bind(window_hours)
        .fetch_all(&self.pool)
        .await?;

        // Counts are execution tallies over a bounded window — always far below
        // f64's 2^53 exact-integer range, so the cast is lossless in practice.
        #[allow(clippy::cast_precision_loss)]
        let error_rate = if agg.total > 0 {
            agg.errors as f64 / agg.total as f64
        } else {
            0.0
        };
        Ok(MetricsSummary {
            window_hours,
            total: agg.total,
            errors: agg.errors,
            error_rate,
            avg_duration_ms: agg.avg_ms,
            p50_duration_ms: agg.p50,
            p95_duration_ms: agg.p95,
            by_status: by_status
                .into_iter()
                .map(|r| MetricsCount {
                    key: r.key,
                    count: r.count,
                })
                .collect(),
            by_source: by_source
                .into_iter()
                .map(|r| MetricsCount {
                    key: r.key,
                    count: r.count,
                })
                .collect(),
            series: series
                .into_iter()
                .map(|r| MetricsBucket {
                    ts: r.ts,
                    total: r.total,
                    errors: r.errors,
                })
                .collect(),
        })
    }
}

#[derive(sqlx::FromRow)]
struct AggRow {
    total: i64,
    errors: i64,
    avg_ms: f64,
    p50: f64,
    p95: f64,
}

#[derive(sqlx::FromRow)]
struct CountRow {
    key: String,
    count: i64,
}

#[derive(sqlx::FromRow)]
struct BucketRow {
    ts: chrono::DateTime<chrono::Utc>,
    total: i64,
    errors: i64,
}

#[derive(sqlx::FromRow)]
struct ExecutionLogRow {
    id: uuid::Uuid,
    app_id: uuid::Uuid,
    script_id: uuid::Uuid,
    request_id: uuid::Uuid,
    request_path: Option<String>,
    request_headers: serde_json::Value,
    request_body: Option<serde_json::Value>,
    response_code: Option<i32>,
    response_body: Option<serde_json::Value>,
    logs: serde_json::Value,
    duration_ms: i32,
    status: String,
    source: String,
    created_at: chrono::DateTime<chrono::Utc>,
}

impl From<ExecutionLogRow> for ExecutionLog {
    fn from(r: ExecutionLogRow) -> Self {
        let headers: BTreeMap<String, String> =
            serde_json::from_value(r.request_headers).unwrap_or_default();
        let status = match r.status.as_str() {
            "success" => ExecutionStatus::Success,
            "timeout" => ExecutionStatus::Timeout,
            "budget_exceeded" => ExecutionStatus::BudgetExceeded,
            _ => ExecutionStatus::Error,
        };
        // Unknown values can't occur (CHECK constraint) but default to
        // Http rather than panicking on a forward-compat surprise.
        let source = ExecutionSource::from_wire(&r.source).unwrap_or_default();
        Self {
            id: r.id,
            app_id: r.app_id.into(),
            script_id: r.script_id.into(),
            request_id: RequestId::from(r.request_id),
            request_path: r.request_path.unwrap_or_default(),
            request_headers: headers,
            request_body: r.request_body.unwrap_or(serde_json::Value::Null),
            response_code: r.response_code.and_then(|c| u16::try_from(c).ok()),
            response_body: r.response_body,
            script_logs: r.logs,
            duration_ms: u64::try_from(r.duration_ms).unwrap_or(0),
            status,
            source,
            created_at: r.created_at,
        }
    }
}