sbol-db-server 0.1.8

use std::io;
use std::time::Duration;

use axum::body::{Body, Bytes};
use axum::extract::{Path, Query, State};
use axum::http::header::{ACCEPT, CONTENT_TYPE};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::Json;
use futures::stream;
use sbol_db_core::{
    Direction, GraphId, GraphRecord, ImportReport, IriString, JobId, NeighborhoodQuery,
    NeighborhoodResult, ObjectId, SbolObjectRecord, SerializationFormat,
};
use sbol_db_rdf::triples_to_rdf;
use sbol_db_search_sdk::IndexMutationSource;
use sbol_db_sparql::{ResultFormat, SparqlOptions};
use sbol_db_storage::{
    BatchSequenceMatch, EnqueueOutcome, GraphWriteMode, ImportInput, ImportOverwrite, JobAttempt,
    JobLogRecord, JobStatus, ListJobsFilter, ListObjectsFilter, NewJob, OntologyLoadReport,
    OntologyRecord, OntologyTermRecord, SbolJob, SequenceMatch, SequenceSearchOptions,
    TextSearchQuery,
};
use serde::Deserialize;
use serde_json::json;
use uuid::Uuid;

use crate::error::ApiError;
use crate::export;
use crate::AppState;

const READYZ_TIMEOUT: Duration = Duration::from_secs(1);
const GRAPH_STREAM_PAGE: usize = 25_000;
const BUFFERED_GRAPH_LIMIT: usize = 100_000;

pub async fn healthz() -> &'static str {
    "ok"
}

pub async fn readyz(State(state): State<AppState>) -> impl IntoResponse {
    if !state.metrics.tls_ready_for_traffic() {
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({
                "status": "not_ready",
                "reason": "TLS certificate has not been deployed"
            })),
        );
    }
    if let Err(reason) = state.metrics.data_disk_ready_for_traffic() {
        return (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({
                "status": "not_ready",
                "reason": reason
            })),
        );
    }
    match tokio::time::timeout(READYZ_TIMEOUT, state.service.ping()).await {
        Ok(Ok(())) => (StatusCode::OK, Json(json!({ "status": "ready" }))),
        Ok(Err(err)) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({ "status": "not_ready", "reason": err.to_string() })),
        ),
        Err(_) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({ "status": "not_ready", "reason": "database probe timed out" })),
        ),
    }
}

#[derive(Deserialize)]
pub struct CreateDocumentParams {
    pub format: Option<String>,
    pub namespace: Option<String>,
    pub source_uri: Option<String>,
    pub document_iri: Option<String>,
    pub name: Option<String>,
    pub description: Option<String>,
    pub created_by: Option<String>,
    /// 0 = fail on an existing `document_iri` (default), 1 = replace it,
    /// 2 = merge into it. Replace and merge require `document_iri`.
    pub overwrite: Option<u8>,
}

/// Map the numeric `overwrite` parameter to an [`ImportOverwrite`], enforcing
/// that replace and merge name the graph to act on via `document_iri`.
fn resolve_overwrite(
    overwrite: Option<u8>,
    document_iri: Option<&IriString>,
) -> Result<ImportOverwrite, ApiError> {
    let mode = match overwrite.unwrap_or(0) {
        0 => ImportOverwrite::Fail,
        1 => ImportOverwrite::Replace,
        2 => ImportOverwrite::Merge,
        other => return Err(ApiError::BadRequest(format!("invalid overwrite: {other}"))),
    };
    if mode != ImportOverwrite::Fail && document_iri.is_none() {
        return Err(ApiError::BadRequest(
            "overwrite requires document_iri to identify the graph to replace or merge".to_owned(),
        ));
    }
    Ok(mode)
}

pub async fn create_graph(
    State(state): State<AppState>,
    Query(params): Query<CreateDocumentParams>,
    headers: HeaderMap,
    body: String,
) -> Result<Json<ImportReport>, ApiError> {
    let format = resolve_format(params.format.as_deref(), &headers)?;
    let document_iri = params
        .document_iri
        .map(IriString::new)
        .transpose()
        .map_err(|e| ApiError::BadRequest(e.to_string()))?;
    let overwrite = resolve_overwrite(params.overwrite, document_iri.as_ref())?;
    let report = state
        .service
        .import_document(ImportInput {
            body,
            format,
            namespace: params.namespace,
            source_uri: params.source_uri,
            document_iri,
            created_by: params.created_by,
            name: params.name,
            description: params.description,
            overwrite,
        })
        .await?;
    state
        .app
        .schedule_search_reconciliation(IndexMutationSource::RestImport)
        .await?;
    Ok(Json(report))
}

const BULK_IMPORT_MAX_DOCUMENTS: usize = 100;

#[derive(Deserialize)]
pub struct BulkImportBody {
    pub graphs: Vec<BulkImportItem>,
}

#[derive(Deserialize)]
pub struct BulkImportItem {
    pub body: String,
    pub format: String,
    #[serde(default)]
    pub namespace: Option<String>,
    #[serde(default)]
    pub source_uri: Option<String>,
    #[serde(default)]
    pub document_iri: Option<String>,
    #[serde(default)]
    pub name: Option<String>,
    #[serde(default)]
    pub description: Option<String>,
    #[serde(default)]
    pub created_by: Option<String>,
}

#[derive(serde::Serialize)]
pub struct BulkImportResponse {
    pub imported: usize,
    pub reports: Vec<ImportReport>,
}

/// Atomic bulk import. The entire batch commits or none of it does — a
/// failure on document N rolls back documents 1..N and returns the error
/// unchanged from `import_document`. The body cap is the server's
/// `SBOL_DB_MAX_BODY_BYTES`; the per-call document cap is enforced here.
pub async fn create_graphs_bulk(
    State(state): State<AppState>,
    Json(body): Json<BulkImportBody>,
) -> Result<Json<BulkImportResponse>, ApiError> {
    if body.graphs.is_empty() {
        return Ok(Json(BulkImportResponse {
            imported: 0,
            reports: Vec::new(),
        }));
    }
    if body.graphs.len() > BULK_IMPORT_MAX_DOCUMENTS {
        return Err(ApiError::BadRequest(format!(
            "request exceeds maximum of {BULK_IMPORT_MAX_DOCUMENTS} documents per call"
        )));
    }
    let mut inputs = Vec::with_capacity(body.graphs.len());
    for item in body.graphs {
        let format = parse_format(&item.format)
            .ok_or_else(|| ApiError::BadRequest(format!("unknown format: {}", item.format)))?;
        let document_iri = item
            .document_iri
            .map(IriString::new)
            .transpose()
            .map_err(|e| ApiError::BadRequest(e.to_string()))?;
        inputs.push(ImportInput {
            body: item.body,
            format,
            namespace: item.namespace,
            source_uri: item.source_uri,
            document_iri,
            created_by: item.created_by,
            name: item.name,
            description: item.description,
            overwrite: ImportOverwrite::Fail,
        });
    }
    let reports = state.service.import_documents(inputs).await?;
    if !reports.is_empty() {
        state
            .app
            .schedule_search_reconciliation(IndexMutationSource::RestImport)
            .await?;
    }
    Ok(Json(BulkImportResponse {
        imported: reports.len(),
        reports,
    }))
}

pub async fn get_graph(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Json<GraphRecord>, ApiError> {
    let doc = state
        .service
        .get_graph(GraphId(id))
        .await?
        .ok_or_else(|| ApiError::NotFound(format!("document {id}")))?;
    Ok(Json(doc))
}

/// Delete one document graph by its surrogate id. Deletion is at graph
/// granularity: a submission is one graph, so this drops the graph's triples,
/// its derived objects, and its registry row.
pub async fn delete_graph(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<StatusCode, ApiError> {
    if state.service.delete_graph(GraphId(id)).await? {
        state
            .app
            .schedule_search_reconciliation(IndexMutationSource::RestGraphDelete)
            .await?;
        Ok(StatusCode::NO_CONTENT)
    } else {
        Err(ApiError::NotFound(format!("document {id}")))
    }
}

#[derive(Deserialize)]
pub struct DeleteGraphByDocParams {
    pub document_iri: String,
}

/// Delete one document graph by its document IRI (resolving to the surrogate id
/// first). Same graph-granular semantics as [`delete_graph`].
pub async fn delete_graph_by_document_iri(
    State(state): State<AppState>,
    Query(params): Query<DeleteGraphByDocParams>,
) -> Result<StatusCode, ApiError> {
    let iri = IriString::new(params.document_iri.clone())
        .map_err(|e| ApiError::BadRequest(e.to_string()))?;
    let Some(id) = state.service.graph_id_by_document_iri(iri.as_str()).await? else {
        return Err(ApiError::NotFound(format!(
            "document {}",
            params.document_iri
        )));
    };
    state.service.delete_graph(id).await?;
    state
        .app
        .schedule_search_reconciliation(IndexMutationSource::RestGraphDelete)
        .await?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Deserialize)]
pub struct GetObjectParams {
    pub iri: String,
}

pub async fn get_object_by_iri(
    State(state): State<AppState>,
    Query(params): Query<GetObjectParams>,
) -> Result<Json<SbolObjectRecord>, ApiError> {
    let obj = state
        .service
        .get_object_by_iri(&params.iri)
        .await?
        .ok_or_else(|| ApiError::NotFound(format!("object {}", params.iri)))?;
    Ok(Json(obj))
}

/// Hard cap on the size of a `POST /objects/lookup` request, applied
/// before hitting Postgres. Matches the spirit of `max_rows` on SPARQL —
/// keep batch APIs bounded even when the body limit allows more.
const LOOKUP_MAX_IRIS: usize = 1000;

#[derive(Deserialize)]
pub struct LookupObjectsBody {
    pub iris: Vec<String>,
}

#[derive(serde::Serialize)]
pub struct LookupObjectsResponse {
    pub found: Vec<SbolObjectRecord>,
    pub missing: Vec<String>,
}

pub async fn lookup_objects(
    State(state): State<AppState>,
    Json(body): Json<LookupObjectsBody>,
) -> Result<Json<LookupObjectsResponse>, ApiError> {
    if body.iris.is_empty() {
        return Ok(Json(LookupObjectsResponse {
            found: Vec::new(),
            missing: Vec::new(),
        }));
    }
    if body.iris.len() > LOOKUP_MAX_IRIS {
        return Err(ApiError::BadRequest(format!(
            "request exceeds maximum of {LOOKUP_MAX_IRIS} IRIs per call"
        )));
    }
    for iri in &body.iris {
        IriString::new(iri.clone()).map_err(|e| ApiError::BadRequest(e.to_string()))?;
    }
    let refs: Vec<&str> = body.iris.iter().map(String::as_str).collect();
    let found = state.service.get_objects_by_iris(&refs).await?;
    let found_set: std::collections::HashSet<&str> = found.iter().map(|r| r.iri.as_str()).collect();
    let missing: Vec<String> = body
        .iris
        .iter()
        .filter(|iri| !found_set.contains(iri.as_str()))
        .cloned()
        .collect();
    Ok(Json(LookupObjectsResponse { found, missing }))
}

const LIST_MAX_LIMIT: u32 = 5000;
const LIST_DEFAULT_LIMIT: u32 = 1000;

#[derive(Deserialize)]
pub struct ListObjectsParams {
    #[serde(default)]
    pub sbol_class: Option<String>,
    #[serde(default)]
    pub role: Option<String>,
    #[serde(default)]
    pub graph_id: Option<Uuid>,
    #[serde(default)]
    pub iri_contains: Option<String>,
    #[serde(default)]
    pub after: Option<String>,
    #[serde(default = "default_list_limit")]
    pub limit: u32,
}

fn default_list_limit() -> u32 {
    LIST_DEFAULT_LIMIT
}

#[derive(serde::Serialize)]
pub struct ListObjectsResponse {
    pub objects: Vec<SbolObjectRecord>,
    /// Cursor for the next page (last `iri` of the current page) when the
    /// page filled to `limit`; `None` when the listing has been exhausted.
    pub next_cursor: Option<String>,
}

pub async fn list_objects(
    State(state): State<AppState>,
    Query(params): Query<ListObjectsParams>,
) -> Result<Json<ListObjectsResponse>, ApiError> {
    let limit = params.limit.clamp(1, LIST_MAX_LIMIT);
    let filter = ListObjectsFilter {
        sbol_class: params.sbol_class,
        role: params.role,
        graph_id: params.graph_id.map(GraphId),
        iri_contains: params.iri_contains,
        after_iri: params.after,
        limit,
    };
    let objects = state.service.list_objects(&filter).await?;
    let next_cursor = if objects.len() as u32 >= limit {
        objects.last().map(|o| o.iri.as_str().to_owned())
    } else {
        None
    };
    Ok(Json(ListObjectsResponse {
        objects,
        next_cursor,
    }))
}

pub(crate) const SEARCH_DEFAULT_LIMIT: i64 = 50;
const SEARCH_MAX_LIMIT: i64 = 1000;

fn default_search_limit() -> i64 {
    SEARCH_DEFAULT_LIMIT
}

#[derive(Deserialize)]
pub struct TextSearchParams {
    pub q: String,
    #[serde(default)]
    pub object_type: Option<String>,
    #[serde(default)]
    pub property_uri: Option<String>,
    #[serde(default)]
    pub offset: i64,
    #[serde(default = "default_search_limit")]
    pub limit: i64,
}

#[derive(serde::Serialize)]
pub struct TextSearchResponse {
    pub objects: Vec<SbolObjectRecord>,
    pub total: i64,
    pub offset: i64,
    pub limit: i64,
}

/// Substring search over the derived object view. `q` matches an object's
/// name/display_id/description (or, with `property_uri`, one predicate's
/// literal value); `object_type` restricts by SBOL class. `limit=0` returns
/// the total match count with no rows, which backs a count-only query.
pub async fn text_search(
    State(state): State<AppState>,
    Query(params): Query<TextSearchParams>,
) -> Result<Json<TextSearchResponse>, ApiError> {
    if params.q.trim().is_empty() {
        return Err(ApiError::BadRequest("q is required".to_owned()));
    }
    let limit = params.limit.clamp(0, SEARCH_MAX_LIMIT);
    let offset = params.offset.max(0);
    let (objects, total) = state
        .service
        .search_objects(&TextSearchQuery {
            text: params.q,
            sbol_class: params.object_type,
            property_uri: params.property_uri,
            offset,
            limit,
        })
        .await?;
    Ok(Json(TextSearchResponse {
        objects,
        total,
        offset,
        limit,
    }))
}

/// Parse the `version` query parameter, returning whether the caller asked for
/// SBOL2 output. Absent or `sbol3` means the native SBOL3 view.
fn wants_sbol2(version: Option<&str>) -> Result<bool, ApiError> {
    match version.unwrap_or("sbol3") {
        "sbol3" | "3" => Ok(false),
        "sbol2" | "2" => Ok(true),
        other => Err(ApiError::BadRequest(format!("unknown version: {other}"))),
    }
}

#[derive(Deserialize)]
pub struct ExportObjectParams {
    pub format: Option<String>,
    #[serde(default)]
    pub version: Option<String>,
}

pub async fn export_object(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
    Query(params): Query<ExportObjectParams>,
) -> Result<impl IntoResponse, ApiError> {
    let iri = state
        .service
        .get_object_iri_by_id(ObjectId(id))
        .await?
        .ok_or_else(|| ApiError::NotFound(format!("object {id}")))?;
    let raw_format = params.format.as_deref().unwrap_or("turtle");
    let format = parse_export_format(raw_format)
        .ok_or_else(|| ApiError::BadRequest(format!("unknown format: {raw_format}")))?;
    let sbol2 = wants_sbol2(params.version.as_deref())?;
    let body = export::export_subject_rdf(state.service.as_ref(), &iri, format, sbol2).await?;
    let content_type = match format {
        SerializationFormat::Turtle => "text/turtle",
        SerializationFormat::JsonLd => "application/ld+json",
        SerializationFormat::NTriples => "application/n-triples",
        SerializationFormat::RdfXml => "application/rdf+xml",
        _ => "text/plain",
    };
    Ok(([(CONTENT_TYPE, content_type)], body))
}

#[derive(Deserialize)]
pub struct NeighborhoodParams {
    pub iri: String,
    #[serde(default = "default_depth")]
    pub depth: u32,
    #[serde(default = "default_direction")]
    pub direction: String,
    /// Comma-separated list of predicate IRIs. serde_urlencoded (axum's
    /// default Query parser) doesn't support repeated keys, so callers send
    /// `?predicates=p1,p2`.
    #[serde(default)]
    pub predicates: Option<String>,
    #[serde(default = "default_max_nodes")]
    pub max_nodes: u32,
    #[serde(default)]
    pub literals: bool,
}

fn parse_predicates(value: Option<String>) -> Vec<IriString> {
    value
        .map(|s| {
            s.split(',')
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .map(IriString::unchecked)
                .collect()
        })
        .unwrap_or_default()
}

fn default_depth() -> u32 {
    2
}
fn default_direction() -> String {
    "forward".to_owned()
}
fn default_max_nodes() -> u32 {
    2048
}

fn parse_direction(s: &str) -> Result<Direction, ApiError> {
    match s.to_ascii_lowercase().as_str() {
        "forward" | "out" => Ok(Direction::Forward),
        "backward" | "back" | "in" => Ok(Direction::Backward),
        "both" | "either" => Ok(Direction::Both),
        other => Err(ApiError::BadRequest(format!("unknown direction: {other}"))),
    }
}

fn build_neighborhood_query(p: NeighborhoodParams) -> Result<NeighborhoodQuery, ApiError> {
    let root = IriString::new(p.iri).map_err(|e| ApiError::BadRequest(e.to_string()))?;
    let direction = parse_direction(&p.direction)?;
    Ok(NeighborhoodQuery {
        root_iri: root,
        depth: p.depth,
        direction,
        predicate_allowlist: parse_predicates(p.predicates),
        max_nodes: Some(p.max_nodes),
        include_literals: p.literals,
    })
}

pub async fn neighborhood(
    State(state): State<AppState>,
    Query(params): Query<NeighborhoodParams>,
) -> Result<Json<NeighborhoodResult>, ApiError> {
    let query = build_neighborhood_query(params)?;
    let result = state.service.walk(&query).await?;
    Ok(Json(result))
}

#[derive(Deserialize)]
pub struct NeighborhoodRdfParams {
    pub iri: String,
    #[serde(default = "default_depth")]
    pub depth: u32,
    #[serde(default = "default_direction")]
    pub direction: String,
    #[serde(default)]
    pub predicates: Option<String>,
    #[serde(default = "default_max_nodes")]
    pub max_nodes: u32,
    #[serde(default = "default_literals_for_rdf")]
    pub literals: bool,
    #[serde(default = "default_rdf_format")]
    pub format: String,
    #[serde(default)]
    pub version: Option<String>,
}

fn default_rdf_format() -> String {
    "turtle".to_owned()
}
fn default_literals_for_rdf() -> bool {
    // RDF export usually wants literals so the dump is faithful.
    true
}

pub async fn neighborhood_rdf(
    State(state): State<AppState>,
    Query(params): Query<NeighborhoodRdfParams>,
) -> Result<impl IntoResponse, ApiError> {
    let format = parse_export_format(&params.format)
        .ok_or_else(|| ApiError::BadRequest(format!("unknown format: {}", params.format)))?;
    let sbol2 = wants_sbol2(params.version.as_deref())?;
    let query = build_neighborhood_query(NeighborhoodParams {
        iri: params.iri,
        depth: params.depth,
        direction: params.direction,
        predicates: params.predicates,
        max_nodes: params.max_nodes,
        literals: params.literals,
    })?;
    let result = state.service.walk(&query).await?;
    let body = if sbol2 {
        let ntriples = sbol_db_rdf::neighborhood_to_rdf(&result, SerializationFormat::NTriples)?;
        export::downgrade_sbol3_ntriples(&ntriples, format)?
    } else {
        sbol_db_rdf::neighborhood_to_rdf(&result, format)?
    };
    let content_type = match format {
        SerializationFormat::Turtle => "text/turtle",
        SerializationFormat::JsonLd => "application/ld+json",
        SerializationFormat::NTriples => "application/n-triples",
        SerializationFormat::RdfXml => "application/rdf+xml",
        _ => "text/plain",
    };
    Ok(([(CONTENT_TYPE, content_type)], body))
}

#[derive(Deserialize, Default)]
pub struct SparqlGetParams {
    pub query: Option<String>,
    /// Override the format that would otherwise be derived from `Accept`.
    pub format: Option<String>,
    /// SPARQL 1.1 protocol `default-graph-uri`. SynBioHub sends this to scope
    /// reads to a user/public graph.
    #[serde(rename = "default-graph-uri")]
    pub default_graph_uri: Option<String>,
}

#[derive(Deserialize, Default)]
pub struct SparqlFormParams {
    pub query: Option<String>,
    pub format: Option<String>,
    #[serde(rename = "default-graph-uri")]
    pub default_graph_uri: Option<String>,
}

pub async fn sparql_get(
    State(state): State<AppState>,
    Query(params): Query<SparqlGetParams>,
    headers: HeaderMap,
) -> Result<impl IntoResponse, ApiError> {
    let query = params
        .query
        .ok_or_else(|| ApiError::BadRequest("missing ?query= parameter".to_owned()))?;
    let format = resolve_sparql_format(params.format.as_deref(), &headers)?;
    run_sparql(state, &query, format, params.default_graph_uri).await
}

pub async fn sparql_post(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: axum::body::Bytes,
) -> Result<impl IntoResponse, ApiError> {
    let content_type = headers
        .get(CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .map(|s| s.split(';').next().unwrap_or("").trim().to_owned())
        .unwrap_or_default();
    let (query_str, override_format, default_graph): (String, Option<String>, Option<String>) =
        match content_type.as_str() {
            "application/sparql-query" | "" => {
                let q = std::str::from_utf8(&body)
                    .map_err(|_| ApiError::BadRequest("query body is not UTF-8".to_owned()))?
                    .to_owned();
                (q, None, None)
            }
            "application/x-www-form-urlencoded" => {
                let form: SparqlFormParams = serde_urlencoded::from_bytes(&body)
                    .map_err(|e| ApiError::BadRequest(format!("invalid form body: {e}")))?;
                let q = form
                    .query
                    .ok_or_else(|| ApiError::BadRequest("missing query= field".to_owned()))?;
                (q, form.format, form.default_graph_uri)
            }
            other => {
                return Err(ApiError::BadRequest(format!(
                    "unsupported content-type for /sparql: {other}"
                )));
            }
        };
    let format = resolve_sparql_format(override_format.as_deref(), &headers)?;
    run_sparql(state, &query_str, format, default_graph).await
}

async fn run_sparql(
    state: AppState,
    query: &str,
    format: Option<ResultFormat>,
    default_graph: Option<String>,
) -> Result<axum::response::Response, ApiError> {
    // This is the public triplestore endpoint, so every query is evaluated as
    // SPARQL. SBOLExplorer-shaped ranked, sequence, and `/similar` requests are
    // recognized only by the dedicated compatibility listener in
    // `explorer::endpoint`; routing them here would change ordinary SynBioHub
    // store queries into fuzzy index searches when Explorer is disabled.
    //
    // The public endpoint imposes no authorization ceiling; the client's
    // `default-graph-uri` is honored under the union default graph.
    let options = SparqlOptions::default();
    let outcome = state
        .sparql
        .execute(query, format, default_graph.as_deref(), &options)
        .await?;
    let mut response =
        axum::response::Response::builder().header(CONTENT_TYPE, outcome.payload.content_type);
    if outcome.payload.truncated {
        response = response.header("X-SBOL-DB-Truncated", "true");
    }
    response
        .body(axum::body::Body::from(outcome.payload.body))
        .map_err(|e| ApiError::Domain(sbol_db_core::DomainError::Serialization(e.to_string())))
}

fn resolve_sparql_format(
    explicit: Option<&str>,
    headers: &HeaderMap,
) -> Result<Option<ResultFormat>, ApiError> {
    if let Some(s) = explicit {
        return s
            .parse::<ResultFormat>()
            .map(Some)
            .map_err(|e| ApiError::BadRequest(format!("{e}")));
    }
    let accept = headers
        .get(ACCEPT)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    for raw in accept.split(',') {
        let media = raw.split(';').next().unwrap_or("").trim();
        if media.is_empty() || media == "*/*" {
            continue;
        }
        if let Ok(f) = media.parse::<ResultFormat>() {
            return Ok(Some(f));
        }
    }
    // No explicit / no recognized Accept: let the engine pick the form's
    // natural default.
    Ok(None)
}

fn resolve_format(
    query_format: Option<&str>,
    headers: &HeaderMap,
) -> Result<SerializationFormat, ApiError> {
    if let Some(f) = query_format {
        return parse_format(f).ok_or_else(|| ApiError::BadRequest(format!("unknown format: {f}")));
    }
    let ct = headers
        .get(CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    let format = match ct.split(';').next().unwrap_or("").trim() {
        "text/turtle" | "application/x-turtle" => SerializationFormat::Turtle,
        "application/ld+json" => SerializationFormat::JsonLd,
        "application/rdf+xml" | "text/xml" => SerializationFormat::RdfXml,
        "application/n-triples" => SerializationFormat::NTriples,
        "chemical/x-genbank" | "application/x-genbank" => SerializationFormat::GenBank,
        "chemical/x-fasta" | "text/x-fasta" | "application/x-fasta" => SerializationFormat::Fasta,
        _ => {
            return Err(ApiError::BadRequest(
                "specify ?format= or set Content-Type to a supported import media type".to_owned(),
            ))
        }
    };
    Ok(format)
}

fn parse_format(s: &str) -> Option<SerializationFormat> {
    match s.to_ascii_lowercase().as_str() {
        "turtle" | "ttl" => Some(SerializationFormat::Turtle),
        "jsonld" => Some(SerializationFormat::JsonLd),
        "rdfxml" | "rdf" | "xml" => Some(SerializationFormat::RdfXml),
        "ntriples" | "nt" => Some(SerializationFormat::NTriples),
        "nquads" | "nq" => Some(SerializationFormat::NQuads),
        "trig" => Some(SerializationFormat::TriG),
        "json" => Some(SerializationFormat::Json),
        "genbank" | "gb" | "gbk" => Some(SerializationFormat::GenBank),
        "fasta" | "fa" | "fna" | "faa" => Some(SerializationFormat::Fasta),
        _ => None,
    }
}

fn parse_export_format(s: &str) -> Option<SerializationFormat> {
    match parse_format(s) {
        Some(SerializationFormat::GenBank) | Some(SerializationFormat::Fasta) | None => None,
        Some(format) => Some(format),
    }
}

#[derive(Deserialize)]
pub struct SequenceSearchParams {
    pub pattern: String,
    #[serde(default = "default_sequence_max_hits")]
    pub max_hits: u32,
    #[serde(default)]
    pub forward_only: bool,
}

fn default_sequence_max_hits() -> u32 {
    1024
}

pub async fn sequence_search(
    State(state): State<AppState>,
    Query(params): Query<SequenceSearchParams>,
) -> Result<Json<Vec<SequenceMatch>>, ApiError> {
    let matches = state
        .service
        .search(
            &params.pattern,
            SequenceSearchOptions {
                max_hits: Some(params.max_hits),
                forward_only: if params.forward_only {
                    Some(true)
                } else {
                    None
                },
            },
        )
        .await?;
    Ok(Json(matches))
}

const SEARCH_MAX_PATTERNS: usize = 256;

#[derive(Deserialize)]
pub struct SequenceSearchBody {
    pub patterns: Vec<String>,
    #[serde(default)]
    pub max_hits: Option<u32>,
    #[serde(default)]
    pub forward_only: Option<bool>,
}

pub async fn sequence_search_batch(
    State(state): State<AppState>,
    Json(body): Json<SequenceSearchBody>,
) -> Result<Json<Vec<BatchSequenceMatch>>, ApiError> {
    if body.patterns.is_empty() {
        return Ok(Json(Vec::new()));
    }
    if body.patterns.len() > SEARCH_MAX_PATTERNS {
        return Err(ApiError::BadRequest(format!(
            "request exceeds maximum of {SEARCH_MAX_PATTERNS} patterns per call"
        )));
    }
    let opts = SequenceSearchOptions {
        max_hits: Some(body.max_hits.unwrap_or(default_sequence_max_hits())),
        forward_only: match body.forward_only {
            Some(true) => Some(true),
            _ => None,
        },
    };
    let results = state.service.search_many(&body.patterns, opts).await?;
    Ok(Json(results))
}

#[derive(Deserialize)]
pub struct OntologyLoadParams {
    pub prefix: String,
    pub url: Option<String>,
    pub name: Option<String>,
}

pub async fn ontology_load(
    State(state): State<AppState>,
    Json(body): Json<OntologyLoadParams>,
) -> Result<Json<OntologyLoadReport>, ApiError> {
    let (url, name) = resolve_ontology_defaults(&body.prefix, body.url, body.name)
        .map_err(ApiError::BadRequest)?;
    let report = state
        .service
        .load_ontology_from_url(&body.prefix.to_ascii_uppercase(), &name, &url)
        .await?;
    // The lab dashboard, the SPARQL schema sidebar, and the overview
    // ontology count all depend on the loaded-ontologies list. Drop
    // their TTL caches so the next read sees the new ontology
    // immediately rather than up-to-60-seconds later.
    state.invalidate_lab_caches();
    Ok(Json(report))
}

pub async fn ontology_list(
    State(state): State<AppState>,
) -> Result<Json<Vec<OntologyRecord>>, ApiError> {
    Ok(Json(state.service.list_ontologies().await?))
}

#[derive(Deserialize)]
pub struct OntologyTermQuery {
    pub iri: String,
}

pub async fn ontology_term(
    State(state): State<AppState>,
    Query(params): Query<OntologyTermQuery>,
) -> Result<Json<OntologyTermRecord>, ApiError> {
    let canonical = canonical_term_iri(&params.iri);
    let term = state
        .service
        .get_ontology_term(&canonical)
        .await?
        .ok_or_else(|| ApiError::NotFound(format!("ontology term {}", params.iri)))?;
    Ok(Json(term))
}

#[derive(serde::Serialize)]
pub struct OntologyDescendant {
    pub iri: String,
    pub depth: i16,
}

#[derive(Deserialize)]
pub struct OntologyTermsQuery {
    pub prefix: String,
    pub q: Option<String>,
    pub limit: Option<i64>,
    pub offset: Option<i64>,
}

#[derive(serde::Serialize)]
pub struct OntologyTermsPage {
    pub prefix: String,
    pub total: i64,
    pub limit: i64,
    pub offset: i64,
    pub terms: Vec<OntologyTermRecord>,
}

const ONTOLOGY_TERMS_DEFAULT_LIMIT: i64 = 100;
const ONTOLOGY_TERMS_MAX_LIMIT: i64 = 500;

pub async fn ontology_terms(
    State(state): State<AppState>,
    Query(params): Query<OntologyTermsQuery>,
) -> Result<Json<OntologyTermsPage>, ApiError> {
    if params.prefix.trim().is_empty() {
        return Err(ApiError::BadRequest("prefix is required".to_owned()));
    }
    let limit = params
        .limit
        .unwrap_or(ONTOLOGY_TERMS_DEFAULT_LIMIT)
        .clamp(1, ONTOLOGY_TERMS_MAX_LIMIT);
    let offset = params.offset.unwrap_or(0).max(0);
    let prefix_upper = params.prefix.to_ascii_uppercase();
    let (terms, total) = state
        .service
        .list_ontology_terms(&prefix_upper, limit, offset, params.q.as_deref())
        .await?;
    Ok(Json(OntologyTermsPage {
        prefix: prefix_upper,
        total,
        limit,
        offset,
        terms,
    }))
}

pub async fn ontology_descendants(
    State(state): State<AppState>,
    Query(params): Query<OntologyTermQuery>,
) -> Result<Json<Vec<OntologyDescendant>>, ApiError> {
    let canonical = canonical_term_iri(&params.iri);
    let rows = state.service.descendants(&canonical).await?;
    let out = rows
        .into_iter()
        .map(|(iri, depth)| OntologyDescendant { iri, depth })
        .collect();
    Ok(Json(out))
}

fn canonical_term_iri(input: &str) -> String {
    if input.contains("://") {
        return input.to_owned();
    }
    if let Some((prefix, suffix)) = input.split_once(':') {
        let p = prefix.to_ascii_uppercase();
        return format!("http://purl.obolibrary.org/obo/{p}_{suffix}");
    }
    input.to_owned()
}

fn resolve_ontology_defaults(
    prefix: &str,
    url: Option<String>,
    name: Option<String>,
) -> Result<(String, String), String> {
    let prefix_lower = prefix.to_ascii_lowercase();
    match prefix_lower.as_str() {
        "so" => Ok((
            url.unwrap_or_else(|| "http://purl.obolibrary.org/obo/so.obo".to_owned()),
            name.unwrap_or_else(|| "Sequence Ontology".to_owned()),
        )),
        "sbo" => Ok((
            url.unwrap_or_else(|| "http://purl.obolibrary.org/obo/sbo.obo".to_owned()),
            name.unwrap_or_else(|| "Systems Biology Ontology".to_owned()),
        )),
        _ => {
            let url = url.ok_or_else(|| {
                format!("`url` is required for ontology prefix {prefix} (no default known)")
            })?;
            let name = name.unwrap_or_else(|| prefix.to_owned());
            Ok((url, name))
        }
    }
}

#[derive(Deserialize)]
pub struct EnqueueJobBody {
    pub kind: String,
    pub payload: serde_json::Value,
    #[serde(default)]
    pub queue: Option<String>,
    #[serde(default)]
    pub priority: Option<i16>,
    #[serde(default)]
    pub max_attempts: Option<i32>,
    #[serde(default)]
    pub idempotency_key: Option<String>,
    #[serde(default)]
    pub correlation_id: Option<Uuid>,
}

#[derive(serde::Serialize)]
pub struct EnqueueJobResponse {
    pub job: SbolJob,
    pub deduplicated: bool,
}

pub async fn enqueue_job(
    State(state): State<AppState>,
    Json(body): Json<EnqueueJobBody>,
) -> Result<Json<EnqueueJobResponse>, ApiError> {
    let outcome = state
        .jobs
        .enqueue(NewJob {
            kind: body.kind,
            payload: body.payload,
            queue: body.queue,
            priority: body.priority,
            max_attempts: body.max_attempts,
            idempotency_key: body.idempotency_key,
            available_at: None,
            parent_job_id: None,
            correlation_id: body.correlation_id,
        })
        .await?;
    let (deduplicated, job) = match outcome {
        EnqueueOutcome::Inserted(j) => (false, j),
        EnqueueOutcome::AlreadyExists(j) => (true, j),
    };
    Ok(Json(EnqueueJobResponse { job, deduplicated }))
}

pub async fn get_job(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Json<SbolJob>, ApiError> {
    let job = state
        .jobs
        .get(JobId(id))
        .await?
        .ok_or_else(|| ApiError::NotFound(format!("job {id}")))?;
    Ok(Json(job))
}

#[derive(Deserialize)]
pub struct ListJobsParams {
    #[serde(default)]
    pub kind: Option<String>,
    #[serde(default)]
    pub status: Option<String>,
    #[serde(default)]
    pub queue: Option<String>,
    #[serde(default)]
    pub correlation_id: Option<Uuid>,
    #[serde(default = "default_jobs_list_limit")]
    pub limit: u32,
}

fn default_jobs_list_limit() -> u32 {
    100
}

pub async fn list_jobs(
    State(state): State<AppState>,
    Query(params): Query<ListJobsParams>,
) -> Result<Json<Vec<SbolJob>>, ApiError> {
    let status = match params.status.as_deref() {
        None => None,
        Some(s) => Some(parse_job_status(s)?),
    };
    let filter = ListJobsFilter {
        kind: params.kind,
        status,
        queue: params.queue,
        correlation_id: params.correlation_id,
        since: None,
        limit: params.limit,
    };
    Ok(Json(state.jobs.list(&filter).await?))
}

pub async fn cancel_job(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let cancelled = state.jobs.cancel(JobId(id)).await?;
    Ok(Json(json!({ "cancelled": cancelled })))
}

pub async fn list_job_attempts(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Json<Vec<JobAttempt>>, ApiError> {
    // 404 if the job itself doesn't exist; an empty attempts list for an
    // existing-but-never-dequeued job is a legitimate 200.
    let job = state.jobs.get(JobId(id)).await?;
    if job.is_none() {
        return Err(ApiError::NotFound(format!("job {id}")));
    }
    let attempts = state.jobs.list_attempts(JobId(id)).await?;
    Ok(Json(attempts))
}

#[derive(Deserialize)]
pub struct ListJobLogsQuery {
    #[serde(default)]
    pub after_id: Option<i64>,
    #[serde(default)]
    pub limit: Option<u32>,
}

pub async fn list_job_logs(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
    Query(query): Query<ListJobLogsQuery>,
) -> Result<Json<Vec<JobLogRecord>>, ApiError> {
    let job_id = JobId(id);
    state
        .jobs
        .get(job_id)
        .await?
        .ok_or_else(|| ApiError::NotFound(format!("job {id}")))?;
    let logs = state
        .jobs
        .list_logs(job_id, query.after_id, query.limit.unwrap_or(500))
        .await?;
    Ok(Json(logs))
}

fn parse_job_status(s: &str) -> Result<JobStatus, ApiError> {
    Ok(match s {
        "queued" => JobStatus::Queued,
        "running" => JobStatus::Running,
        "succeeded" => JobStatus::Succeeded,
        "failed" => JobStatus::Failed,
        "cancelled" => JobStatus::Cancelled,
        "dead" => JobStatus::Dead,
        other => return Err(ApiError::BadRequest(format!("unknown job status: {other}"))),
    })
}

// ── SynBioHub / Virtuoso compatibility: authenticated write endpoints ────────
//
// These mirror Virtuoso's write surface so SynBioHub can target sbol-db with no
// config change. They are mounted under an HTTP Basic auth layer (see
// `crate::auth`). The SPARQL update engine and the graph-store service are
// constructed per request from the shared service (both are cheap, pool-backed
// handles).

#[derive(Deserialize, Default)]
pub struct SparqlAuthParams {
    /// SynBioHub/Virtuoso pass the *update* string in `query=` (a Virtuoso
    /// convention), not the standard `update=`.
    pub query: Option<String>,
    pub update: Option<String>,
    #[serde(rename = "default-graph-uri")]
    pub default_graph_uri: Option<String>,
}

/// `GET|POST /sparql-auth` — execute a SPARQL 1.1 Update. The update text and
/// `default-graph-uri` arrive either as URL query parameters (synbiohub3 POSTs
/// them in the URL) or as a form body (classic SynBioHub); the form body wins
/// when both are present. A raw `application/sparql-update` body is also
/// accepted.
pub async fn sparql_auth(
    State(state): State<AppState>,
    Query(url_params): Query<SparqlAuthParams>,
    headers: HeaderMap,
    body: axum::body::Bytes,
) -> Result<impl IntoResponse, ApiError> {
    let mut update = url_params.query.or(url_params.update);
    let mut default_graph = url_params.default_graph_uri;

    let content_type = content_type_of(&headers);
    if content_type == "application/x-www-form-urlencoded" && !body.is_empty() {
        let form: SparqlAuthParams = serde_urlencoded::from_bytes(&body)
            .map_err(|e| ApiError::BadRequest(format!("invalid form body: {e}")))?;
        if let Some(q) = form.query.or(form.update) {
            update = Some(q);
        }
        if form.default_graph_uri.is_some() {
            default_graph = form.default_graph_uri;
        }
    } else if content_type == "application/sparql-update" && !body.is_empty() {
        let raw = std::str::from_utf8(&body)
            .map_err(|_| ApiError::BadRequest("update body is not UTF-8".to_owned()))?;
        update = Some(raw.to_owned());
    }

    let update = update
        .ok_or_else(|| ApiError::BadRequest("missing update (query= parameter)".to_owned()))?;

    let options = SparqlOptions::default();
    let outcome = state
        .sparql_update
        .execute(&update, default_graph.as_deref(), &options)
        .await?;
    if outcome.inserted != 0 || outcome.deleted != 0 {
        state
            .app
            .schedule_search_reconciliation(IndexMutationSource::SparqlUpdate)
            .await?;
    }

    // SynBioHub parses the update response as SPARQL-results-JSON and reads the
    // `callret-0` binding (Virtuoso's update-status convention). Its
    // `deleteStaggered` loop re-runs a DELETE until that status says "nothing to
    // do", so a no-op update must report exactly that to terminate the loop.
    let status = if outcome.inserted == 0 && outcome.deleted == 0 {
        "nothing to do".to_owned()
    } else {
        format!(
            "Modify <{}>, {} triples removed, {} triples added -- done",
            default_graph.as_deref().unwrap_or(""),
            outcome.deleted,
            outcome.inserted
        )
    };
    let results = json!({
        "head": { "link": [], "vars": ["callret-0"] },
        "results": {
            "distinct": false,
            "ordered": true,
            "bindings": [
                { "callret-0": { "type": "literal", "value": status } }
            ]
        }
    });
    Ok((
        [(CONTENT_TYPE, "application/sparql-results+json")],
        serde_json::to_string(&results).expect("serialize sparql-results json"),
    ))
}

#[derive(Deserialize, Default)]
pub struct GraphCrudParams {
    /// Virtuoso/SynBioHub spell the target graph `graph-uri`; the W3C Graph
    /// Store Protocol uses `graph`. Accept both.
    #[serde(rename = "graph-uri")]
    pub graph_uri: Option<String>,
    pub graph: Option<String>,
}

fn graph_crud_target(params: &GraphCrudParams) -> Result<String, ApiError> {
    params
        .graph_uri
        .clone()
        .or_else(|| params.graph.clone())
        .ok_or_else(|| ApiError::BadRequest("missing graph-uri parameter".to_owned()))
}

fn content_type_of(headers: &HeaderMap) -> String {
    headers
        .get(CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .map(|s| {
            s.split(';')
                .next()
                .unwrap_or("")
                .trim()
                .to_ascii_lowercase()
        })
        .unwrap_or_default()
}

/// Map an upload Content-Type to a parser format. `text/n3` and unknown types
/// fall back to Turtle, whose parser is a superset that also accepts
/// N-Triples/N3-basic — which is what SynBioHub's chunked uploads contain.
fn rdf_format_from_content_type(content_type: &str) -> SerializationFormat {
    match content_type {
        "application/rdf+xml" => SerializationFormat::RdfXml,
        "application/n-triples" => SerializationFormat::NTriples,
        "application/ld+json" => SerializationFormat::JsonLd,
        _ => SerializationFormat::Turtle,
    }
}

/// `POST /sparql-graph-crud-auth/?graph-uri=…` — merge (append) the posted RDF
/// into the named graph. SynBioHub splits large submissions into chunks POSTed
/// sequentially to the same graph, so this must accumulate, not replace.
pub async fn graph_store_post(
    State(state): State<AppState>,
    Query(params): Query<GraphCrudParams>,
    headers: HeaderMap,
    body: axum::body::Bytes,
) -> Result<impl IntoResponse, ApiError> {
    graph_store_write(state, params, headers, body, GraphWriteMode::Merge).await
}

/// `PUT /sparql-graph-crud-auth/?graph-uri=…` — replace the named graph's
/// entire contents with the posted RDF.
pub async fn graph_store_put(
    State(state): State<AppState>,
    Query(params): Query<GraphCrudParams>,
    headers: HeaderMap,
    body: axum::body::Bytes,
) -> Result<impl IntoResponse, ApiError> {
    graph_store_write(state, params, headers, body, GraphWriteMode::Replace).await
}

async fn graph_store_write(
    state: AppState,
    params: GraphCrudParams,
    headers: HeaderMap,
    body: axum::body::Bytes,
    mode: GraphWriteMode,
) -> Result<axum::response::Response, ApiError> {
    let graph = graph_crud_target(&params)?;
    let format = rdf_format_from_content_type(&content_type_of(&headers));
    let body = std::str::from_utf8(&body)
        .map_err(|_| ApiError::BadRequest("RDF body is not UTF-8".to_owned()))?;
    let inserted = state
        .service
        .graph_store_write(&graph, body, format, mode)
        .await?;
    if inserted != 0 {
        state
            .app
            .schedule_search_reconciliation(IndexMutationSource::GraphStore)
            .await?;
    }
    Ok((StatusCode::OK, Json(json!({ "inserted": inserted }))).into_response())
}

/// `DELETE /sparql-graph-crud-auth/?graph-uri=…` — drop the named graph.
pub async fn graph_store_delete(
    State(state): State<AppState>,
    Query(params): Query<GraphCrudParams>,
) -> Result<impl IntoResponse, ApiError> {
    let graph = graph_crud_target(&params)?;
    let deleted = state.service.graph_store_clear(&graph).await?;
    if deleted != 0 {
        state
            .app
            .schedule_search_reconciliation(IndexMutationSource::GraphStore)
            .await?;
    }
    Ok((StatusCode::OK, Json(json!({ "deleted": deleted }))))
}

/// `GET /sparql-graph-crud-auth/?graph-uri=…` — serialize a named graph as RDF.
/// Format from `?format=` or the `Accept` header; defaults to Turtle.
pub async fn graph_store_get(
    State(state): State<AppState>,
    Query(params): Query<GraphCrudParams>,
    headers: HeaderMap,
) -> Result<Response, ApiError> {
    let graph = graph_crud_target(&params)?;
    let format = resolve_format(None, &headers).unwrap_or(SerializationFormat::Turtle);
    let content_type = rdf_content_type(format);
    if matches!(
        format,
        SerializationFormat::Turtle | SerializationFormat::NTriples
    ) {
        // N-Triples is a valid Turtle subset. Serializing each bounded page as
        // N-Triples therefore gives both formats one syntactically valid,
        // incremental response without retaining the production public graph.
        let service = state.service.clone();
        let stream = stream::try_unfold(
            (service, graph, None::<String>, false),
            |(service, graph, cursor, done)| async move {
                if done {
                    return Ok(None);
                }
                let page = service
                    .graph_store_read_page(&graph, cursor.as_deref(), GRAPH_STREAM_PAGE)
                    .await
                    .map_err(|error| io::Error::other(error.to_string()))?;
                if page.items.is_empty() {
                    return Ok(None);
                }
                let body = triples_to_rdf(&page.items, SerializationFormat::NTriples)
                    .map_err(|error| io::Error::other(error.to_string()))?;
                let done = page.next_cursor.is_none();
                Ok::<_, io::Error>(Some((
                    Bytes::from(body),
                    (service, graph, page.next_cursor, done),
                )))
            },
        );
        let mut response = Response::new(Body::from_stream(stream));
        response.headers_mut().insert(
            CONTENT_TYPE,
            content_type.parse().expect("static RDF content type"),
        );
        return Ok(response);
    }

    // RDF/XML and JSON-LD serializers need a whole document. Keep that path
    // complete for ordinary graphs, but fail explicitly rather than silently
    // truncating a production-scale graph. Turtle/N-Triples is the unbounded
    // export surface.
    let mut triples = Vec::new();
    let mut cursor = None;
    loop {
        let page = state
            .service
            .graph_store_read_page(&graph, cursor.as_deref(), GRAPH_STREAM_PAGE)
            .await?;
        if triples.len().saturating_add(page.items.len()) > BUFFERED_GRAPH_LIMIT {
            return Err(ApiError::BadRequest(
                "this graph is too large for a buffered RDF/XML or JSON-LD response; request Turtle or N-Triples for a complete streaming export"
                    .to_owned(),
            ));
        }
        triples.extend(page.items);
        cursor = page.next_cursor;
        if cursor.is_none() {
            break;
        }
    }
    let body = triples_to_rdf(&triples, format)?;
    Ok(([(CONTENT_TYPE, content_type)], body).into_response())
}

fn rdf_content_type(format: SerializationFormat) -> &'static str {
    match format {
        SerializationFormat::Turtle => "text/turtle",
        SerializationFormat::NTriples => "application/n-triples",
        SerializationFormat::JsonLd => "application/ld+json",
        SerializationFormat::RdfXml => "application/rdf+xml",
        _ => "text/turtle",
    }
}