artifactize 0.10.0

use std::path::Path;

use rusqlite::params;
use serde::Serialize;

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<RequestDefinition>,
}

#[derive(Debug, Clone, Serialize)]
pub struct RequestDefinition {
    pub eval: super::definitions::Eval,
    pub artifact: super::definitions::Artifact,
}

#[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())
}

/// The request that owns `view`'s shared execution: a follower of another Run joins the
/// original waiting request, which alone records the Human definition (tools and owner
/// schemas), the claim and the result. `view` itself when it is the original.
pub async fn read_original(state: &Path, view: &RequestView) -> Result<RequestView, String> {
    let original = view
        .execution
        .as_ref()
        .map(|execution| &execution.provenance.request_id)
        .or_else(|| {
            view.request
                .provenance
                .as_ref()
                .map(|source| &source.request_id)
        });
    match original.filter(|id| *id != &view.request.id) {
        Some(id) => read_request(state, id).await,
        None => Ok(view.clone()),
    }
}

/// One entry per waiting Human sign-off: followers collapse onto the original request that
/// owns their shared execution, at the position of the first entry that names it.
pub async fn read_waiting_originals(
    state: &Path,
    repo: Option<&Path>,
) -> Result<Vec<RequestView>, String> {
    let waiting = read_waiting(state, repo).await?;
    let mut originals: Vec<RequestView> = Vec::with_capacity(waiting.len());
    for view in &waiting {
        let source = view
            .execution
            .as_ref()
            .map_or(&view.request.id, |execution| {
                &execution.provenance.request_id
            });
        if originals.iter().any(|seen| &seen.request.id == source) {
            continue;
        }
        let original = match waiting.iter().find(|other| &other.request.id == source) {
            Some(original) => Some(original.clone()),
            None => read_original(state, view).await.ok(),
        };
        // A follower whose original is gone or no longer waits stays listed as itself.
        let original = original
            .filter(|original| {
                original.request.status() == crate::types::RequestStatus::WaitingHuman
            })
            .unwrap_or_else(|| view.clone());
        if !originals
            .iter()
            .any(|seen| seen.request.id == original.request.id)
        {
            originals.push(original);
        }
    }
    Ok(originals)
}

/// 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::parse::<crate::types::RunId>)
        .transpose()?;
    let id = filter
        .id
        .map(str::parse::<crate::types::RequestId>)
        .transpose()?;
    let repo = repo.map(|repo| crate::platform::path_text(&repo));
    let waiting = filter.waiting;
    let session = filter
        .session
        .map(str::parse::<crate::types::SessionId>)
        .transpose()?;
    connection
        .call(move |db| -> Result<_, Error> {
            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
            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),r.data
                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<crate::types::RequestId>>(3)?,
                            row.get::<_, Option<crate::types::ReviewerId>>(4)?,
                            row.get::<_, Option<crate::types::Timestamp>>(5)?,
                            row.get::<_, Option<String>>(6)?,
                            row.get::<_, Option<String>>(7)?,
                            row.get::<_, String>(8)?,
                        ))
                    })?
                    .map(|row| {
                        let (
                            data,
                            repo,
                            execution,
                            request_id,
                            reviewer,
                            claimed_at,
                            artifacts,
                            eval,
                            run_data,
                        ) = row?;
                        if super::unreadable::evidence::<super::Run>("run", &run_data).is_none() {
                            return Ok(None);
                        }
                        outside_workspace(Path::new(&repo), &state)
                            .map_err(|e| Error::Invalid(e.to_string()))?;
                        let Some(request) = super::unreadable::evidence::<Request>("request", &data) else {
                            return Ok(None);
                        };
                        let artifacts: Option<std::collections::BTreeMap<crate::types::ArtifactName, super::definitions::Artifact>> = artifacts
                            .map(|data| serde_json::from_str(&data))
                            .transpose()?;
                        let eval: Option<super::definitions::Eval> =
                            eval.map(|data| serde_json::from_str(&data)).transpose()?;
                        let definition = eval
                            .zip(
                                artifacts
                                    .as_ref()
                                    .and_then(|artifacts| artifacts.get(request.target.as_str())),
                            )
                            .map(|(eval, artifact)| RequestDefinition { eval, artifact: artifact.clone() });
                        Ok(Some(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.and_then(|data| super::unreadable::evidence("execution", &data)),
                        }))
                    })
                    .filter_map(|result| result.transpose())
                    .collect::<Result<Vec<_>, Error>>()?
            };
            transaction.commit()?;
            Ok(views)
        })
        .await
        .map_err(|e| e.to_string())
}