store.rs

//! SQLite persistence layer via sqlx.
//!
//! Provides [`Store`] — a connection-pool wrapper with CRUD operations for
//! three domain entities: [`Project`], [`Session`], and [`Terminal`].
//!
//! WAL mode is enabled at pool-open time so concurrent async readers do not
//! block writers.  Foreign-key enforcement is also turned on.
//!
//! Optimistic concurrency: state-update helpers include a WHERE clause that
//! matches the caller's last-known `updated_at` timestamp.  A return value of
//! `false` means either the row was not found or a concurrent writer already
//! changed it — the caller should re-read and retry.

use chrono::{DateTime, Utc};
use sqlx::{
    sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous},
    SqlitePool,
};
use std::str::FromStr;
use uuid::Uuid;

// ---------------------------------------------------------------------------
// Domain types
// ---------------------------------------------------------------------------

/// State of a cloned git project.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProjectState {
    Cloning,
    Ready,
    Failed,
}

impl ProjectState {
    fn as_str(&self) -> &'static str {
        match self {
            Self::Cloning => "Cloning",
            Self::Ready => "Ready",
            Self::Failed => "Failed",
        }
    }

    /// Lowercase state string for REST API responses.
    pub fn as_api_str(&self) -> &'static str {
        match self {
            Self::Cloning => "cloning",
            Self::Ready => "ready",
            Self::Failed => "failed",
        }
    }
}

impl FromStr for ProjectState {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "Cloning" => Ok(Self::Cloning),
            "Ready" => Ok(Self::Ready),
            "Failed" => Ok(Self::Failed),
            other => Err(format!("unknown ProjectState: {other}")),
        }
    }
}

/// A cloned git repository tracked by the runner.
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct Project {
    pub id: String,
    pub name: String,
    pub git_url: String,
    pub local_path: String,
    /// Default Docker image for sessions in this project (image resolution chain level 2).
    pub default_image: Option<String>,
    pub state: ProjectState,
    pub clone_error: Option<String>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub deleted_at: Option<DateTime<Utc>>,
}

/// Parameters required to create a new project.
#[derive(Debug, Clone)]
pub struct CreateProject {
    pub name: String,
    pub git_url: String,
    pub local_path: String,
}

// ---------------------------------------------------------------------------

/// State of a container session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionState {
    Starting,
    Running,
    Stopped,
    Failed,
}

impl SessionState {
    fn as_str(&self) -> &'static str {
        match self {
            Self::Starting => "Starting",
            Self::Running => "Running",
            Self::Stopped => "Stopped",
            Self::Failed => "Failed",
        }
    }

    /// Lowercase state string used in REST API responses.
    pub fn as_api_str(&self) -> &'static str {
        match self {
            Self::Starting => "starting",
            Self::Running => "running",
            Self::Stopped => "stopped",
            Self::Failed => "failed",
        }
    }
}

impl FromStr for SessionState {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "Starting" => Ok(Self::Starting),
            "Running" => Ok(Self::Running),
            "Stopped" => Ok(Self::Stopped),
            "Failed" => Ok(Self::Failed),
            other => Err(format!("unknown SessionState: {other}")),
        }
    }
}

/// A running container session attached to a project.
#[derive(Debug, Clone)]
pub struct Session {
    pub id: String,
    pub project_id: String,
    pub branch: String,
    pub image: String,
    pub container_id: Option<String>,
    pub state: SessionState,
    pub profile: String,
    /// Isolated working directory path for this session.
    pub work_dir: String,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub stopped_at: Option<DateTime<Utc>>,
    pub error_reason: Option<String>,
    pub deleted_at: Option<DateTime<Utc>>,
}

/// Parameters required to create a new session.
#[derive(Debug, Clone)]
pub struct CreateSession {
    pub project_id: String,
    pub branch: String,
    pub image: String,
    pub profile: String,
    /// Isolated working directory path.
    #[allow(dead_code)]
    pub work_dir: String,
}

// ---------------------------------------------------------------------------

/// State of a terminal PTY within a session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TerminalState {
    Starting,
    Running,
    Stopped,
    Failed,
}

impl TerminalState {
    fn as_str(&self) -> &'static str {
        match self {
            Self::Starting => "Starting",
            Self::Running => "Running",
            Self::Stopped => "Stopped",
            Self::Failed => "Failed",
        }
    }

    /// Lowercase state string used in REST API responses.
    pub fn as_api_str(&self) -> &'static str {
        match self {
            Self::Starting => "starting",
            Self::Running => "running",
            Self::Stopped => "stopped",
            Self::Failed => "failed",
        }
    }
}

impl FromStr for TerminalState {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s {
            "Starting" => Ok(Self::Starting),
            "Running" => Ok(Self::Running),
            "Stopped" => Ok(Self::Stopped),
            "Failed" => Ok(Self::Failed),
            other => Err(format!("unknown TerminalState: {other}")),
        }
    }
}

/// A terminal PTY instance belonging to a session.
///
/// Session : Terminal = 1 : N.  Each terminal represents an independent
/// PTY process within the same container/environment.
#[derive(Debug, Clone)]
pub struct Terminal {
    pub id: String,
    pub session_id: String,
    pub profile: String,
    pub state: TerminalState,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// Parameters required to create a new terminal.
#[derive(Debug, Clone)]
pub struct CreateTerminal {
    pub session_id: String,
    pub profile: String,
}

// ---------------------------------------------------------------------------
// Raw row types (sqlx FromRow)
// ---------------------------------------------------------------------------

#[derive(sqlx::FromRow)]
struct ProjectRow {
    id: String,
    name: String,
    git_url: String,
    local_path: String,
    default_image: Option<String>,
    state: String,
    clone_error: Option<String>,
    created_at: String,
    updated_at: String,
    deleted_at: Option<String>,
}

impl TryFrom<ProjectRow> for Project {
    type Error = sqlx::Error;

    fn try_from(row: ProjectRow) -> Result<Self, Self::Error> {
        Ok(Self {
            id: row.id,
            name: row.name,
            git_url: row.git_url,
            local_path: row.local_path,
            default_image: row.default_image,
            state: row
                .state
                .parse()
                .map_err(|e: String| sqlx::Error::Decode(e.into()))?,
            clone_error: row.clone_error,
            created_at: parse_dt(&row.created_at)?,
            updated_at: parse_dt(&row.updated_at)?,
            deleted_at: row.deleted_at.as_deref().map(parse_dt).transpose()?,
        })
    }
}

#[derive(sqlx::FromRow)]
struct SessionRow {
    id: String,
    project_id: String,
    branch: String,
    image: String,
    container_id: Option<String>,
    state: String,
    profile: String,
    work_dir: String,
    created_at: String,
    updated_at: String,
    stopped_at: Option<String>,
    error_reason: Option<String>,
    deleted_at: Option<String>,
}

impl TryFrom<SessionRow> for Session {
    type Error = sqlx::Error;

    fn try_from(row: SessionRow) -> Result<Self, Self::Error> {
        Ok(Self {
            id: row.id,
            project_id: row.project_id,
            branch: row.branch,
            image: row.image,
            container_id: row.container_id,
            state: row
                .state
                .parse()
                .map_err(|e: String| sqlx::Error::Decode(e.into()))?,
            profile: row.profile,
            work_dir: row.work_dir,
            created_at: parse_dt(&row.created_at)?,
            updated_at: parse_dt(&row.updated_at)?,
            stopped_at: row.stopped_at.as_deref().map(parse_dt).transpose()?,
            error_reason: row.error_reason,
            deleted_at: row.deleted_at.as_deref().map(parse_dt).transpose()?,
        })
    }
}

#[derive(sqlx::FromRow)]
struct TerminalRow {
    id: String,
    session_id: String,
    profile: String,
    state: String,
    created_at: String,
    updated_at: String,
}

impl TryFrom<TerminalRow> for Terminal {
    type Error = sqlx::Error;

    fn try_from(row: TerminalRow) -> Result<Self, Self::Error> {
        Ok(Self {
            id: row.id,
            session_id: row.session_id,
            profile: row.profile,
            state: row
                .state
                .parse()
                .map_err(|e: String| sqlx::Error::Decode(e.into()))?,
            created_at: parse_dt(&row.created_at)?,
            updated_at: parse_dt(&row.updated_at)?,
        })
    }
}

fn parse_dt(s: &str) -> Result<DateTime<Utc>, sqlx::Error> {
    DateTime::parse_from_rfc3339(s)
        .map(|dt| dt.with_timezone(&Utc))
        .map_err(|e| sqlx::Error::Decode(format!("invalid datetime '{s}': {e}").into()))
}

fn now_str() -> String {
    Utc::now().to_rfc3339()
}

// ---------------------------------------------------------------------------
// Delete guard error
// ---------------------------------------------------------------------------

/// Error returned by [`Store::soft_delete_session_unless_last`].
#[derive(Debug)]
pub enum SoftDeleteError {
    /// The operation was blocked because the session is the last one for its
    /// project — deleting it would violate the project-session invariant.
    LastSession,
    /// A database-level error occurred.
    Database(sqlx::Error),
}

// ---------------------------------------------------------------------------
// Store
// ---------------------------------------------------------------------------

/// Async SQLite store backed by a connection pool.
///
/// Create with [`Store::new`]; the pool is shared cheaply via `Clone`.
#[derive(Clone, Debug)]
pub struct Store {
    pool: SqlitePool,
}

impl Store {
    /// Open (or create) the SQLite database at `db_url`, enable WAL mode and
    /// foreign-key enforcement, then run pending migrations.
    ///
    /// `db_url` can be `"sqlite::memory:"` for tests or a file path such as
    /// `"sqlite:relay.db"`.
    /// Create a `Store` from an existing pool (useful for testing).
    /// The caller is responsible for running migrations if needed.
    #[cfg(test)]
    pub(crate) async fn from_pool_with_migrations(pool: SqlitePool) -> Result<Self, sqlx::Error> {
        sqlx::migrate!("./migrations").run(&pool).await?;
        Ok(Self { pool })
    }

    /// Create an in-memory `Store` for unit tests.
    ///
    /// Uses a single-connection pool so that all operations share the same
    /// in-memory database (`:memory:` creates a fresh DB per connection).
    /// Runs all migrations automatically.
    #[cfg(test)]
    pub async fn for_tests() -> Self {
        let opts = SqliteConnectOptions::from_str("sqlite::memory:")
            .expect("valid url")
            .create_if_missing(true)
            .foreign_keys(true);
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(opts)
            .await
            .expect("in-memory pool");
        Self::from_pool_with_migrations(pool)
            .await
            .expect("migrations")
    }

    pub async fn new(db_url: &str) -> Result<Self, sqlx::Error> {
        let opts = SqliteConnectOptions::from_str(db_url)?
            .create_if_missing(true)
            .journal_mode(SqliteJournalMode::Wal)
            .synchronous(SqliteSynchronous::Normal)
            .foreign_keys(true);

        let pool = SqlitePoolOptions::new()
            .max_connections(16)
            .connect_with(opts)
            .await?;

        sqlx::migrate!("./migrations").run(&pool).await?;

        Ok(Self { pool })
    }

    /// Gracefully close the connection pool.
    ///
    /// Waits for all checked-out connections to be returned, then shuts down
    /// the pool. Safe to call multiple times (subsequent calls are no-ops).
    pub async fn close(&self) {
        self.pool.close().await;
    }

    // -----------------------------------------------------------------------
    // Projects
    // -----------------------------------------------------------------------

    /// Insert a new project with `Cloning` state.
    pub async fn create_project(&self, req: CreateProject) -> Result<Project, sqlx::Error> {
        let id = Uuid::new_v4().to_string();
        let now = now_str();

        sqlx::query(
            "INSERT INTO projects (id, name, git_url, local_path, state, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, 'Cloning', ?5, ?5)",
        )
        .bind(&id)
        .bind(&req.name)
        .bind(&req.git_url)
        .bind(&req.local_path)
        .bind(&now)
        .execute(&self.pool)
        .await?;

        // Re-read to return the full row (avoids duplicating mapping logic).
        self.get_project(&id)
            .await?
            .ok_or_else(|| sqlx::Error::RowNotFound)
    }

    /// Fetch a project by id, including soft-deleted rows.
    pub async fn get_project(&self, id: &str) -> Result<Option<Project>, sqlx::Error> {
        let row = sqlx::query_as::<_, ProjectRow>(
            "SELECT id, name, git_url, local_path, default_image, state, clone_error,
                    created_at, updated_at, deleted_at
             FROM projects WHERE id = ?1",
        )
        .bind(id)
        .fetch_optional(&self.pool)
        .await?;

        row.map(Project::try_from).transpose()
    }

    /// List all non-deleted projects.
    pub async fn list_projects(&self) -> Result<Vec<Project>, sqlx::Error> {
        let rows = sqlx::query_as::<_, ProjectRow>(
            "SELECT id, name, git_url, local_path, default_image, state, clone_error,
                    created_at, updated_at, deleted_at
             FROM projects WHERE deleted_at IS NULL
             ORDER BY created_at",
        )
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(Project::try_from).collect()
    }

    /// Update the project's state with optimistic concurrency.
    ///
    /// `updated_at_check` must equal the caller's last-known `updated_at`
    /// value.  Returns `true` if the row was updated, `false` if not found or
    /// the optimistic check failed (stale read).
    pub async fn update_project_state(
        &self,
        id: &str,
        state: ProjectState,
        clone_error: Option<&str>,
        updated_at_check: &str,
    ) -> Result<bool, sqlx::Error> {
        let now = now_str();
        let rows_affected = sqlx::query(
            "UPDATE projects
             SET state = ?1, clone_error = ?2, updated_at = ?3
             WHERE id = ?4 AND updated_at = ?5 AND deleted_at IS NULL",
        )
        .bind(state.as_str())
        .bind(clone_error)
        .bind(&now)
        .bind(id)
        .bind(updated_at_check)
        .execute(&self.pool)
        .await?
        .rows_affected();

        Ok(rows_affected > 0)
    }

    /// Update the `local_path` column for a project (used after id is known).
    pub async fn update_project_local_path(
        &self,
        id: &str,
        local_path: &str,
    ) -> Result<bool, sqlx::Error> {
        let rows_affected =
            sqlx::query("UPDATE projects SET local_path = ?1 WHERE id = ?2 AND deleted_at IS NULL")
                .bind(local_path)
                .bind(id)
                .execute(&self.pool)
                .await?
                .rows_affected();

        Ok(rows_affected > 0)
    }

    /// Update the `default_image` column for a project.
    pub async fn update_project_default_image(
        &self,
        id: &str,
        default_image: Option<&str>,
    ) -> Result<bool, sqlx::Error> {
        let now = now_str();
        let rows_affected = sqlx::query(
            "UPDATE projects SET default_image = ?1, updated_at = ?2
             WHERE id = ?3 AND deleted_at IS NULL",
        )
        .bind(default_image)
        .bind(&now)
        .bind(id)
        .execute(&self.pool)
        .await?
        .rows_affected();

        Ok(rows_affected > 0)
    }

    /// Soft-delete a project.  Returns `true` if the row was marked deleted.
    pub async fn soft_delete_project(&self, id: &str) -> Result<bool, sqlx::Error> {
        let now = now_str();
        let rows_affected = sqlx::query(
            "UPDATE projects SET deleted_at = ?1
             WHERE id = ?2 AND deleted_at IS NULL",
        )
        .bind(&now)
        .bind(id)
        .execute(&self.pool)
        .await?
        .rows_affected();

        Ok(rows_affected > 0)
    }

    // -----------------------------------------------------------------------
    // Sessions
    // -----------------------------------------------------------------------

    /// Insert a new session with `Starting` state.
    pub async fn create_session(&self, req: CreateSession) -> Result<Session, sqlx::Error> {
        let id = Uuid::new_v4().to_string();
        let now = now_str();

        sqlx::query(
            "INSERT INTO sessions (id, project_id, branch, image, state, profile, work_dir, created_at, updated_at)
             VALUES (?1, ?2, ?3, ?4, 'Starting', ?5, ?6, ?7, ?7)",
        )
        .bind(&id)
        .bind(&req.project_id)
        .bind(&req.branch)
        .bind(&req.image)
        .bind(&req.profile)
        .bind(&req.work_dir)
        .bind(&now)
        .execute(&self.pool)
        .await?;

        self.get_session(&id)
            .await?
            .ok_or_else(|| sqlx::Error::RowNotFound)
    }

    /// Fetch a session by id, including soft-deleted rows.
    pub async fn get_session(&self, id: &str) -> Result<Option<Session>, sqlx::Error> {
        let row = sqlx::query_as::<_, SessionRow>(
            "SELECT id, project_id, branch, image, container_id, state, profile, work_dir,
                    created_at, updated_at, stopped_at, error_reason, deleted_at
             FROM sessions WHERE id = ?1",
        )
        .bind(id)
        .fetch_optional(&self.pool)
        .await?;

        row.map(Session::try_from).transpose()
    }

    /// List non-deleted sessions for a project.
    pub async fn list_sessions_for_project(
        &self,
        project_id: &str,
    ) -> Result<Vec<Session>, sqlx::Error> {
        let rows = sqlx::query_as::<_, SessionRow>(
            "SELECT id, project_id, branch, image, container_id, state, profile, work_dir,
                    created_at, updated_at, stopped_at, error_reason, deleted_at
             FROM sessions
             WHERE project_id = ?1 AND deleted_at IS NULL
             ORDER BY created_at",
        )
        .bind(project_id)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(Session::try_from).collect()
    }

    /// List all non-deleted sessions (global, paginated by limit/offset).
    pub async fn list_all_sessions(
        &self,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<Session>, sqlx::Error> {
        let limit = i64::try_from(limit).unwrap_or(i64::MAX);
        let offset = i64::try_from(offset).unwrap_or(i64::MAX);
        let rows = sqlx::query_as::<_, SessionRow>(
            "SELECT id, project_id, branch, image, container_id, state, profile, work_dir,
                    created_at, updated_at, stopped_at, error_reason, deleted_at
             FROM sessions
             WHERE deleted_at IS NULL
             ORDER BY created_at
             LIMIT ?1 OFFSET ?2",
        )
        .bind(limit)
        .bind(offset)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(Session::try_from).collect()
    }

    /// Update the session's state with optimistic concurrency.
    ///
    /// Returns `true` if the row was updated, `false` if the optimistic check
    /// failed or the session was not found / already deleted.
    pub async fn update_session_state(
        &self,
        id: &str,
        state: SessionState,
        container_id: Option<&str>,
        error_reason: Option<&str>,
        updated_at_check: &str,
    ) -> Result<bool, sqlx::Error> {
        let now = now_str();
        let stopped_at =
            matches!(state, SessionState::Stopped | SessionState::Failed).then(|| now.clone());

        let rows_affected = sqlx::query(
            "UPDATE sessions
             SET state = ?1, container_id = COALESCE(?2, container_id),
                 error_reason = ?3, stopped_at = COALESCE(?4, stopped_at), updated_at = ?5
             WHERE id = ?6 AND updated_at = ?7 AND deleted_at IS NULL",
        )
        .bind(state.as_str())
        .bind(container_id)
        .bind(error_reason)
        .bind(stopped_at)
        .bind(&now)
        .bind(id)
        .bind(updated_at_check)
        .execute(&self.pool)
        .await?
        .rows_affected();

        Ok(rows_affected > 0)
    }

    /// List all non-deleted sessions that are in `Starting` or `Running` state.
    ///
    /// Used by the reconciliation pass at server startup to find sessions whose
    /// containers may have disappeared while the runner was down.
    pub async fn list_active_sessions(&self) -> Result<Vec<Session>, sqlx::Error> {
        let rows = sqlx::query_as::<_, SessionRow>(
            "SELECT id, project_id, branch, image, container_id, state, profile, work_dir,
                    created_at, updated_at, stopped_at, error_reason, deleted_at
             FROM sessions
             WHERE deleted_at IS NULL AND state IN ('Starting', 'Running')
             ORDER BY created_at",
        )
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(Session::try_from).collect()
    }

    /// Force-set a session's state to `Stopped` without an optimistic-concurrency
    /// check.  Used only by the reconciliation pass where stale-read retry is not
    /// meaningful (the container is already gone).
    pub async fn force_stop_session(
        &self,
        id: &str,
        error_reason: &str,
    ) -> Result<bool, sqlx::Error> {
        let now = now_str();
        let rows_affected = sqlx::query(
            "UPDATE sessions
             SET state = 'Stopped', error_reason = ?1, stopped_at = ?2, updated_at = ?2
             WHERE id = ?3 AND deleted_at IS NULL AND state IN ('Starting', 'Running')",
        )
        .bind(error_reason)
        .bind(&now)
        .bind(id)
        .execute(&self.pool)
        .await?
        .rows_affected();

        Ok(rows_affected > 0)
    }

    /// Count all non-deleted sessions for the given project (any state).
    ///
    /// Used to enforce the invariant that a project must always have at least
    /// one session — the last session cannot be deleted.
    pub async fn count_all_sessions_for_project(
        &self,
        project_id: &str,
    ) -> Result<usize, sqlx::Error> {
        let row: (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM sessions
             WHERE project_id = ?1 AND deleted_at IS NULL",
        )
        .bind(project_id)
        .fetch_one(&self.pool)
        .await?;

        Ok(row.0 as usize)
    }

    /// Soft-delete a session.  Returns `true` if the row was marked deleted.
    pub async fn soft_delete_session(&self, id: &str) -> Result<bool, sqlx::Error> {
        let now = now_str();
        let rows_affected = sqlx::query(
            "UPDATE sessions SET deleted_at = ?1
             WHERE id = ?2 AND deleted_at IS NULL",
        )
        .bind(&now)
        .bind(id)
        .execute(&self.pool)
        .await?
        .rows_affected();

        Ok(rows_affected > 0)
    }

    /// Atomically soft-delete a session only when it is NOT the last
    /// non-deleted session for its project.
    ///
    /// Returns:
    /// - `Ok(true)`  — session was deleted
    /// - `Ok(false)` — session not found or already deleted
    /// - `Err(LastSession)` — session is the last for the project; deletion blocked
    pub async fn soft_delete_session_unless_last(&self, id: &str) -> Result<bool, SoftDeleteError> {
        let now = now_str();

        // A single UPDATE with a correlated subquery ensures atomicity under
        // SQLite's WAL serialisation without a separate transaction.
        // The WHERE clause only proceeds if the session is not the last one.
        let rows_affected = sqlx::query(
            "UPDATE sessions SET deleted_at = ?1
             WHERE id = ?2
               AND deleted_at IS NULL
               AND (
                 SELECT COUNT(*) FROM sessions s2
                 WHERE s2.project_id = (SELECT project_id FROM sessions WHERE id = ?2)
                   AND s2.deleted_at IS NULL
               ) > 1",
        )
        .bind(&now)
        .bind(id)
        .execute(&self.pool)
        .await
        .map_err(SoftDeleteError::Database)?
        .rows_affected();

        if rows_affected > 0 {
            return Ok(true);
        }

        // Determine why: does the session exist at all?
        let exists: bool = sqlx::query_scalar(
            "SELECT EXISTS(SELECT 1 FROM sessions WHERE id = ?1 AND deleted_at IS NULL)",
        )
        .bind(id)
        .fetch_one(&self.pool)
        .await
        .map_err(SoftDeleteError::Database)?;

        if exists {
            // Session exists but wasn't deleted — must be the last one.
            Err(SoftDeleteError::LastSession)
        } else {
            Ok(false)
        }
    }

    /// Count non-deleted sessions in `Starting` or `Running` state for
    /// the given project.
    ///
    /// Used by the per-project session semaphore to enforce `max_sessions`.
    #[allow(dead_code)]
    pub async fn count_active_sessions_for_project(
        &self,
        project_id: &str,
    ) -> Result<usize, sqlx::Error> {
        let row: (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM sessions
             WHERE project_id = ?1
               AND deleted_at IS NULL
               AND state IN ('Starting', 'Running')",
        )
        .bind(project_id)
        .fetch_one(&self.pool)
        .await?;

        Ok(row.0 as usize)
    }

    // -----------------------------------------------------------------------
    // Terminals
    // -----------------------------------------------------------------------

    /// Insert a new terminal with `Starting` state.
    pub async fn create_terminal(&self, req: CreateTerminal) -> Result<Terminal, sqlx::Error> {
        let id = Uuid::new_v4().to_string();
        let now = now_str();

        sqlx::query(
            "INSERT INTO terminals (id, session_id, profile, state, created_at, updated_at)
             VALUES (?1, ?2, ?3, 'Starting', ?4, ?4)",
        )
        .bind(&id)
        .bind(&req.session_id)
        .bind(&req.profile)
        .bind(&now)
        .execute(&self.pool)
        .await?;

        self.get_terminal(&id)
            .await?
            .ok_or_else(|| sqlx::Error::RowNotFound)
    }

    /// Fetch a terminal by id.
    pub async fn get_terminal(&self, id: &str) -> Result<Option<Terminal>, sqlx::Error> {
        let row = sqlx::query_as::<_, TerminalRow>(
            "SELECT id, session_id, profile, state, created_at, updated_at
             FROM terminals WHERE id = ?1",
        )
        .bind(id)
        .fetch_optional(&self.pool)
        .await?;

        row.map(Terminal::try_from).transpose()
    }

    /// List all terminals for a given session.
    pub async fn list_terminals_for_session(
        &self,
        session_id: &str,
    ) -> Result<Vec<Terminal>, sqlx::Error> {
        let rows = sqlx::query_as::<_, TerminalRow>(
            "SELECT id, session_id, profile, state, created_at, updated_at
             FROM terminals
             WHERE session_id = ?1
             ORDER BY created_at",
        )
        .bind(session_id)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(Terminal::try_from).collect()
    }

    /// Update a terminal's state with optimistic concurrency.
    ///
    /// Returns `true` if the row was updated, `false` if the optimistic check
    /// failed or the terminal was not found.
    pub async fn update_terminal_state(
        &self,
        id: &str,
        state: TerminalState,
        updated_at_check: &str,
    ) -> Result<bool, sqlx::Error> {
        let now = now_str();
        let rows_affected = sqlx::query(
            "UPDATE terminals
             SET state = ?1, updated_at = ?2
             WHERE id = ?3 AND updated_at = ?4",
        )
        .bind(state.as_str())
        .bind(&now)
        .bind(id)
        .bind(updated_at_check)
        .execute(&self.pool)
        .await?
        .rows_affected();

        Ok(rows_affected > 0)
    }

    /// Delete a terminal by id.  Returns `true` if a row was removed.
    pub async fn delete_terminal(&self, id: &str) -> Result<bool, sqlx::Error> {
        let rows_affected = sqlx::query("DELETE FROM terminals WHERE id = ?1")
            .bind(id)
            .execute(&self.pool)
            .await?
            .rows_affected();

        Ok(rows_affected > 0)
    }

    /// Get the first terminal for a session (ordered by creation time).
    #[allow(dead_code)]
    ///
    /// Used by gRPC `AttachSession` for backward compatibility — routes to
    /// the default terminal without the client knowing about Terminal entities.
    pub async fn get_first_terminal_for_session(
        &self,
        session_id: &str,
    ) -> Result<Option<Terminal>, sqlx::Error> {
        let row = sqlx::query_as::<_, TerminalRow>(
            "SELECT id, session_id, profile, state, created_at, updated_at
             FROM terminals
             WHERE session_id = ?1
             ORDER BY created_at ASC
             LIMIT 1",
        )
        .bind(session_id)
        .fetch_optional(&self.pool)
        .await?;

        row.map(Terminal::try_from).transpose()
    }

    // ---------------------------------------------------------------------------
}

// Unit tests
// ---------------------------------------------------------------------------

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

    async fn test_store() -> Store {
        Store::for_tests().await
    }

    #[tokio::test]
    async fn wal_mode_is_enabled() {
        // WAL is not supported for in-memory SQLite; use a real temp file.
        let tmp = tempfile::NamedTempFile::new().expect("tempfile");
        let db_url = format!("sqlite:{}", tmp.path().display());
        let store = Store::new(&db_url).await.expect("file store");
        let row: (String,) = sqlx::query_as("PRAGMA journal_mode")
            .fetch_one(&store.pool)
            .await
            .expect("PRAGMA query failed");
        assert_eq!(row.0, "wal", "expected WAL journal mode");
    }

    #[tokio::test]
    async fn create_and_get_project() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "test".into(),
                git_url: "https://github.com/example/repo".into(),
                local_path: "/tmp/repo".into(),
            })
            .await
            .expect("create_project failed");

        assert_eq!(project.state, ProjectState::Cloning);
        assert!(project.clone_error.is_none());

        let fetched = store
            .get_project(&project.id)
            .await
            .expect("get_project failed")
            .expect("project not found");

        assert_eq!(fetched.id, project.id);
        assert_eq!(fetched.name, "test");
    }

    #[tokio::test]
    async fn list_projects_excludes_soft_deleted() {
        let store = test_store().await;
        let p1 = store
            .create_project(CreateProject {
                name: "p1".into(),
                git_url: "https://example.com/1".into(),
                local_path: "/tmp/p1".into(),
            })
            .await
            .expect("create p1");
        store
            .create_project(CreateProject {
                name: "p2".into(),
                git_url: "https://example.com/2".into(),
                local_path: "/tmp/p2".into(),
            })
            .await
            .expect("create p2");

        store
            .soft_delete_project(&p1.id)
            .await
            .expect("soft delete");

        let list = store.list_projects().await.expect("list");
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].name, "p2");
    }

    #[tokio::test]
    async fn update_project_state_optimistic_concurrency() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "opt".into(),
                git_url: "https://example.com/opt".into(),
                local_path: "/tmp/opt".into(),
            })
            .await
            .expect("create");

        let current_updated_at = project.updated_at.to_rfc3339();

        // First update should succeed.
        let ok = store
            .update_project_state(&project.id, ProjectState::Ready, None, &current_updated_at)
            .await
            .expect("update 1");
        assert!(ok, "first update should succeed");

        // Second update with the stale timestamp should fail.
        let conflict = store
            .update_project_state(
                &project.id,
                ProjectState::Failed,
                Some("oops"),
                &current_updated_at, // stale
            )
            .await
            .expect("update 2");
        assert!(!conflict, "stale update should return false");

        // Verify the state is Ready, not Failed.
        let final_project = store
            .get_project(&project.id)
            .await
            .expect("get")
            .expect("not found");
        assert_eq!(final_project.state, ProjectState::Ready);
    }

    #[tokio::test]
    async fn session_crud_and_state_update() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "sess-proj".into(),
                git_url: "https://example.com/sess".into(),
                local_path: "/tmp/sess".into(),
            })
            .await
            .expect("project");

        let session = store
            .create_session(CreateSession {
                project_id: project.id.clone(),
                branch: "feature".into(),
                image: "ubuntu:24.04".into(),
                profile: "default".into(),
                work_dir: String::new(),
            })
            .await
            .expect("session");

        assert_eq!(session.state, SessionState::Starting);
        assert_eq!(session.project_id, project.id);
        assert_eq!(session.branch, "feature");
        assert_eq!(session.image, "ubuntu:24.04");
        assert!(session.container_id.is_none());

        let updated_at = session.updated_at.to_rfc3339();
        let ok = store
            .update_session_state(
                &session.id,
                SessionState::Running,
                Some("container-abc"),
                None,
                &updated_at,
            )
            .await
            .expect("update to Running");
        assert!(ok);

        let running = store
            .get_session(&session.id)
            .await
            .expect("get")
            .expect("not found");
        assert_eq!(running.state, SessionState::Running);
        assert_eq!(running.container_id.as_deref(), Some("container-abc"));

        // Soft-delete.
        let del = store
            .soft_delete_session(&session.id)
            .await
            .expect("delete");
        assert!(del);

        let list = store
            .list_sessions_for_project(&project.id)
            .await
            .expect("list");
        assert!(list.is_empty());
    }

    #[tokio::test]
    async fn session_optimistic_concurrency() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "oc-proj".into(),
                git_url: "https://example.com/oc".into(),
                local_path: "/tmp/oc".into(),
            })
            .await
            .expect("project");

        let session = store
            .create_session(CreateSession {
                project_id: project.id.clone(),
                branch: "oc".into(),
                image: "ubuntu:24.04".into(),
                profile: "default".into(),
                work_dir: String::new(),
            })
            .await
            .expect("session");

        let ts = session.updated_at.to_rfc3339();

        let ok = store
            .update_session_state(&session.id, SessionState::Running, None, None, &ts)
            .await
            .expect("first update");
        assert!(ok);

        // Stale timestamp — must return false.
        let conflict = store
            .update_session_state(&session.id, SessionState::Stopped, None, None, &ts)
            .await
            .expect("second update");
        assert!(!conflict, "stale update must return false");

        let final_session = store
            .get_session(&session.id)
            .await
            .expect("get")
            .expect("not found");
        assert_eq!(final_session.state, SessionState::Running);
        assert!(final_session.stopped_at.is_none());
    }

    #[tokio::test]
    async fn soft_delete_is_idempotent() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "idem".into(),
                git_url: "https://example.com/idem".into(),
                local_path: "/tmp/idem".into(),
            })
            .await
            .expect("create");

        let first = store
            .soft_delete_project(&project.id)
            .await
            .expect("first delete");
        assert!(first);

        let second = store
            .soft_delete_project(&project.id)
            .await
            .expect("second delete");
        assert!(!second, "already-deleted row must return false");
    }

    // -----------------------------------------------------------------------
    // Terminal CRUD
    // -----------------------------------------------------------------------

    #[tokio::test]
    async fn terminal_crud() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "t-proj".into(),
                git_url: "https://example.com/t".into(),
                local_path: "/tmp/t".into(),
            })
            .await
            .expect("project");

        let session = store
            .create_session(CreateSession {
                project_id: project.id.clone(),
                branch: "main".into(),
                image: "ubuntu:24.04".into(),
                profile: "default".into(),
                work_dir: String::new(),
            })
            .await
            .expect("session");

        // Create a terminal.
        let terminal = store
            .create_terminal(CreateTerminal {
                session_id: session.id.clone(),
                profile: "default".into(),
            })
            .await
            .expect("create terminal");

        assert_eq!(terminal.session_id, session.id);
        assert_eq!(terminal.profile, "default");
        assert_eq!(terminal.state, TerminalState::Starting);

        // Get by id.
        let fetched = store
            .get_terminal(&terminal.id)
            .await
            .expect("get")
            .expect("not found");
        assert_eq!(fetched.id, terminal.id);

        // List by session.
        let list = store
            .list_terminals_for_session(&session.id)
            .await
            .expect("list");
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].id, terminal.id);

        // Update state.
        let ts = terminal.updated_at.to_rfc3339();
        let ok = store
            .update_terminal_state(&terminal.id, TerminalState::Running, &ts)
            .await
            .expect("update");
        assert!(ok);

        let running = store
            .get_terminal(&terminal.id)
            .await
            .expect("get")
            .expect("not found");
        assert_eq!(running.state, TerminalState::Running);

        // Delete.
        let del = store.delete_terminal(&terminal.id).await.expect("delete");
        assert!(del);

        let gone = store.get_terminal(&terminal.id).await.expect("get");
        assert!(gone.is_none());
    }

    #[tokio::test]
    async fn get_first_terminal_for_session() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "ft-proj".into(),
                git_url: "https://example.com/ft".into(),
                local_path: "/tmp/ft".into(),
            })
            .await
            .expect("project");

        let session = store
            .create_session(CreateSession {
                project_id: project.id.clone(),
                branch: "main".into(),
                image: "ubuntu:24.04".into(),
                profile: "default".into(),
                work_dir: String::new(),
            })
            .await
            .expect("session");

        // No terminals yet.
        let none = store
            .get_first_terminal_for_session(&session.id)
            .await
            .expect("get first");
        assert!(none.is_none());

        // Create two terminals.
        let t1 = store
            .create_terminal(CreateTerminal {
                session_id: session.id.clone(),
                profile: "first".into(),
            })
            .await
            .expect("t1");

        // Small delay to ensure different created_at timestamps.
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        let _t2 = store
            .create_terminal(CreateTerminal {
                session_id: session.id.clone(),
                profile: "second".into(),
            })
            .await
            .expect("t2");

        let first = store
            .get_first_terminal_for_session(&session.id)
            .await
            .expect("get first")
            .expect("should exist");
        assert_eq!(first.id, t1.id);
        assert_eq!(first.profile, "first");
    }

    #[tokio::test]
    async fn terminal_optimistic_concurrency() {
        let store = test_store().await;
        let project = store
            .create_project(CreateProject {
                name: "toc".into(),
                git_url: "https://example.com/toc".into(),
                local_path: "/tmp/toc".into(),
            })
            .await
            .expect("project");

        let session = store
            .create_session(CreateSession {
                project_id: project.id.clone(),
                branch: "main".into(),
                image: "ubuntu:24.04".into(),
                profile: "default".into(),
                work_dir: String::new(),
            })
            .await
            .expect("session");

        let terminal = store
            .create_terminal(CreateTerminal {
                session_id: session.id.clone(),
                profile: "default".into(),
            })
            .await
            .expect("terminal");

        let ts = terminal.updated_at.to_rfc3339();

        // First update succeeds.
        let ok = store
            .update_terminal_state(&terminal.id, TerminalState::Running, &ts)
            .await
            .expect("first update");
        assert!(ok);

        // Second update with stale timestamp fails.
        let conflict = store
            .update_terminal_state(&terminal.id, TerminalState::Stopped, &ts)
            .await
            .expect("second update");
        assert!(!conflict, "stale update must return false");

        let final_terminal = store
            .get_terminal(&terminal.id)
            .await
            .expect("get")
            .expect("not found");
        assert_eq!(final_terminal.state, TerminalState::Running);
    }
}