flow_storage/

store.rs

use chrono::{DateTime, Utc};
use parking_lot::Mutex;
use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::Arc;
use thiserror::Error;

#[derive(Debug, Error)]
pub enum StoreError {
    #[error(transparent)]
    Sqlite(#[from] rusqlite::Error),
    #[error(transparent)]
    Json(#[from] serde_json::Error),
    /// Domain-level rejection (e.g. deleting a system collection, deleting a
    /// non-empty collection, renaming a row that does not exist). Distinct
    /// from `Sqlite` so callers can branch on user-visible refusals without
    /// inspecting SQLite error codes.
    #[error("refused: {0}")]
    Refused(String),
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionRecord {
    pub execution_id: String,
    pub status: String,
    pub started_at: DateTime<Utc>,
    pub ended_at: Option<DateTime<Utc>>,
    pub succeeded: u32,
    pub failed: u32,
    pub skipped: u32,
    /// Display name of the flow that ran (`FlowGraph.name`). Empty for runs
    /// recorded before this field existed, or when the caller didn't supply it.
    #[serde(default)]
    pub flow_name: String,
    /// How the run was triggered: `manual` (default) or `scheduled` (fired by
    /// the background scheduler, roadmap E11).
    #[serde(default = "default_trigger")]
    pub trigger: String,
}

fn default_trigger() -> String {
    "manual".into()
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionStepRecord {
    pub execution_id: String,
    pub node_id: String,
    pub status: String,
    pub started_at: DateTime<Utc>,
    pub ended_at: Option<DateTime<Utc>>,
    pub output: Option<serde_json::Value>,
    pub error: Option<String>,
    pub reason: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InterceptionRecord {
    pub execution_id: String,
    pub at: DateTime<Utc>,
    /// Label/id of the node whose failure triggered the interception.
    pub failed_node: String,
    /// The detected issue (error / validation message), if any.
    pub issue: Option<String>,
    /// What the auto-fix changed/added (e.g. the corrective sub-flow's nodes).
    pub summary: String,
    /// Flow DSL before the fix (the flow that ran) and after it - the snapshot +
    /// diff for the History view (roadmap E1 autonomous session history).
    #[serde(default)]
    pub dsl_before: Option<String>,
    #[serde(default)]
    pub dsl_after: Option<String>,
}

/// One AI audit event in an execution's trail (RAO constraint 6): a model
/// invocation, a routing decision, or a human review action. `kind` is the
/// event discriminant (`ai_invocation` / `ai_routing_decision` /
/// `ai_review_required` / `ai_review_resolved`); `payload` holds the
/// kind-specific fields as JSON.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AiAuditRecord {
    pub execution_id: String,
    pub node_id: String,
    pub at: DateTime<Utc>,
    pub kind: String,
    pub payload: serde_json::Value,
}

/// Compact run row for list views: the execution plus its monotonic run number
/// (`seq` = SQLite rowid) and the first failed node, if any.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunSummary {
    pub seq: i64,
    pub execution_id: String,
    pub flow_name: String,
    pub status: String,
    pub started_at: DateTime<Utc>,
    pub ended_at: Option<DateTime<Utc>>,
    pub succeeded: u32,
    pub failed: u32,
    pub skipped: u32,
    pub failed_node: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TemplateCollectionRecord {
    pub slug: String,
    pub name: String,
    pub is_system: bool,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TemplateMembershipRow {
    pub template_slug: String,
    pub collection_slug: String,
    pub hub_source_slug: Option<String>,
    pub hub_version: Option<String>,
    pub installed_at: Option<DateTime<Utc>>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeLibraryRow {
    pub slug: String,
    pub version: String,
    pub installed_at: DateTime<Utc>,
}

/// A persisted recurring-run timer for a saved flow (roadmap E11). Keyed by the
/// saved template slug so a flow has at most one schedule. `next_run_at` is the
/// next instant the background scheduler should fire (recomputed after each run
/// from `frequency` + `anchor_at`); `None` when disabled.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScheduleRecord {
    pub template_slug: String,
    pub collection_slug: String,
    pub flow_name: String,
    pub enabled: bool,
    /// One of `hourly` | `daily` | `weekly` | `monthly` | `yearly` |
    /// `cron` | `every_n_minutes` | `once`.
    pub frequency: String,
    /// User-chosen first-run instant; also the recurrence anchor (its
    /// minute/hour/weekday/day-of-month/month select the recurring slot).
    pub anchor_at: DateTime<Utc>,
    #[serde(default = "default_timezone")]
    pub timezone: String,
    #[serde(default)]
    pub cron: Option<String>,
    #[serde(default)]
    pub every_minutes: Option<u32>,
    #[serde(default)]
    pub until: Option<DateTime<Utc>>,
    #[serde(default)]
    pub max_runs: Option<u32>,
    #[serde(default = "default_catchup")]
    pub catchup: String,
    pub next_run_at: Option<DateTime<Utc>>,
    pub last_run_at: Option<DateTime<Utc>>,
    #[serde(default)]
    pub last_status: Option<String>,
    /// Number of times the background scheduler has fired this flow.
    #[serde(default)]
    pub run_count: u32,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

fn default_timezone() -> String {
    "UTC".into()
}

fn default_catchup() -> String {
    "run-once".into()
}

#[derive(Clone)]
pub struct Store {
    conn: Arc<Mutex<Connection>>,
}

impl Store {
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        let conn = Connection::open(path.as_ref())?;
        Self::init(conn)
    }

    pub fn open_in_memory() -> Result<Self, StoreError> {
        let conn = Connection::open_in_memory()?;
        Self::init(conn)
    }

    fn init(conn: Connection) -> Result<Self, StoreError> {
        // PRAGMA must run before any DDL so the FK on template_membership
        // actually enforces. rusqlite leaves FKs off by default per connection.
        conn.execute_batch(
            r#"
            PRAGMA foreign_keys = ON;
            CREATE TABLE IF NOT EXISTS executions (
                execution_id TEXT PRIMARY KEY,
                status       TEXT NOT NULL,
                started_at   TEXT NOT NULL,
                ended_at     TEXT,
                succeeded    INTEGER NOT NULL DEFAULT 0,
                failed       INTEGER NOT NULL DEFAULT 0,
                skipped      INTEGER NOT NULL DEFAULT 0,
                flow_name    TEXT NOT NULL DEFAULT '',
                trigger      TEXT NOT NULL DEFAULT 'manual'
            );
            CREATE TABLE IF NOT EXISTS execution_steps (
                execution_id TEXT NOT NULL,
                node_id      TEXT NOT NULL,
                status       TEXT NOT NULL,
                started_at   TEXT NOT NULL,
                ended_at     TEXT,
                output       TEXT,
                error        TEXT,
                reason       TEXT,
                PRIMARY KEY (execution_id, node_id),
                FOREIGN KEY (execution_id) REFERENCES executions(execution_id)
            );
            CREATE INDEX IF NOT EXISTS idx_executions_started_at
                ON executions(started_at DESC);
            CREATE TABLE IF NOT EXISTS template_collections (
                slug        TEXT PRIMARY KEY,
                name        TEXT NOT NULL,
                is_system   INTEGER NOT NULL DEFAULT 0,
                created_at  TEXT NOT NULL,
                updated_at  TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS template_membership (
                template_slug    TEXT PRIMARY KEY,
                collection_slug  TEXT NOT NULL,
                hub_source_slug  TEXT,
                hub_version      TEXT,
                installed_at     TEXT,
                FOREIGN KEY (collection_slug) REFERENCES template_collections(slug)
            );
            CREATE INDEX IF NOT EXISTS idx_template_membership_collection
                ON template_membership(collection_slug);
            CREATE TABLE IF NOT EXISTS node_library (
                slug         TEXT PRIMARY KEY,
                version      TEXT NOT NULL,
                installed_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS execution_interceptions (
                id           INTEGER PRIMARY KEY AUTOINCREMENT,
                execution_id TEXT NOT NULL,
                at           TEXT NOT NULL,
                failed_node  TEXT NOT NULL,
                issue        TEXT,
                summary      TEXT NOT NULL,
                dsl_before   TEXT,
                dsl_after    TEXT
            );
            CREATE INDEX IF NOT EXISTS idx_interceptions_execution
                ON execution_interceptions(execution_id);
            CREATE TABLE IF NOT EXISTS execution_ai_audit (
                id           INTEGER PRIMARY KEY AUTOINCREMENT,
                execution_id TEXT NOT NULL,
                node_id      TEXT NOT NULL,
                at           TEXT NOT NULL,
                kind         TEXT NOT NULL,
                payload      TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_ai_audit_execution
                ON execution_ai_audit(execution_id);
            CREATE TABLE IF NOT EXISTS flow_schedules (
                template_slug   TEXT PRIMARY KEY,
                collection_slug TEXT NOT NULL,
                flow_name       TEXT NOT NULL DEFAULT '',
                enabled         INTEGER NOT NULL DEFAULT 0,
                frequency       TEXT NOT NULL,
                anchor_at       TEXT NOT NULL,
                timezone        TEXT NOT NULL DEFAULT 'UTC',
                cron            TEXT,
                every_minutes   INTEGER,
                until_at        TEXT,
                max_runs        INTEGER,
                catchup         TEXT NOT NULL DEFAULT 'run-once',
                next_run_at     TEXT,
                last_run_at     TEXT,
                last_status     TEXT,
                run_count       INTEGER NOT NULL DEFAULT 0,
                created_at      TEXT NOT NULL,
                updated_at      TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_flow_schedules_due
                ON flow_schedules(enabled, next_run_at);
            CREATE TABLE IF NOT EXISTS memory_kv (
                key   TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS flow_workspaces (
                flow_id    TEXT PRIMARY KEY,
                path       TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            "#,
        )?;
        Ok(Self {
            conn: Arc::new(Mutex::new(conn)),
        })
    }

    pub fn upsert_execution(&self, rec: &ExecutionRecord) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO executions (execution_id, status, started_at, ended_at, succeeded, failed, skipped, flow_name, trigger)
            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
            ON CONFLICT(execution_id) DO UPDATE SET
                status = excluded.status,
                ended_at = excluded.ended_at,
                succeeded = excluded.succeeded,
                failed = excluded.failed,
                skipped = excluded.skipped
            "#,
            params![
                rec.execution_id,
                rec.status,
                rec.started_at.to_rfc3339(),
                rec.ended_at.map(|d| d.to_rfc3339()),
                rec.succeeded,
                rec.failed,
                rec.skipped,
                rec.flow_name,
                rec.trigger,
            ],
        )?;
        Ok(())
    }

    pub fn upsert_step(&self, rec: &ExecutionStepRecord) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        let output_json = match &rec.output {
            Some(v) => Some(serde_json::to_string(v)?),
            None => None,
        };
        conn.execute(
            r#"
            INSERT INTO execution_steps (execution_id, node_id, status, started_at, ended_at, output, error, reason)
            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
            ON CONFLICT(execution_id, node_id) DO UPDATE SET
                status = excluded.status,
                ended_at = excluded.ended_at,
                output = excluded.output,
                error = excluded.error,
                reason = excluded.reason
            "#,
            params![
                rec.execution_id,
                rec.node_id,
                rec.status,
                rec.started_at.to_rfc3339(),
                rec.ended_at.map(|d| d.to_rfc3339()),
                output_json,
                rec.error,
                rec.reason,
            ],
        )?;
        Ok(())
    }

    pub fn record_interception(&self, rec: &InterceptionRecord) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO execution_interceptions (execution_id, at, failed_node, issue, summary, dsl_before, dsl_after)
            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
            "#,
            params![
                rec.execution_id,
                rec.at.to_rfc3339(),
                rec.failed_node,
                rec.issue,
                rec.summary,
                rec.dsl_before,
                rec.dsl_after,
            ],
        )?;
        Ok(())
    }

    /// Append one AI audit event. Insert-only: the trail is written as part of
    /// the execution cycle and never updated after the fact.
    pub fn record_ai_audit(&self, rec: &AiAuditRecord) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO execution_ai_audit (execution_id, node_id, at, kind, payload)
            VALUES (?1, ?2, ?3, ?4, ?5)
            "#,
            params![
                rec.execution_id,
                rec.node_id,
                rec.at.to_rfc3339(),
                rec.kind,
                serde_json::to_string(&rec.payload)?,
            ],
        )?;
        Ok(())
    }

    /// The AI audit trail for one execution, in insertion order.
    pub fn list_ai_audit(&self, execution_id: &str) -> Result<Vec<AiAuditRecord>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT execution_id, node_id, at, kind, payload
             FROM execution_ai_audit WHERE execution_id = ?1 ORDER BY id",
        )?;
        let rows = stmt.query_map(params![execution_id], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, String>(2)?,
                r.get::<_, String>(3)?,
                r.get::<_, String>(4)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (execution_id, node_id, at, kind, payload) = row?;
            out.push(AiAuditRecord {
                execution_id,
                node_id,
                at: at.parse().unwrap_or_else(|_| Utc::now()),
                kind,
                payload: serde_json::from_str(&payload).unwrap_or(serde_json::Value::Null),
            });
        }
        Ok(out)
    }

    /// Load the persisted working-memory map (durable agent memory, roadmap E1).
    /// One row per `{{memory.<key>}}` entry; the value is JSON. Unparseable rows
    /// are skipped rather than failing the whole load.
    pub fn load_memory(
        &self,
    ) -> Result<std::collections::HashMap<String, serde_json::Value>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare("SELECT key, value FROM memory_kv")?;
        let rows = stmt.query_map([], |r| {
            Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
        })?;
        let mut map = std::collections::HashMap::new();
        for row in rows {
            let (k, v) = row?;
            if let Ok(val) = serde_json::from_str::<serde_json::Value>(&v) {
                map.insert(k, val);
            }
        }
        Ok(map)
    }

    /// Replace the persisted working memory with a full snapshot of `map`.
    pub fn save_memory(
        &self,
        map: &std::collections::HashMap<String, serde_json::Value>,
    ) -> Result<(), StoreError> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        tx.execute("DELETE FROM memory_kv", [])?;
        {
            let mut stmt = tx.prepare("INSERT INTO memory_kv (key, value) VALUES (?1, ?2)")?;
            for (k, v) in map {
                stmt.execute(params![k, serde_json::to_string(v)?])?;
            }
        }
        tx.commit()?;
        Ok(())
    }

    /// Clear all persisted working memory.
    pub fn clear_memory(&self) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute("DELETE FROM memory_kv", [])?;
        Ok(())
    }

    pub fn list_executions(&self, limit: usize) -> Result<Vec<ExecutionRecord>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            r#"
            SELECT execution_id, status, started_at, ended_at, succeeded, failed, skipped, flow_name, trigger
            FROM executions
            ORDER BY started_at DESC
            LIMIT ?1
            "#,
        )?;
        let rows = stmt
            .query_map(params![limit as i64], row_to_execution)?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    #[allow(clippy::type_complexity)]
    pub fn get_execution(
        &self,
        execution_id: &str,
    ) -> Result<
        Option<(ExecutionRecord, Vec<ExecutionStepRecord>, Vec<InterceptionRecord>)>,
        StoreError,
    > {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT execution_id, status, started_at, ended_at, succeeded, failed, skipped, flow_name, trigger \
             FROM executions WHERE execution_id = ?1",
        )?;
        let exec = stmt.query_row(params![execution_id], row_to_execution).ok();

        let Some(exec) = exec else { return Ok(None) };

        let mut step_stmt = conn.prepare(
            "SELECT execution_id, node_id, status, started_at, ended_at, output, error, reason \
             FROM execution_steps WHERE execution_id = ?1 ORDER BY started_at",
        )?;
        let steps = step_stmt
            .query_map(params![execution_id], |row| {
                let output_json: Option<String> = row.get(5)?;
                Ok(ExecutionStepRecord {
                    execution_id: row.get(0)?,
                    node_id: row.get(1)?,
                    status: row.get(2)?,
                    started_at: parse_dt(row.get::<_, String>(3)?),
                    ended_at: row.get::<_, Option<String>>(4)?.map(parse_dt),
                    output: output_json
                        .as_deref()
                        .and_then(|s| serde_json::from_str(s).ok()),
                    error: row.get(6)?,
                    reason: row.get(7)?,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;

        let mut int_stmt = conn.prepare(
            "SELECT execution_id, at, failed_node, issue, summary, dsl_before, dsl_after \
             FROM execution_interceptions WHERE execution_id = ?1 ORDER BY at",
        )?;
        let interceptions = int_stmt
            .query_map(params![execution_id], |row| {
                Ok(InterceptionRecord {
                    execution_id: row.get(0)?,
                    at: parse_dt(row.get::<_, String>(1)?),
                    failed_node: row.get(2)?,
                    issue: row.get(3)?,
                    summary: row.get(4)?,
                    dsl_before: row.get(5)?,
                    dsl_after: row.get(6)?,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;

        Ok(Some((exec, steps, interceptions)))
    }

    /// Run summaries for list views, newest first - each with its monotonic run
    /// number (rowid) and the first failed node (for a `failed @ <node>` label).
    pub fn list_runs(&self, limit: usize) -> Result<Vec<RunSummary>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            r#"
            SELECT e.rowid, e.execution_id, e.flow_name, e.status, e.started_at, e.ended_at,
                   e.succeeded, e.failed, e.skipped,
                   (SELECT s.node_id FROM execution_steps s
                      WHERE s.execution_id = e.execution_id AND s.status = 'failed'
                      ORDER BY s.started_at LIMIT 1) AS failed_node
            FROM executions e
            ORDER BY e.started_at DESC
            LIMIT ?1
            "#,
        )?;
        let rows = stmt
            .query_map(params![limit as i64], |row| {
                Ok(RunSummary {
                    seq: row.get(0)?,
                    execution_id: row.get(1)?,
                    flow_name: row.get(2)?,
                    status: row.get(3)?,
                    started_at: parse_dt(row.get::<_, String>(4)?),
                    ended_at: row.get::<_, Option<String>>(5)?.map(parse_dt),
                    succeeded: row.get::<_, i64>(6)? as u32,
                    failed: row.get::<_, i64>(7)? as u32,
                    skipped: row.get::<_, i64>(8)? as u32,
                    failed_node: row.get(9)?,
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    pub fn list_template_collections(&self) -> Result<Vec<TemplateCollectionRecord>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT slug, name, is_system, created_at, updated_at \
             FROM template_collections ORDER BY is_system DESC, name COLLATE NOCASE ASC",
        )?;
        let rows = stmt
            .query_map([], |row| {
                Ok(TemplateCollectionRecord {
                    slug: row.get(0)?,
                    name: row.get(1)?,
                    is_system: row.get::<_, i64>(2)? != 0,
                    created_at: parse_dt(row.get::<_, String>(3)?),
                    updated_at: parse_dt(row.get::<_, String>(4)?),
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    pub fn get_template_collection(
        &self,
        slug: &str,
    ) -> Result<Option<TemplateCollectionRecord>, StoreError> {
        let conn = self.conn.lock();
        let row = conn
            .query_row(
                "SELECT slug, name, is_system, created_at, updated_at \
                 FROM template_collections WHERE slug = ?1",
                params![slug],
                |row| {
                    Ok(TemplateCollectionRecord {
                        slug: row.get(0)?,
                        name: row.get(1)?,
                        is_system: row.get::<_, i64>(2)? != 0,
                        created_at: parse_dt(row.get::<_, String>(3)?),
                        updated_at: parse_dt(row.get::<_, String>(4)?),
                    })
                },
            )
            .ok();
        Ok(row)
    }

    /// Insert if absent, otherwise update `name` + `updated_at`. `is_system`
    /// is set only on first insert - a subsequent upsert with `is_system =
    /// false` does NOT downgrade an existing system row. This makes the
    /// boot-time `ensure_default` idempotent without racing user renames.
    pub fn upsert_template_collection(
        &self,
        slug: &str,
        name: &str,
        is_system: bool,
    ) -> Result<TemplateCollectionRecord, StoreError> {
        let now = Utc::now().to_rfc3339();
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO template_collections (slug, name, is_system, created_at, updated_at)
            VALUES (?1, ?2, ?3, ?4, ?4)
            ON CONFLICT(slug) DO UPDATE SET
                name = excluded.name,
                updated_at = excluded.updated_at
            "#,
            params![slug, name, is_system as i64, now],
        )?;
        let rec = conn
            .query_row(
                "SELECT slug, name, is_system, created_at, updated_at \
                 FROM template_collections WHERE slug = ?1",
                params![slug],
                |row| {
                    Ok(TemplateCollectionRecord {
                        slug: row.get(0)?,
                        name: row.get(1)?,
                        is_system: row.get::<_, i64>(2)? != 0,
                        created_at: parse_dt(row.get::<_, String>(3)?),
                        updated_at: parse_dt(row.get::<_, String>(4)?),
                    })
                },
            )
            .map_err(StoreError::from)?;
        Ok(rec)
    }

    pub fn rename_template_collection(
        &self,
        slug: &str,
        new_name: &str,
    ) -> Result<TemplateCollectionRecord, StoreError> {
        let now = Utc::now().to_rfc3339();
        let conn = self.conn.lock();
        let updated = conn.execute(
            "UPDATE template_collections SET name = ?1, updated_at = ?2 WHERE slug = ?3",
            params![new_name, now, slug],
        )?;
        if updated == 0 {
            return Err(StoreError::Refused(format!("collection `{slug}` not found")));
        }
        let rec = conn
            .query_row(
                "SELECT slug, name, is_system, created_at, updated_at \
                 FROM template_collections WHERE slug = ?1",
                params![slug],
                |row| {
                    Ok(TemplateCollectionRecord {
                        slug: row.get(0)?,
                        name: row.get(1)?,
                        is_system: row.get::<_, i64>(2)? != 0,
                        created_at: parse_dt(row.get::<_, String>(3)?),
                        updated_at: parse_dt(row.get::<_, String>(4)?),
                    })
                },
            )
            .map_err(StoreError::from)?;
        Ok(rec)
    }

    /// Refuses on `is_system = true` and on any extant membership row pointing
    /// at the collection. Caller is expected to move templates out (set their
    /// membership to another collection, or delete them) first.
    pub fn delete_template_collection(&self, slug: &str) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        let is_system: Option<i64> = conn
            .query_row(
                "SELECT is_system FROM template_collections WHERE slug = ?1",
                params![slug],
                |row| row.get(0),
            )
            .ok();
        match is_system {
            None => {
                return Err(StoreError::Refused(format!(
                    "collection `{slug}` not found"
                )))
            }
            Some(1) => {
                return Err(StoreError::Refused(format!(
                    "collection `{slug}` is system-managed and cannot be deleted"
                )))
            }
            _ => {}
        }
        let members: i64 = conn.query_row(
            "SELECT COUNT(*) FROM template_membership WHERE collection_slug = ?1",
            params![slug],
            |row| row.get(0),
        )?;
        if members > 0 {
            return Err(StoreError::Refused(format!(
                "collection `{slug}` has {members} member(s); move them out before deleting"
            )));
        }
        conn.execute(
            "DELETE FROM template_collections WHERE slug = ?1",
            params![slug],
        )?;
        Ok(())
    }

    pub fn list_template_memberships(&self) -> Result<Vec<TemplateMembershipRow>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT template_slug, collection_slug, hub_source_slug, hub_version, installed_at \
             FROM template_membership",
        )?;
        let rows = stmt
            .query_map([], |row| {
                Ok(TemplateMembershipRow {
                    template_slug: row.get(0)?,
                    collection_slug: row.get(1)?,
                    hub_source_slug: row.get(2)?,
                    hub_version: row.get(3)?,
                    installed_at: row.get::<_, Option<String>>(4)?.map(parse_dt),
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    pub fn get_template_membership(
        &self,
        template_slug: &str,
    ) -> Result<Option<TemplateMembershipRow>, StoreError> {
        let conn = self.conn.lock();
        let row = conn
            .query_row(
                "SELECT template_slug, collection_slug, hub_source_slug, hub_version, installed_at \
                 FROM template_membership WHERE template_slug = ?1",
                params![template_slug],
                |row| {
                    Ok(TemplateMembershipRow {
                        template_slug: row.get(0)?,
                        collection_slug: row.get(1)?,
                        hub_source_slug: row.get(2)?,
                        hub_version: row.get(3)?,
                        installed_at: row.get::<_, Option<String>>(4)?.map(parse_dt),
                    })
                },
            )
            .ok();
        Ok(row)
    }

    pub fn upsert_template_membership(
        &self,
        row: &TemplateMembershipRow,
    ) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO template_membership
                (template_slug, collection_slug, hub_source_slug, hub_version, installed_at)
            VALUES (?1, ?2, ?3, ?4, ?5)
            ON CONFLICT(template_slug) DO UPDATE SET
                collection_slug = excluded.collection_slug,
                hub_source_slug = excluded.hub_source_slug,
                hub_version = excluded.hub_version,
                installed_at = excluded.installed_at
            "#,
            params![
                row.template_slug,
                row.collection_slug,
                row.hub_source_slug,
                row.hub_version,
                row.installed_at.map(|d| d.to_rfc3339()),
            ],
        )?;
        Ok(())
    }

    pub fn delete_template_membership(&self, template_slug: &str) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            "DELETE FROM template_membership WHERE template_slug = ?1",
            params![template_slug],
        )?;
        Ok(())
    }

    /// The per-flow workspace override, keyed by `FlowGraph.id`. `None` when the
    /// flow uses the edition default. Local machine state - never in the flow file.
    pub fn get_flow_workspace(&self, flow_id: &str) -> Result<Option<String>, StoreError> {
        let conn = self.conn.lock();
        let path = conn
            .query_row(
                "SELECT path FROM flow_workspaces WHERE flow_id = ?1",
                params![flow_id],
                |row| row.get::<_, String>(0),
            )
            .ok();
        Ok(path)
    }

    pub fn set_flow_workspace(&self, flow_id: &str, path: &str) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO flow_workspaces (flow_id, path, updated_at)
            VALUES (?1, ?2, ?3)
            ON CONFLICT(flow_id) DO UPDATE SET
                path = excluded.path,
                updated_at = excluded.updated_at
            "#,
            params![flow_id, path, chrono::Utc::now().to_rfc3339()],
        )?;
        Ok(())
    }

    pub fn clear_flow_workspace(&self, flow_id: &str) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            "DELETE FROM flow_workspaces WHERE flow_id = ?1",
            params![flow_id],
        )?;
        Ok(())
    }

    pub fn list_node_library_rows(&self) -> Result<Vec<NodeLibraryRow>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT slug, version, installed_at \
             FROM node_library ORDER BY slug ASC",
        )?;
        let rows = stmt
            .query_map([], |row| {
                Ok(NodeLibraryRow {
                    slug: row.get(0)?,
                    version: row.get(1)?,
                    installed_at: parse_dt(row.get::<_, String>(2)?),
                })
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    pub fn get_node_library_row(&self, slug: &str) -> Result<Option<NodeLibraryRow>, StoreError> {
        let conn = self.conn.lock();
        let row = conn
            .query_row(
                "SELECT slug, version, installed_at \
                 FROM node_library WHERE slug = ?1",
                params![slug],
                |row| {
                    Ok(NodeLibraryRow {
                        slug: row.get(0)?,
                        version: row.get(1)?,
                        installed_at: parse_dt(row.get::<_, String>(2)?),
                    })
                },
            )
            .ok();
        Ok(row)
    }

    /// Insert if absent, else update `version` + `installed_at`.
    pub fn upsert_node_library_row(
        &self,
        row: &NodeLibraryRow,
    ) -> Result<NodeLibraryRow, StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO node_library (slug, version, installed_at)
            VALUES (?1, ?2, ?3)
            ON CONFLICT(slug) DO UPDATE SET
                version = excluded.version,
                installed_at = excluded.installed_at
            "#,
            params![
                row.slug,
                row.version,
                row.installed_at.to_rfc3339(),
            ],
        )?;
        let stored = conn
            .query_row(
                "SELECT slug, version, installed_at \
                 FROM node_library WHERE slug = ?1",
                params![row.slug],
                |r| {
                    Ok(NodeLibraryRow {
                        slug: r.get(0)?,
                        version: r.get(1)?,
                        installed_at: parse_dt(r.get::<_, String>(2)?),
                    })
                },
            )
            .map_err(StoreError::from)?;
        Ok(stored)
    }

    pub fn delete_node_library_row(&self, slug: &str) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        let affected = conn.execute("DELETE FROM node_library WHERE slug = ?1", params![slug])?;
        if affected == 0 {
            return Err(StoreError::Refused(format!("node `{slug}` not found")));
        }
        Ok(())
    }

    /// Insert or replace a flow's schedule (keyed by `template_slug`). `created_at`
    /// is preserved on update.
    pub fn upsert_schedule(&self, rec: &ScheduleRecord) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            r#"
            INSERT INTO flow_schedules
                (template_slug, collection_slug, flow_name, enabled, frequency,
                 anchor_at, timezone, cron, every_minutes, until_at, max_runs, catchup,
                 next_run_at, last_run_at, last_status, run_count, created_at, updated_at)
            VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
            ON CONFLICT(template_slug) DO UPDATE SET
                collection_slug = excluded.collection_slug,
                flow_name = excluded.flow_name,
                enabled = excluded.enabled,
                frequency = excluded.frequency,
                anchor_at = excluded.anchor_at,
                timezone = excluded.timezone,
                cron = excluded.cron,
                every_minutes = excluded.every_minutes,
                until_at = excluded.until_at,
                max_runs = excluded.max_runs,
                catchup = excluded.catchup,
                next_run_at = excluded.next_run_at,
                last_run_at = excluded.last_run_at,
                last_status = excluded.last_status,
                run_count = excluded.run_count,
                updated_at = excluded.updated_at
            "#,
            params![
                rec.template_slug,
                rec.collection_slug,
                rec.flow_name,
                rec.enabled as i64,
                rec.frequency,
                rec.anchor_at.to_rfc3339(),
                rec.timezone,
                rec.cron,
                rec.every_minutes.map(|v| v as i64),
                rec.until.map(|d| d.to_rfc3339()),
                rec.max_runs.map(|v| v as i64),
                rec.catchup,
                rec.next_run_at.map(|d| d.to_rfc3339()),
                rec.last_run_at.map(|d| d.to_rfc3339()),
                rec.last_status,
                rec.run_count as i64,
                rec.created_at.to_rfc3339(),
                rec.updated_at.to_rfc3339(),
            ],
        )?;
        Ok(())
    }

    pub fn get_schedule(&self, template_slug: &str) -> Result<Option<ScheduleRecord>, StoreError> {
        let conn = self.conn.lock();
        let row = conn
            .query_row(
                "SELECT template_slug, collection_slug, flow_name, enabled, frequency, \
                 anchor_at, timezone, cron, every_minutes, until_at, max_runs, catchup, \
                 next_run_at, last_run_at, last_status, run_count, created_at, updated_at \
                 FROM flow_schedules WHERE template_slug = ?1",
                params![template_slug],
                row_to_schedule,
            )
            .ok();
        Ok(row)
    }

    pub fn list_schedules(&self) -> Result<Vec<ScheduleRecord>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT template_slug, collection_slug, flow_name, enabled, frequency, \
             anchor_at, timezone, cron, every_minutes, until_at, max_runs, catchup, \
             next_run_at, last_run_at, last_status, run_count, created_at, updated_at \
             FROM flow_schedules ORDER BY updated_at DESC",
        )?;
        let rows = stmt
            .query_map([], row_to_schedule)?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    /// Enabled schedules whose `next_run_at` is at or before `now` - the rows the
    /// background scheduler should fire on this tick.
    pub fn list_due_schedules(
        &self,
        now: DateTime<Utc>,
    ) -> Result<Vec<ScheduleRecord>, StoreError> {
        let conn = self.conn.lock();
        let mut stmt = conn.prepare(
            "SELECT template_slug, collection_slug, flow_name, enabled, frequency, \
             anchor_at, timezone, cron, every_minutes, until_at, max_runs, catchup, \
             next_run_at, last_run_at, last_status, run_count, created_at, updated_at \
             FROM flow_schedules \
             WHERE enabled = 1 AND next_run_at IS NOT NULL AND next_run_at <= ?1 \
             ORDER BY next_run_at ASC",
        )?;
        let rows = stmt
            .query_map(params![now.to_rfc3339()], row_to_schedule)?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    /// Advance a schedule's timer on a fire: set `last_run_at`, the recomputed
    /// `next_run_at`, and bump the run counter.
    pub fn mark_schedule_run(
        &self,
        template_slug: &str,
        last_run_at: DateTime<Utc>,
        next_run_at: Option<DateTime<Utc>>,
        last_status: &str,
    ) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            "UPDATE flow_schedules \
             SET last_run_at = ?2, next_run_at = ?3, last_status = ?4, \
                 run_count = run_count + 1, enabled = CASE WHEN ?3 IS NULL THEN 0 ELSE enabled END, updated_at = ?2 \
             WHERE template_slug = ?1",
            params![
                template_slug,
                last_run_at.to_rfc3339(),
                next_run_at.map(|d| d.to_rfc3339()),
                last_status,
            ],
        )?;
        Ok(())
    }

    pub fn advance_schedule_without_run(
        &self,
        template_slug: &str,
        at: DateTime<Utc>,
        next_run_at: Option<DateTime<Utc>>,
        last_status: &str,
    ) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            "UPDATE flow_schedules \
             SET next_run_at = ?2, last_status = ?3, \
                 enabled = CASE WHEN ?2 IS NULL THEN 0 ELSE enabled END, updated_at = ?4 \
             WHERE template_slug = ?1",
            params![
                template_slug,
                next_run_at.map(|d| d.to_rfc3339()),
                last_status,
                at.to_rfc3339(),
            ],
        )?;
        Ok(())
    }

    pub fn set_schedule_enabled(
        &self,
        template_slug: &str,
        enabled: bool,
        next_run_at: Option<DateTime<Utc>>,
    ) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        let now = Utc::now().to_rfc3339();
        conn.execute(
            "UPDATE flow_schedules SET enabled = ?2, next_run_at = ?3, updated_at = ?4 \
             WHERE template_slug = ?1",
            params![
                template_slug,
                enabled as i64,
                next_run_at.map(|d| d.to_rfc3339()),
                now,
            ],
        )?;
        Ok(())
    }

    pub fn migrate_schedule(
        &self,
        old_slug: &str,
        new_slug: &str,
        collection_slug: &str,
        flow_name: &str,
    ) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        let now = Utc::now().to_rfc3339();
        conn.execute(
            "UPDATE flow_schedules \
             SET template_slug = ?2, collection_slug = ?3, flow_name = ?4, updated_at = ?5 \
             WHERE template_slug = ?1",
            params![old_slug, new_slug, collection_slug, flow_name, now],
        )?;
        Ok(())
    }

    pub fn delete_schedule(&self, template_slug: &str) -> Result<(), StoreError> {
        let conn = self.conn.lock();
        conn.execute(
            "DELETE FROM flow_schedules WHERE template_slug = ?1",
            params![template_slug],
        )?;
        Ok(())
    }
}

fn row_to_execution(row: &rusqlite::Row<'_>) -> rusqlite::Result<ExecutionRecord> {
    Ok(ExecutionRecord {
        execution_id: row.get(0)?,
        status: row.get(1)?,
        started_at: parse_dt(row.get::<_, String>(2)?),
        ended_at: row.get::<_, Option<String>>(3)?.map(parse_dt),
        succeeded: row.get::<_, i64>(4)? as u32,
        failed: row.get::<_, i64>(5)? as u32,
        skipped: row.get::<_, i64>(6)? as u32,
        flow_name: row.get(7)?,
        trigger: row.get(8)?,
    })
}

fn row_to_schedule(row: &rusqlite::Row<'_>) -> rusqlite::Result<ScheduleRecord> {
    Ok(ScheduleRecord {
        template_slug: row.get(0)?,
        collection_slug: row.get(1)?,
        flow_name: row.get(2)?,
        enabled: row.get::<_, i64>(3)? != 0,
        frequency: row.get(4)?,
        anchor_at: parse_dt(row.get::<_, String>(5)?),
        timezone: row.get(6)?,
        cron: row.get(7)?,
        every_minutes: row.get::<_, Option<i64>>(8)?.map(|v| v as u32),
        until: row.get::<_, Option<String>>(9)?.map(parse_dt),
        max_runs: row.get::<_, Option<i64>>(10)?.map(|v| v as u32),
        catchup: row.get(11)?,
        next_run_at: row.get::<_, Option<String>>(12)?.map(parse_dt),
        last_run_at: row.get::<_, Option<String>>(13)?.map(parse_dt),
        last_status: row.get(14)?,
        run_count: row.get::<_, i64>(15)? as u32,
        created_at: parse_dt(row.get::<_, String>(16)?),
        updated_at: parse_dt(row.get::<_, String>(17)?),
    })
}

fn parse_dt(raw: String) -> DateTime<Utc> {
    DateTime::parse_from_rfc3339(&raw)
        .map(|d| d.with_timezone(&Utc))
        .unwrap_or_else(|_| Utc::now())
}

#[cfg(test)]
mod tests {
    use super::*;

    fn rec(id: &str, status: &str) -> ExecutionRecord {
        ExecutionRecord {
            execution_id: id.into(),
            status: status.into(),
            started_at: Utc::now(),
            ended_at: Some(Utc::now()),
            succeeded: 1,
            failed: 0,
            skipped: 0,
            flow_name: "demo".into(),
            trigger: "manual".into(),
        }
    }

    fn step(exec: &str, node: &str, status: &str) -> ExecutionStepRecord {
        ExecutionStepRecord {
            execution_id: exec.into(),
            node_id: node.into(),
            status: status.into(),
            started_at: Utc::now(),
            ended_at: Some(Utc::now()),
            output: Some(serde_json::json!({ "ok": true })),
            error: None,
            reason: None,
        }
    }

    #[test]
    fn round_trip_execution_and_steps() {
        let store = Store::open_in_memory().unwrap();
        store.upsert_execution(&rec("e1", "succeeded")).unwrap();
        store.upsert_step(&step("e1", "n1", "succeeded")).unwrap();
        store.upsert_step(&step("e1", "n2", "skipped")).unwrap();

        let listed = store.list_executions(10).unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].execution_id, "e1");

        let detail = store.get_execution("e1").unwrap().unwrap();
        assert_eq!(detail.0.status, "succeeded");
        assert_eq!(detail.1.len(), 2);
    }

    #[test]
    fn memory_round_trips_and_replaces() {
        let store = Store::open_in_memory().unwrap();
        assert!(store.load_memory().unwrap().is_empty());

        let mut m = std::collections::HashMap::new();
        m.insert("count".to_string(), serde_json::json!(3));
        m.insert("note".to_string(), serde_json::json!({ "k": "v" }));
        store.save_memory(&m).unwrap();
        let loaded = store.load_memory().unwrap();
        assert_eq!(loaded.get("count"), Some(&serde_json::json!(3)));
        assert_eq!(loaded.get("note"), Some(&serde_json::json!({ "k": "v" })));

        // save_memory replaces the whole snapshot.
        let mut m2 = std::collections::HashMap::new();
        m2.insert("only".to_string(), serde_json::json!(true));
        store.save_memory(&m2).unwrap();
        let reloaded = store.load_memory().unwrap();
        assert_eq!(reloaded.len(), 1);
        assert_eq!(reloaded.get("only"), Some(&serde_json::json!(true)));

        store.clear_memory().unwrap();
        assert!(store.load_memory().unwrap().is_empty());
    }

    #[test]
    fn upsert_replaces_existing_step() {
        let store = Store::open_in_memory().unwrap();
        store.upsert_execution(&rec("e1", "running")).unwrap();
        let mut s = step("e1", "n1", "running");
        s.ended_at = None;
        store.upsert_step(&s).unwrap();

        let mut s2 = step("e1", "n1", "succeeded");
        s2.error = Some("ignored".into());
        store.upsert_step(&s2).unwrap();

        let detail = store.get_execution("e1").unwrap().unwrap();
        assert_eq!(detail.1.len(), 1);
        assert_eq!(detail.1[0].status, "succeeded");
    }

    #[test]
    fn list_executions_orders_newest_first() {
        let store = Store::open_in_memory().unwrap();
        let mut a = rec("a", "succeeded");
        a.started_at = Utc::now() - chrono::Duration::hours(2);
        let mut b = rec("b", "failed");
        b.started_at = Utc::now() - chrono::Duration::hours(1);
        store.upsert_execution(&a).unwrap();
        store.upsert_execution(&b).unwrap();
        let listed = store.list_executions(10).unwrap();
        assert_eq!(listed[0].execution_id, "b");
        assert_eq!(listed[1].execution_id, "a");
    }

    fn membership(template: &str, collection: &str) -> TemplateMembershipRow {
        TemplateMembershipRow {
            template_slug: template.into(),
            collection_slug: collection.into(),
            hub_source_slug: None,
            hub_version: None,
            installed_at: None,
        }
    }

    #[test]
    fn upsert_and_list_template_collections_pins_system_first() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_template_collection("default", "Default", true)
            .unwrap();
        store
            .upsert_template_collection("alpha", "Alpha collection", false)
            .unwrap();
        store
            .upsert_template_collection("beta", "Beta collection", false)
            .unwrap();
        let listed = store.list_template_collections().unwrap();
        assert_eq!(listed[0].slug, "default", "system collection pins first");
        assert!(listed[0].is_system);
        assert_eq!(listed[1].slug, "alpha");
        assert_eq!(listed[2].slug, "beta");
    }

    #[test]
    fn upsert_does_not_downgrade_system_flag() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_template_collection("default", "Default", true)
            .unwrap();
        // A subsequent upsert with is_system=false (which is what
        // ensure_default does on every boot) must leave the flag set.
        store
            .upsert_template_collection("default", "Default", false)
            .unwrap();
        let rec = store
            .get_template_collection("default")
            .unwrap()
            .expect("row exists");
        assert!(rec.is_system, "is_system must not be downgraded on upsert");
    }

    #[test]
    fn rename_template_collection_changes_name_only() {
        let store = Store::open_in_memory().unwrap();
        let original = store
            .upsert_template_collection("alpha", "Alpha", false)
            .unwrap();
        let renamed = store
            .rename_template_collection("alpha", "Renamed Alpha")
            .unwrap();
        assert_eq!(renamed.slug, original.slug, "slug must not change on rename");
        assert_eq!(renamed.name, "Renamed Alpha");
    }

    #[test]
    fn rename_missing_collection_returns_refused() {
        let store = Store::open_in_memory().unwrap();
        let err = store
            .rename_template_collection("ghost", "x")
            .expect_err("missing slug refused");
        assert!(matches!(err, StoreError::Refused(_)), "got {err:?}");
    }

    #[test]
    fn delete_refuses_system_and_non_empty_collections() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_template_collection("default", "Default", true)
            .unwrap();
        store
            .upsert_template_collection("alpha", "Alpha", false)
            .unwrap();
        store
            .upsert_template_membership(&membership("foo", "alpha"))
            .unwrap();

        let sys_err = store
            .delete_template_collection("default")
            .expect_err("system refused");
        assert!(matches!(sys_err, StoreError::Refused(_)), "got {sys_err:?}");

        let non_empty_err = store
            .delete_template_collection("alpha")
            .expect_err("non-empty refused");
        assert!(
            matches!(non_empty_err, StoreError::Refused(_)),
            "got {non_empty_err:?}"
        );

        store.delete_template_membership("foo").unwrap();
        store.delete_template_collection("alpha").unwrap();
        let listed = store.list_template_collections().unwrap();
        assert!(listed.iter().all(|c| c.slug != "alpha"));
    }

    #[test]
    fn membership_upsert_replaces_and_get_returns_row() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_template_collection("alpha", "Alpha", false)
            .unwrap();
        store
            .upsert_template_collection("beta", "Beta", false)
            .unwrap();
        store
            .upsert_template_membership(&membership("foo", "alpha"))
            .unwrap();
        let got = store
            .get_template_membership("foo")
            .unwrap()
            .expect("row exists");
        assert_eq!(got.collection_slug, "alpha");

        store
            .upsert_template_membership(&membership("foo", "beta"))
            .unwrap();
        let moved = store
            .get_template_membership("foo")
            .unwrap()
            .expect("row exists");
        assert_eq!(moved.collection_slug, "beta", "upsert replaces in place");
    }

    #[test]
    fn membership_fk_violation_when_collection_unknown() {
        let store = Store::open_in_memory().unwrap();
        // No collection inserted; FK must reject.
        let err = store
            .upsert_template_membership(&membership("foo", "ghost"))
            .expect_err("FK violation");
        assert!(matches!(err, StoreError::Sqlite(_)), "got {err:?}");
    }

    #[test]
    fn delete_membership_is_idempotent() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_template_collection("alpha", "Alpha", false)
            .unwrap();
        store
            .upsert_template_membership(&membership("foo", "alpha"))
            .unwrap();
        store.delete_template_membership("foo").unwrap();
        // Deleting again is a no-op, not an error - callers can be sloppy.
        store.delete_template_membership("foo").unwrap();
        assert!(store.get_template_membership("foo").unwrap().is_none());
    }

    #[test]
    fn hub_source_fields_round_trip() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_template_collection("alpha", "Alpha", false)
            .unwrap();
        let installed_at = Utc::now();
        store
            .upsert_template_membership(&TemplateMembershipRow {
                template_slug: "foo".into(),
                collection_slug: "alpha".into(),
                hub_source_slug: Some("hub-foo".into()),
                hub_version: Some("1.2.3".into()),
                installed_at: Some(installed_at),
            })
            .unwrap();
        let got = store
            .get_template_membership("foo")
            .unwrap()
            .expect("row exists");
        assert_eq!(got.hub_source_slug.as_deref(), Some("hub-foo"));
        assert_eq!(got.hub_version.as_deref(), Some("1.2.3"));
        // RFC3339 round-trip can lose sub-second precision; compare to-second.
        let got_at = got.installed_at.expect("installed_at present");
        assert_eq!(
            got_at.timestamp(),
            installed_at.timestamp(),
            "installed_at round-trips to second precision"
        );
    }

    fn node_row(slug: &str, version: &str) -> NodeLibraryRow {
        NodeLibraryRow {
            slug: slug.into(),
            version: version.into(),
            installed_at: Utc::now(),
        }
    }

    #[test]
    fn node_library_upsert_list_get_roundtrip() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_node_library_row(&node_row("alpha", "1.0.0"))
            .unwrap();
        store
            .upsert_node_library_row(&node_row("beta", "2.0.0"))
            .unwrap();
        let listed = store.list_node_library_rows().unwrap();
        assert_eq!(listed.len(), 2);
        assert_eq!(listed[0].slug, "alpha");
        assert_eq!(listed[1].slug, "beta");
        let got = store
            .get_node_library_row("alpha")
            .unwrap()
            .expect("row exists");
        assert_eq!(got.version, "1.0.0");
    }

    #[test]
    fn node_library_upsert_updates_version() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_node_library_row(&node_row("alpha", "1.0.0"))
            .unwrap();
        store
            .upsert_node_library_row(&node_row("alpha", "1.1.0"))
            .unwrap();
        let got = store
            .get_node_library_row("alpha")
            .unwrap()
            .expect("row exists");
        assert_eq!(got.version, "1.1.0");
    }

    #[test]
    fn node_library_delete_removes_row() {
        let store = Store::open_in_memory().unwrap();
        store
            .upsert_node_library_row(&node_row("alpha", "1.0.0"))
            .unwrap();
        store.delete_node_library_row("alpha").unwrap();
        assert!(store.get_node_library_row("alpha").unwrap().is_none());
    }

    #[test]
    fn node_library_delete_refuses_missing_row() {
        let store = Store::open_in_memory().unwrap();
        let err = store
            .delete_node_library_row("ghost")
            .expect_err("missing slug refused");
        assert!(matches!(err, StoreError::Refused(_)), "got {err:?}");
    }

    fn schedule(slug: &str, enabled: bool, next: Option<DateTime<Utc>>) -> ScheduleRecord {
        let now = Utc::now();
        ScheduleRecord {
            template_slug: slug.into(),
            collection_slug: "default".into(),
            flow_name: "demo".into(),
            enabled,
            frequency: "daily".into(),
            anchor_at: now,
            timezone: "UTC".into(),
            cron: None,
            every_minutes: None,
            until: None,
            max_runs: None,
            catchup: "run-once".into(),
            next_run_at: next,
            last_run_at: None,
            last_status: None,
            run_count: 0,
            created_at: now,
            updated_at: now,
        }
    }

    #[test]
    fn schedule_upsert_get_and_replace() {
        let store = Store::open_in_memory().unwrap();
        store.upsert_schedule(&schedule("foo", true, Some(Utc::now()))).unwrap();
        let got = store.get_schedule("foo").unwrap().expect("row exists");
        assert!(got.enabled);
        assert_eq!(got.frequency, "daily");

        let mut updated = schedule("foo", false, None);
        updated.frequency = "weekly".into();
        store.upsert_schedule(&updated).unwrap();
        let got = store.get_schedule("foo").unwrap().expect("row exists");
        assert!(!got.enabled);
        assert_eq!(got.frequency, "weekly", "upsert replaces in place");
    }

    #[test]
    fn list_due_excludes_disabled_and_future() {
        let store = Store::open_in_memory().unwrap();
        let past = Utc::now() - chrono::Duration::minutes(5);
        let future = Utc::now() + chrono::Duration::hours(1);
        store.upsert_schedule(&schedule("due", true, Some(past))).unwrap();
        store.upsert_schedule(&schedule("future", true, Some(future))).unwrap();
        store.upsert_schedule(&schedule("off", false, Some(past))).unwrap();
        let due = store.list_due_schedules(Utc::now()).unwrap();
        assert_eq!(due.len(), 1);
        assert_eq!(due[0].template_slug, "due");
    }

    #[test]
    fn mark_run_advances_timer_and_delete_removes() {
        let store = Store::open_in_memory().unwrap();
        let past = Utc::now() - chrono::Duration::minutes(5);
        store.upsert_schedule(&schedule("foo", true, Some(past))).unwrap();
        let next = Utc::now() + chrono::Duration::days(1);
        store
            .mark_schedule_run("foo", Utc::now(), Some(next), "succeeded")
            .unwrap();
        let got = store.get_schedule("foo").unwrap().expect("row exists");
        assert!(got.last_run_at.is_some());
        assert_eq!(got.last_status.as_deref(), Some("succeeded"));
        assert_eq!(got.run_count, 1, "mark_schedule_run bumps the counter");
        assert!(got.next_run_at.unwrap() > Utc::now());
        assert!(store.list_due_schedules(Utc::now()).unwrap().is_empty());

        store.delete_schedule("foo").unwrap();
        assert!(store.get_schedule("foo").unwrap().is_none());
    }

    #[test]
    fn flow_workspace_set_get_clear_roundtrip() {
        let store = Store::open_in_memory().unwrap();
        assert!(store.get_flow_workspace("flow-1").unwrap().is_none());

        store.set_flow_workspace("flow-1", "/tmp/project").unwrap();
        assert_eq!(
            store.get_flow_workspace("flow-1").unwrap().as_deref(),
            Some("/tmp/project")
        );

        // Upsert replaces the path.
        store.set_flow_workspace("flow-1", "/tmp/other").unwrap();
        assert_eq!(
            store.get_flow_workspace("flow-1").unwrap().as_deref(),
            Some("/tmp/other")
        );

        store.clear_flow_workspace("flow-1").unwrap();
        assert!(store.get_flow_workspace("flow-1").unwrap().is_none());
    }
}