artifactize 0.6.0

use std::{path::Path, time::Duration};

use rusqlite::params;
use serde::Serialize;
use serde_json::{Value, json};
use tokio_rusqlite::Connection;

use super::{
    DATABASE, Execution, HumanClaim, Request,
    receipts::{Error, check_files, schema_initialized},
};
use crate::workspace::{canonical_target, outside_workspace};

#[derive(Debug, Clone, Serialize)]
pub struct RequestView {
    #[serde(flatten)]
    pub request: Request,
    pub claim: Option<HumanClaim>,
    pub execution: Option<Execution>,
    pub definition: Option<Value>,
}

#[derive(Default)]
struct Filter<'a> {
    run: Option<&'a str>,
    id: Option<&'a str>,
    repo: Option<&'a Path>,
    waiting: bool,
    session: Option<&'a str>,
}

pub async fn read_requests(state: &Path, run: Option<&str>) -> Result<Vec<RequestView>, String> {
    read(
        state,
        Filter {
            run,
            ..Filter::default()
        },
    )
    .await
}

pub async fn read_request(state: &Path, id: &str) -> Result<RequestView, String> {
    read(
        state,
        Filter {
            id: Some(id),
            ..Filter::default()
        },
    )
    .await?
    .pop()
    .ok_or_else(|| "Review request not found.".into())
}

/// The request whose Agent review ran as session `id`, if this state has it.
pub async fn read_session_request(state: &Path, id: &str) -> Result<Option<RequestView>, String> {
    Ok(read(
        state,
        Filter {
            session: Some(id),
            ..Filter::default()
        },
    )
    .await?
    .pop())
}

/// WAITING_HUMAN requests, newest Run first, from the canonical `repo`'s Runs or every Run.
pub async fn read_waiting(state: &Path, repo: Option<&Path>) -> Result<Vec<RequestView>, String> {
    read(
        state,
        Filter {
            repo,
            waiting: true,
            ..Filter::default()
        },
    )
    .await
}

async fn read(state: &Path, filter: Filter<'_>) -> Result<Vec<RequestView>, String> {
    let state = canonical_target(state).map_err(|e| e.to_string())?;
    let repo = filter
        .repo
        .map(canonical_target)
        .transpose()
        .map_err(|e| e.to_string())?;
    if let Some(repo) = &repo {
        outside_workspace(repo, &state).map_err(|e| e.to_string())?;
    }
    check_files(&state)?;
    if !state
        .join(DATABASE)
        .try_exists()
        .map_err(|e| e.to_string())?
    {
        return Ok(Vec::new());
    }
    let connection = Connection::open_with_flags(
        state.join(DATABASE),
        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
    )
    .await
    .map_err(|e| e.to_string())?;
    let run = filter.run.map(str::to_owned);
    let id = filter.id.map(str::to_owned);
    let repo = repo.map(|repo| repo.to_string_lossy().into_owned());
    let waiting = filter.waiting;
    let session = filter.session.map(str::to_owned);
    connection.call(move |db| -> Result<_, Error> {
        db.busy_timeout(Duration::from_secs(5))?;
        let transaction = db.transaction()?;
        if !schema_initialized(&transaction)? {
            return Ok(Vec::new());
        }
        let views = {
            // A follower shows the claim of the waiting request whose execution it follows.
            let mut statement = transaction.prepare("SELECT q.data,r.repo,e.data,h.id,h.claimed_by,h.claimed_at,
                json_extract(r.data,'$.definitions.artifacts'),
                (SELECT value FROM json_each(r.data,'$.definitions.evals') WHERE json_extract(value,'$.id')=q.eval_id)
                FROM requests q JOIN runs r ON r.id=q.run_id
                LEFT JOIN executions e ON e.id=q.execution_id
                LEFT JOIN requests h ON h.id=json_extract(e.data,'$.provenance.requestId') AND h.claimed_by IS NOT NULL
                WHERE (?1 IS NULL OR q.run_id=?1) AND (?2 IS NULL OR q.id=?2)
                AND (?3 IS NULL OR r.repo=?3) AND (NOT ?4 OR q.status='WAITING_HUMAN')
                AND (?5 IS NULL OR json_extract(q.data,'$.sessionId')=?5)
                ORDER BY r.rowid DESC,q.ordinal")?;
            statement.query_map(params![run, id, repo, waiting, session], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, Option<String>>(2)?,
                    row.get::<_, Option<String>>(3)?, row.get::<_, Option<String>>(4)?, row.get::<_, Option<String>>(5)?,
                    row.get::<_, Option<String>>(6)?, row.get::<_, Option<String>>(7)?))
            })?.map(|row| {
                let (data, repo, execution, request_id, reviewer, claimed_at, artifacts, eval) = row?;
                outside_workspace(Path::new(&repo), &state).map_err(|e| Error::Invalid(e.to_string()))?;
                let request: Request = serde_json::from_str(&data)?;
                let artifacts: Option<Value> = artifacts.map(|data| serde_json::from_str(&data)).transpose()?;
                let eval: Option<Value> = eval.map(|data| serde_json::from_str(&data)).transpose()?;
                let definition = eval.zip(artifacts.as_ref().and_then(|artifacts| artifacts.get(&request.target)))
                    .map(|(eval, artifact)| json!({"eval":eval,"artifact":artifact}));
                Ok(RequestView {
                    request,
                    definition,
                    claim: request_id.zip(reviewer).zip(claimed_at).map(|((request_id, reviewer), claimed_at)| HumanClaim {request_id, reviewer, claimed_at}),
                    execution: execution.map(|data| serde_json::from_str(&data)).transpose()?,
                })
            }).collect::<Result<Vec<_>, Error>>()?
        };
        transaction.commit()?;
        Ok(views)
    }).await.map_err(|e| e.to_string())
}