7.6 Checkpoints on Loco's database connection

// app/src/engine/session_storage.rs
// graph-flow's SessionStorage trait (source-verified): save(Session), get(&str) -> Option<Session>, delete(&str).
// Session derives Serialize/Deserialize and carries an optimistic-locking `version` that storage increments.
use async_trait::async_trait;
use graph_flow::{GraphError, Session, SessionStorage};
use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, Statement};

pub struct SeaOrmSessionStorage { db: DatabaseConnection }
impl SeaOrmSessionStorage { pub fn new(db: DatabaseConnection) -> Self { Self { db } } }

fn se<E: std::fmt::Display>(e: E) -> GraphError { GraphError::ContextError(e.to_string()) }   // TODO(impl): use the storage error variant

#[async_trait]
impl SessionStorage for SeaOrmSessionStorage {
    async fn save(&self, session: Session) -> graph_flow::Result<()> {
        let data = serde_json::to_value(&session).map_err(se)?;
        // Insert, or update only if nobody else saved since we loaded (optimistic lock).
        let stmt = Statement::from_sql_and_values(DbBackend::Postgres,
            "INSERT INTO workflow_sessions (id, graph_id, current_task_id, data, version, updated_at)
             VALUES ($1, $2, $3, $4, 1, now())
             ON CONFLICT (id) DO UPDATE
               SET graph_id = EXCLUDED.graph_id, current_task_id = EXCLUDED.current_task_id,
                   data = EXCLUDED.data, version = workflow_sessions.version + 1, updated_at = now()
             WHERE workflow_sessions.version = $5",
            [session.id.clone().into(), session.graph_id.clone().into(), session.current_task_id.clone().into(),
             data.into(), (session.version as i64).into()]);
        let res = self.db.execute_raw(stmt).await.map_err(se)?;      // VERIFY: `execute_raw` (SeaORM 2.0) vs `execute`
        if res.rows_affected() == 0 { return Err(se("session version conflict")); }
        Ok(())
    }

    async fn get(&self, id: &str) -> graph_flow::Result<Option<Session>> {
        let stmt = Statement::from_sql_and_values(DbBackend::Postgres,
            "SELECT data, version FROM workflow_sessions WHERE id = $1", [id.into()]);
        let Some(row) = self.db.query_one_raw(stmt).await.map_err(se)? else { return Ok(None) };   // VERIFY name
        let data: serde_json::Value = row.try_get("", "data").map_err(se)?;
        let version: i64 = row.try_get("", "version").map_err(se)?;
        let mut s: Session = serde_json::from_value(data).map_err(se)?;
        s.version = version as u64;
        Ok(Some(s))
    }

    async fn delete(&self, id: &str) -> graph_flow::Result<()> {
        let stmt = Statement::from_sql_and_values(DbBackend::Postgres, "DELETE FROM workflow_sessions WHERE id = $1", [id.into()]);
        self.db.execute_raw(stmt).await.map_err(se)?;
        Ok(())
    }
}