ARCADIA-Studio/crates/oa-server/src/routes.rs

use crate::{
    AppState, blocking,
    error::{ApiError, ApiResult, IssueDto, issues, source_issue},
    store::{
        MODEL_LIMITS, PreparedProject, SOURCE_LIMITS, StoredProject, check_model_size,
        check_source_size, validate_model,
    },
};
use axum::{
    Json, Router,
    body::Bytes,
    extract::{
        Path, Query, State,
        rejection::{BytesRejection, PathRejection, QueryRejection},
    },
    http::{HeaderMap, StatusCode, header},
    response::{IntoResponse, Response},
    routing::{get, post},
};
use oa_core::source_project::{self, SourceBundle};
use oa_core::{Command, Id, Kind, Layer, Project, RelationKind};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::{Value, json};
use std::collections::{BTreeMap, BTreeSet, VecDeque};

type ProjectPath = Result<Path<Id>, PathRejection>;
type Body = Result<Bytes, BytesRejection>;

pub(crate) fn router() -> Router<AppState> {
    Router::new()
        .route("/health", get(health))
        .route("/health/live", get(health))
        .route("/health/ready", get(health))
        .route("/openapi.json", get(openapi))
        .route("/api/v1/openapi.json", get(openapi))
        .route("/api/v1/projects", get(list_projects).post(create_project))
        .route("/api/v1/projects/import", post(import_project))
        .route(
            "/api/v1/projects/import-source-project",
            post(import_source_project),
        )
        .route(
            "/api/v1/projects/{id}",
            get(get_project)
                .patch(rename_project)
                .delete(delete_project),
        )
        .route("/api/v1/projects/{id}/model", get(get_model).put(put_model))
        .route(
            "/api/v1/projects/{id}/source",
            get(get_source).put(put_source),
        )
        .route(
            "/api/v1/projects/{id}/source-project",
            get(get_source_project).put(put_source_project),
        )
        .route("/api/v1/projects/{id}/elements", get(list_elements))
        .route("/api/v1/projects/{id}/relations", get(list_relations))
        .route("/api/v1/projects/{id}/artifact-links", get(list_artifacts))
        .route("/api/v1/projects/{id}/traces", get(list_traces))
        .route("/api/v1/projects/{id}/query", post(neighborhood))
        .route(
            "/api/v1/projects/{id}/validation",
            get(get_validation).post(validate_candidate),
        )
        .route("/api/v1/projects/{id}/transactions", post(transactions))
        .route(
            "/api/v1/projects/{id}/transitions/preview",
            post(preview_transition),
        )
        .route(
            "/api/v1/projects/{id}/transitions/apply",
            post(commit_transition),
        )
        .fallback(|| async {
            ApiError::new(
                StatusCode::NOT_FOUND,
                "route_not_found",
                "API route does not exist.",
            )
        })
        .method_not_allowed_fallback(|| async {
            ApiError::new(
                StatusCode::METHOD_NOT_ALLOWED,
                "method_not_allowed",
                "HTTP method is not supported by this route.",
            )
        })
}

async fn health() -> Json<Value> {
    Json(json!({"status":"ok"}))
}

async fn openapi() -> Response {
    (
        [(header::CONTENT_TYPE, "application/json")],
        include_str!("../../../docs/openapi.json"),
    )
        .into_response()
}

fn path(value: ProjectPath) -> ApiResult<Id> {
    value
        .map(|Path(id)| id)
        .map_err(|error| ApiError::bad_request(error.body_text()))
}

fn query<T>(value: Result<Query<T>, QueryRejection>) -> ApiResult<T> {
    value
        .map(|Query(value)| value)
        .map_err(|error| ApiError::bad_request(error.body_text()))
}

fn bytes(value: Body) -> ApiResult<Bytes> {
    value.map_err(|error| ApiError::new(error.status(), "invalid_body", error.body_text()))
}

fn json_body<T: DeserializeOwned>(body: &[u8]) -> ApiResult<T> {
    serde_json::from_slice(body)
        .map_err(|error| ApiError::bad_request(format!("Invalid JSON: {error}")))
}

fn media(headers: &HeaderMap, expected: &str) -> ApiResult<()> {
    let actual = headers
        .get(header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        .map(str::trim);
    if actual != Some(expected) {
        return Err(ApiError::new(
            StatusCode::UNSUPPORTED_MEDIA_TYPE,
            "unsupported_media_type",
            format!("Content-Type must be {expected}."),
        ));
    }
    Ok(())
}

fn expected(headers: &HeaderMap) -> ApiResult<String> {
    let mut values = headers.get_all(header::IF_MATCH).iter();
    let value = values
        .next()
        .ok_or_else(|| {
            ApiError::new(
                StatusCode::PRECONDITION_REQUIRED,
                "precondition_required",
                "Supply the project's ETag in If-Match.",
            )
        })?
        .to_str()
        .map_err(|_| ApiError::bad_request("Invalid If-Match header."))?;
    if values.next().is_some()
        || value.len() < 2
        || !value.starts_with('"')
        || !value.ends_with('"')
        || value[1..value.len() - 1].contains(['"', ','])
    {
        return Err(ApiError::bad_request(
            "If-Match must contain exactly one strong, quoted ETag; wildcard/list/weak tags are unsupported.",
        ));
    }
    Ok(value.to_owned())
}

fn response(stored: &StoredProject, status: StatusCode, value: impl Serialize) -> Response {
    let mut response = (status, Json(value)).into_response();
    response
        .headers_mut()
        .insert(header::ETAG, stored.etag().parse().expect("ETag is ASCII"));
    response
}

fn not_modified(stored: &StoredProject, headers: &HeaderMap) -> Option<Response> {
    let etag = stored.etag();
    let matched = headers
        .get_all(header::IF_NONE_MATCH)
        .iter()
        .filter_map(|value| value.to_str().ok())
        .flat_map(|value| value.split(','))
        .map(str::trim)
        .any(|value| {
            value == "*" || value == etag || value.strip_prefix("W/") == Some(etag.as_str())
        });
    matched.then(|| {
        let mut response = StatusCode::NOT_MODIFIED.into_response();
        response
            .headers_mut()
            .insert(header::ETAG, etag.parse().expect("ETag is ASCII"));
        response
    })
}

fn summary(stored: &StoredProject) -> Value {
    json!({"resource_id":stored.resource_id,"revision":stored.revision,"model_id":stored.model.id,
        "schema_version":stored.model.schema_version,"name":stored.model.name,
        "elements":stored.model.elements.len(),"relations":stored.model.relations.len(),"migration":stored.migration})
}

fn parse_source(body: &[u8]) -> ApiResult<oa_core::persistence::DecodeResult> {
    let source =
        std::str::from_utf8(body).map_err(|_| ApiError::bad_request("Source must be UTF-8."))?;
    oa_core::language::parse_with_report_limited(source, MODEL_LIMITS).map_err(|diagnostics| {
        let code = if diagnostics.iter().any(|d| d.code == "model_limit") {
            "model_limit"
        } else {
            "invalid_source"
        };
        let mut error = ApiError::new(
            StatusCode::UNPROCESSABLE_ENTITY,
            code,
            "Arcadia source could not be applied.",
        );
        error.issues = diagnostics
            .into_iter()
            .map(|issue| IssueDto {
                severity: "error",
                code: issue.code,
                message: issue.message,
                field: issue.field,
                object_id: issue.object_id,
                line: Some(issue.line),
                column: Some(issue.column),
                path: None,
                span: Some(issue.span.into()),
                related: vec![],
            })
            .collect();
        error
    })
}

fn import_json(body: &[u8]) -> ApiResult<oa_core::persistence::DecodeResult> {
    // Format is chosen by HTTP. The shared decoder owns supported older-schema migration.
    let _: Value = json_body(body)?;
    oa_core::persistence::decode_with_report_limited(body, MODEL_LIMITS).map_err(
        |error| match error {
            oa_core::persistence::FileError::Limit(detail) => {
                ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "model_limit", detail)
            }
            oa_core::persistence::FileError::Model { message, issues } => {
                let mut error =
                    ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "invalid_model", message);
                error.issues = issues
                    .into_iter()
                    .map(|issue| IssueDto {
                        severity: "error",
                        code: issue.code,
                        message: issue.message,
                        field: issue.field,
                        object_id: issue.element,
                        line: None,
                        column: None,
                        path: None,
                        span: None,
                        related: vec![],
                    })
                    .collect();
                error
            }
            oa_core::persistence::FileError::Json(error) => {
                ApiError::bad_request(format!("Invalid model JSON: {error}"))
            }
            error => ApiError::new(
                StatusCode::UNPROCESSABLE_ENTITY,
                "invalid_model",
                error.to_string(),
            ),
        },
    )
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct NameRequest {
    name: String,
}

async fn create_project(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let request: NameRequest = json_body(&body)?;
        let mut model = Project::empty();
        model.name = request.name;
        created(store.create(model)?)
    })
    .await
}

fn created(stored: StoredProject) -> ApiResult<Response> {
    let mut result = response(&stored, StatusCode::CREATED, summary(&stored));
    result.headers_mut().insert(
        header::LOCATION,
        format!("/api/v1/projects/{}", stored.resource_id)
            .parse()
            .expect("UUID URL is ASCII"),
    );
    Ok(result)
}

async fn import_project(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let is_json = media(&headers, "application/json").is_ok();
    if !is_json {
        media(&headers, "text/plain")?;
    }
    let body = bytes(body)?;
    blocking(state, move |store| {
        let model = if is_json {
            import_json(&body)?
        } else {
            parse_source(&body)?
        };
        created(store.create_decoded(model)?)
    })
    .await
}

async fn import_source_project(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let bundle: SourceBundle = json_body(&body)?;
        created(store.create_source_project(bundle)?)
    })
    .await
}

#[derive(Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct Page {
    limit: Option<usize>,
    after: Option<Id>,
    q: Option<String>,
}

impl Page {
    fn limit(&self) -> ApiResult<usize> {
        let limit = self.limit.unwrap_or(100);
        if !(1..=1000).contains(&limit) {
            return Err(ApiError::bad_request("limit must be between 1 and 1000."));
        }
        Ok(limit)
    }
}

async fn list_projects(
    State(state): State<AppState>,
    pagination: Result<Query<Page>, QueryRejection>,
) -> ApiResult<Response> {
    let page = query(pagination)?;
    let limit = page.limit()?;
    blocking(state, move |store| {
        let needle = page.q.as_deref().unwrap_or("").to_lowercase();
        let catalog = store.catalog()?;
        let mut matches = catalog.iter().filter(|stored| {
            page.after.is_none_or(|id| stored.resource_id > id)
                && stored.name.to_lowercase().contains(&needle)
        });
        let items: Vec<_> = matches.by_ref().take(limit).collect();
        let next = if matches.next().is_some() {
            items.last().map(|stored| stored.resource_id)
        } else {
            None
        };
        Ok(Json(json!({"items":items,"next_after":next})).into_response())
    })
    .await
}

async fn get_project(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
) -> ApiResult<Response> {
    let id = path(id)?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        Ok(not_modified(stored, &headers)
            .unwrap_or_else(|| response(stored, StatusCode::OK, stored)))
    })
    .await
}

async fn rename_project(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let id = path(id)?;
    let expected = expected(&headers)?;
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let request: NameRequest = json_body(&body)?;
        let snapshot = store.checked(id, &expected)?;
        let current = &snapshot;
        let prepared = oa_core::prepare_command(
            &current.model,
            &Command::EditProject {
                id: current.model.id,
                alias: current.model.alias.clone(),
                name: request.name,
                description: current.model.description.clone(),
            },
        )
        .map_err(|error| {
            ApiError::new(
                StatusCode::UNPROCESSABLE_ENTITY,
                "command_rejected",
                error.to_string(),
            )
        })?;
        let prepared = PreparedProject::from_command(current, prepared)?;
        let (stored, changed) = store.replace(id, &expected, prepared)?;
        Ok(response(
            &stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"changed":changed}),
        ))
    })
    .await
}

async fn delete_project(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
) -> ApiResult<Response> {
    let id = path(id)?;
    let expected = expected(&headers)?;
    blocking(state, move |store| {
        store.delete(id, &expected)?;
        Ok(StatusCode::NO_CONTENT.into_response())
    })
    .await
}

async fn get_model(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
) -> ApiResult<Response> {
    let id = path(id)?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        Ok(not_modified(stored, &headers)
            .unwrap_or_else(|| response(stored, StatusCode::OK, &stored.model)))
    })
    .await
}

async fn get_source(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
) -> ApiResult<Response> {
    let id = path(id)?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        if let Some(result) = not_modified(stored, &headers) {
            return Ok(result);
        }
        let mut result = (
            [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
            oa_core::language::format(&stored.model),
        )
            .into_response();
        result
            .headers_mut()
            .insert(header::ETAG, stored.etag().parse().expect("ETag is ASCII"));
        Ok(result)
    })
    .await
}

async fn get_source_project(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
) -> ApiResult<Response> {
    let id = path(id)?;
    blocking(state, move |store| {
        let stored = store.get(id)?;
        Ok(not_modified(&stored, &headers)
            .unwrap_or_else(|| response(&stored, StatusCode::OK, &stored.source_project)))
    })
    .await
}

async fn put_source_project(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let id = path(id)?;
    let expected = expected(&headers)?;
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        store.checked(id, &expected)?;
        let bundle: SourceBundle = json_body(&body)?;
        let prepared = PreparedProject::from_bundle(bundle)?;
        let diagnostics = prepared.source_issues();
        let (stored, changed) = store.replace(id, &expected, prepared)?;
        Ok(response(
            &stored,
            StatusCode::OK,
            json!({
                "revision": stored.revision, "changed": changed, "issues": diagnostics,
                "source_project": stored.source_project, "migration": stored.migration
            }),
        ))
    })
    .await
}

async fn put_model(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    replace_model(state, id, headers, body, true).await
}

async fn put_source(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    replace_model(state, id, headers, body, false).await
}

async fn replace_model(
    state: AppState,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
    is_json: bool,
) -> ApiResult<Response> {
    let id = path(id)?;
    let expected = expected(&headers)?;
    media(
        &headers,
        if is_json {
            "application/json"
        } else {
            "text/plain"
        },
    )?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        store.checked(id, &expected)?;
        let model = if is_json {
            import_json(&body)?
        } else {
            parse_source(&body)?
        };
        let (stored, changed) = store.replace_decoded(id, &expected, model)?;
        Ok(response(
            &stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"changed":changed,"issues":issues(&stored.model),"migration":stored.migration}),
        ))
    })
    .await
}

#[derive(Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct ElementQuery {
    limit: Option<usize>,
    after: Option<Id>,
    revision: Option<u64>,
    kind: Option<Kind>,
    layer: Option<Layer>,
    alias: Option<String>,
    q: Option<String>,
}

#[derive(Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct RelationQuery {
    limit: Option<usize>,
    after: Option<Id>,
    revision: Option<u64>,
    kind: Option<RelationKind>,
    source: Option<Id>,
    target: Option<Id>,
    incident_to: Option<Id>,
}

fn check_read_revision(stored: &StoredProject, revision: Option<u64>) -> ApiResult<()> {
    if revision.is_some_and(|revision| revision != stored.revision) {
        let mut error = ApiError::new(
            StatusCode::CONFLICT,
            "snapshot_changed",
            "Only the current revision is retained. Restart pagination from the current snapshot.",
        );
        error.current_revision = Some(stored.revision);
        return Err(error);
    }
    Ok(())
}

async fn list_elements(
    State(state): State<AppState>,
    id: ProjectPath,
    parameters: Result<Query<ElementQuery>, QueryRejection>,
) -> ApiResult<Response> {
    let id = path(id)?;
    let query = query(parameters)?;
    let limit = Page {
        limit: query.limit,
        ..Page::default()
    }
    .limit()?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        check_read_revision(stored, query.revision)?;
        let needle = query.q.as_deref().unwrap_or("").to_lowercase();
        let mut items = Vec::new();
        let mut next = None;
        for element in stored.model.elements.values() {
            if query.after.is_some_and(|id| element.id <= id)
                || query.kind.is_some_and(|kind| element.kind() != kind)
                || !element.name.to_lowercase().contains(&needle)
                || query
                    .alias
                    .as_ref()
                    .is_some_and(|alias| element.alias != *alias)
                || query.layer.is_some_and(|layer| element.layer() != layer)
            {
                continue;
            }
            let value = serde_json::to_value(element)
                .map_err(|_| ApiError::internal("Element serialization failed."))?;
            if items.len() == limit {
                next = items
                    .last()
                    .and_then(|item: &Value| item.get("id"))
                    .cloned();
                break;
            }
            items.push(value);
        }
        Ok(response(
            stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"items":items,"next_after":next}),
        ))
    })
    .await
}

async fn list_relations(
    State(state): State<AppState>,
    id: ProjectPath,
    parameters: Result<Query<RelationQuery>, QueryRejection>,
) -> ApiResult<Response> {
    let id = path(id)?;
    let query = query(parameters)?;
    let limit = Page {
        limit: query.limit,
        ..Page::default()
    }
    .limit()?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        check_read_revision(stored, query.revision)?;
        let mut matches = stored.model.relations.values().filter(|relation| {
            query.after.is_none_or(|id| relation.id > id)
                && query.kind.is_none_or(|kind| relation.kind() == kind)
                && query.source.is_none_or(|id| relation.source() == id)
                && query.target.is_none_or(|id| relation.target() == id)
                && query
                    .incident_to
                    .is_none_or(|id| relation.source() == id || relation.target() == id)
        });
        let items: Vec<_> = matches.by_ref().take(limit).collect();
        let next = if matches.next().is_some() {
            items.last().map(|relation| relation.id)
        } else {
            None
        };
        Ok(response(
            stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"items":items,"next_after":next}),
        ))
    })
    .await
}

#[derive(Default, Deserialize)]
#[serde(rename_all = "snake_case")]
enum Direction {
    Incoming,
    Outgoing,
    #[default]
    Both,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Neighborhood {
    roots: Vec<Id>,
    #[serde(default)]
    direction: Direction,
    #[serde(default)]
    relation_kinds: Vec<RelationKind>,
    #[serde(default = "default_depth")]
    max_depth: usize,
    #[serde(default = "default_objects")]
    max_objects: usize,
    revision: Option<u64>,
}
fn default_depth() -> usize {
    1
}
fn default_objects() -> usize {
    200
}

async fn neighborhood(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let id = path(id)?;
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let query: Neighborhood = json_body(&body)?;
        if query.roots.is_empty() || query.roots.len() > 100 || query.max_depth > 4 || !(1..=1000).contains(&query.max_objects) || query.roots.len() > query.max_objects {
            return Err(ApiError::bad_request("Use 1..100 roots, max_depth 0..4, and max_objects 1..1000 (at least the root count)."));
        }
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        check_read_revision(stored, query.revision)?;
        if query.roots.iter().any(|id| !stored.model.elements.contains_key(id) && !stored.model.relations.contains_key(id)) {
            return Err(ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "missing_root", "Every root must name an existing element or relation."));
        }
        let mut objects = BTreeSet::new();
        let mut traversed: BTreeSet<_> = query.roots.into_iter().collect();
        let mut queue: VecDeque<_> = traversed.iter().map(|id| (*id, 0)).collect();
        for root in &traversed { endpoint_closure(&stored.model, *root, &mut objects); }
        if objects.len() > query.max_objects {
            return Err(ApiError::bad_request("Root relation endpoint closure exceeds max_objects."));
        }
        let mut adjacency: BTreeMap<Id, Vec<(Id, Id)>> = BTreeMap::new();
        for relation in stored.model.relations.values() {
            if !query.relation_kinds.is_empty() && !query.relation_kinds.contains(&relation.kind()) { continue; }
            if matches!(query.direction, Direction::Outgoing | Direction::Both) {
                adjacency.entry(relation.source()).or_default().push((relation.id, relation.target()));
            }
            if matches!(query.direction, Direction::Incoming | Direction::Both) {
                adjacency.entry(relation.target()).or_default().push((relation.id, relation.source()));
            }
        }
        let mut truncated = false;
        while let Some((current, depth)) = queue.pop_front() {
            if depth == query.max_depth { continue; }
            for &(relation, neighbor) in adjacency.get(&current).into_iter().flatten() {
                let mut additions = BTreeSet::new();
                endpoint_closure(&stored.model, relation, &mut additions);
                let cost = additions.difference(&objects).count();
                if objects.len() + cost > query.max_objects { truncated = true; continue; }
                objects.extend(additions);
                if traversed.insert(neighbor) { queue.push_back((neighbor, depth + 1)); }
            }
        }
        Ok(response(stored, StatusCode::OK, json!({"revision":stored.revision,
            "elements":objects.iter().filter_map(|id| stored.model.elements.get(id)).collect::<Vec<_>>(),
            "relations":objects.iter().filter_map(|id| stored.model.relations.get(id)).collect::<Vec<_>>(),"truncated":truncated})))
    }).await
}

fn endpoint_closure(model: &Project, root: Id, objects: &mut BTreeSet<Id>) {
    let mut pending = vec![root];
    while let Some(id) = pending.pop() {
        if !objects.insert(id) {
            continue;
        }
        if let Some(relation) = model.relations.get(&id) {
            pending.extend([relation.source(), relation.target()]);
        }
    }
}

#[derive(Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct LinkQuery {
    limit: Option<usize>,
    after: Option<Id>,
    revision: Option<u64>,
    object: Option<Id>,
}

async fn list_artifacts(
    State(state): State<AppState>,
    id: ProjectPath,
    parameters: Result<Query<LinkQuery>, QueryRejection>,
) -> ApiResult<Response> {
    list_links(state, id, parameters, true).await
}

async fn list_traces(
    State(state): State<AppState>,
    id: ProjectPath,
    parameters: Result<Query<LinkQuery>, QueryRejection>,
) -> ApiResult<Response> {
    list_links(state, id, parameters, false).await
}

async fn list_links(
    state: AppState,
    id: ProjectPath,
    parameters: Result<Query<LinkQuery>, QueryRejection>,
    artifacts: bool,
) -> ApiResult<Response> {
    let id = path(id)?;
    let query = query(parameters)?;
    let limit = Page {
        limit: query.limit,
        ..Page::default()
    }
    .limit()?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        check_read_revision(stored, query.revision)?;
        let mut records = Vec::new();
        if artifacts {
            for link in stored
                .model
                .artifact_links
                .values()
                .filter(|link| {
                    query.after.is_none_or(|id| link.id > id)
                        && query.object.is_none_or(|id| link.owner == id)
                })
                .take(limit + 1)
            {
                records.push(
                    serde_json::to_value(link)
                        .map_err(|_| ApiError::internal("Artifact serialization failed."))?,
                );
            }
        } else {
            for link in stored
                .model
                .traces
                .values()
                .filter(|link| {
                    query.after.is_none_or(|id| link.id > id)
                        && query.object.is_none_or(|id| link.oa == id || link.sa == id)
                })
                .take(limit + 1)
            {
                records.push(
                    serde_json::to_value(link)
                        .map_err(|_| ApiError::internal("Trace serialization failed."))?,
                );
            }
        }
        let next = if records.len() > limit {
            records.truncate(limit);
            records.last().and_then(|record| record.get("id")).cloned()
        } else {
            None
        };
        Ok(response(
            stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"items":records,"next_after":next}),
        ))
    })
    .await
}

async fn get_validation(State(state): State<AppState>, id: ProjectPath) -> ApiResult<Response> {
    let id = path(id)?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        Ok(response(
            stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"valid":true,"issues":issues(&stored.model)}),
        ))
    })
    .await
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ValidationRequest {
    format: String,
    // Keep RawValue directly in the struct: internally tagged enums buffer
    // their content and cannot preserve raw model JSON for duplicate-key checks.
    model: Option<Box<serde_json::value::RawValue>>,
    source: Option<String>,
    source_project: Option<SourceBundle>,
}

async fn validate_candidate(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let id = path(id)?;
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let snapshot = store.get(id)?;
        let stored = &snapshot;
        let request: ValidationRequest = json_body(&body)?;
        let parsed = match (request.format.as_str(), request.model, request.source, request.source_project) {
            ("json", Some(model), None, None) => import_json(model.get().as_bytes()),
            ("arcadia", None, Some(source), None) => parse_source(source.as_bytes()),
            ("source-project", None, None, Some(source_project)) => {
                let result = check_source_size(&source_project).and_then(|()| {
                    // Advisory locations refer to submitted bytes, before Apply
                    // would insert missing persistent identities.
                    let diagnostics: Vec<_> = source_project::lint_bundle(&source_project, SOURCE_LIMITS).into_iter().map(source_issue).collect();
                    let migration = if !diagnostics.iter().any(|issue| issue.severity == "error") {
                        PreparedProject::from_bundle(source_project)?.migration().cloned()
                    } else {
                        None
                    };
                    Ok((diagnostics, migration))
                });
                let (diagnostics, migration) = result.unwrap_or_else(|error| (candidate_issues(error), None));
                let valid = !diagnostics.iter().any(|issue| issue.severity == "error");
                return Ok(response(stored, StatusCode::OK, json!({"revision":stored.revision,"valid":valid,"issues":diagnostics,"migration":migration})));
            }
            _ => return Err(ApiError::bad_request("Validation requires one matching format/payload: json with model, arcadia with source, or source-project with source_project.")),
        };
        let (valid, issues, migration) = match parsed.and_then(|decoded| {
            validate_model(&decoded.project)?;
            Ok(decoded)
        }) {
            Ok(decoded) => (true, issues(&decoded.project), decoded.migration),
            Err(error) => (false, candidate_issues(error), None),
        };
        Ok(response(
            stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"valid":valid,"issues":issues,"migration":migration}),
        ))
    })
    .await
}

fn candidate_issues(error: ApiError) -> Vec<IssueDto> {
    if !error.issues.is_empty() {
        return error.issues;
    }
    vec![IssueDto {
        severity: "error",
        code: error.code.into(),
        message: error.detail,
        field: None,
        object_id: None,
        line: None,
        column: None,
        path: None,
        span: None,
        related: vec![],
    }]
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TransactionRequest {
    commands: Vec<Command>,
    #[serde(default)]
    dry_run: bool,
}

async fn transactions(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let id = path(id)?;
    let expected = expected(&headers)?;
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let request: TransactionRequest = json_body(&body)?;
        check_commands(&request.commands)?;
        let snapshot = store.checked(id, &expected)?;
        let current = &snapshot;
        let candidate = oa_core::prepare_command(&current.model, &Command::Batch(request.commands))
            .map_err(|error| ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "command_rejected", error.to_string()))?;
        let prepared = PreparedProject::from_command(current, candidate)?;
        if request.dry_run {
            return Ok(response(current, StatusCode::OK, json!({"revision":current.revision,"dry_run":true,"changed":prepared.changed(current),"issues":prepared.source_issues(),"model":prepared.model(),"source_project":prepared.bundle()})));
        }
        let (stored, changed) = store.replace(id, &expected, prepared)?;
        Ok(response(&stored, StatusCode::OK, json!({"revision":stored.revision,"dry_run":false,"changed":changed,"issues":issues(&stored.model)})))
    }).await
}

fn check_commands(commands: &[Command]) -> ApiResult<()> {
    let mut pending: Vec<_> = commands.iter().collect();
    let mut count = 0;
    while let Some(command) = pending.pop() {
        count += 1;
        if count > 200 {
            return Err(ApiError::bad_request(
                "A transaction may contain at most 200 commands, including nested batches.",
            ));
        }
        if let Command::Batch(batch) = command {
            pending.extend(batch);
        }
        if let Command::Replace(model) = command {
            check_model_size(model)?;
        }
    }
    if count == 0 {
        return Err(ApiError::bad_request(
            "A transaction requires at least one command.",
        ));
    }
    Ok(())
}

async fn preview_transition(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let id = path(id)?;
    let expected = expected(&headers)?;
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let request: oa_core::transitions::TransitionRequest = json_body(&body)?;
        if request.sources.len() > 100 || request.mappings.len() > 200 || request.resolutions.len() > 400 {
            return Err(ApiError::bad_request("Transition requests allow at most 100 sources, 200 mappings and 400 field resolutions."));
        }
        let snapshot = store.checked(id, &expected)?;
        let current = &snapshot;
        let plan = oa_core::transitions::plan_transition(&current.model, &request)
            .map_err(|error| ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "transition_rejected", error.to_string()))?;
        if !plan.commands.is_empty() { check_commands(&plan.commands)?; }
        if plan.conflicts.is_empty() {
            let candidate = oa_core::transitions::prepare_transition(&current.model, &plan)
                .map_err(|error| ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "transition_rejected", error.to_string()))?;
            PreparedProject::from_command(current, candidate)?;
        }
        Ok(response(current, StatusCode::OK, json!({"base_revision":current.revision,"plan":plan})))
    }).await
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TransitionApplyRequest {
    base_revision: u64,
    plan: oa_core::transitions::TransitionPlan,
}

async fn commit_transition(
    State(state): State<AppState>,
    id: ProjectPath,
    headers: HeaderMap,
    body: Body,
) -> ApiResult<Response> {
    let id = path(id)?;
    let expected = expected(&headers)?;
    media(&headers, "application/json")?;
    let body = bytes(body)?;
    blocking(state, move |store| {
        let request: TransitionApplyRequest = json_body(&body)?;
        let snapshot = store.checked(id, &expected)?;
        let current = &snapshot;
        if current.revision != request.base_revision {
            let mut error = ApiError::new(
                StatusCode::PRECONDITION_FAILED,
                "plan_revision_conflict",
                "Transition preview revision changed. Create a new preview.",
            );
            error.current_revision = Some(current.revision);
            return Err(error);
        }
        if !request.plan.commands.is_empty() {
            check_commands(&request.plan.commands)?;
        }
        let candidate = oa_core::transitions::prepare_transition(&current.model, &request.plan)
            .map_err(|error| {
                ApiError::new(
                    StatusCode::UNPROCESSABLE_ENTITY,
                    "transition_rejected",
                    error.to_string(),
                )
            })?;
        let prepared = PreparedProject::from_command(current, candidate)?;
        let (stored, changed) = store.replace(id, &expected, prepared)?;
        Ok(response(
            &stored,
            StatusCode::OK,
            json!({"revision":stored.revision,"changed":changed,"issues":issues(&stored.model)}),
        ))
    })
    .await
}