//! Embedding dataset management routes.

use std::{collections::HashSet, sync::Arc};

use axum::{
    Extension, Json, Router,
    extract::{Path, Query},
    http::StatusCode,
    middleware::from_fn,
    response::{IntoResponse, Response},
    routing::{MethodRouter, get, post},
};
use philharmonic_policy::{
    CorpusItem, EmbeddingDataset, EmbeddingDatasetStatus, SourceItem, TenantEndpointConfig, atom,
    decode_corpus, decode_source_items, encode_source_items,
};
use philharmonic_store::{
    ContentStore, ContentStoreExt, EntityRefValue, EntityStore, EntityStoreExt, IdentityStore,
    RevisionInput, RevisionRow, StoreExt,
};
use philharmonic_types::{
    CanonicalJson, ContentHash, ContentValue, Entity, EntityId, JsonValue, ScalarValue, Sha256,
    UnixMillis, Uuid,
};
use serde::{Deserialize, Serialize};
use serde_json::Map as JsonMap;

use crate::{
    ApiError, AuthContext, EmbedDatasetCaps, EmbedDatasetDispatcher, PaginatedResponse,
    PaginationParams, RequestContext, RequestScope, RequiredPermission,
    middleware::authz::authorize,
    pagination::{CursorKey, PaginationError, decode_cursor, encode_cursor, page_size},
    store::{ApiStore, ApiStoreHandle},
};

/// Maximum number of source items accepted per dataset.
pub(crate) const MAX_ITEMS_PER_DATASET: usize = 10_000;
/// Maximum UTF-8 byte length of one source item text.
pub(crate) const MAX_TEXT_BYTES: usize = 65_536;
/// Maximum JSON-encoded byte length of one source item payload.
pub(crate) const MAX_PAYLOAD_JSON_BYTES: usize = 65_536;
/// Maximum encoded source-items blob size in bytes.
pub(crate) const MAX_SOURCE_ITEMS_BLOB_BYTES: usize = 256 * 1024 * 1024;
/// Maximum number of corpus items accepted from a mechanics embed-job
/// response. Bounds parse-time allocation against an unbounded mechanics
/// or upstream-provider response.
pub(crate) const MAX_CORPUS_ITEMS: usize = 10_000;
/// Maximum embedding-vector dimension accepted from a mechanics embed-job
/// response. Bounds per-vector allocation. Default sized for known
/// production embedding models with headroom: bge-m3 = 1024, OpenAI
/// `text-embedding-3-large` = 3072.
pub(crate) const MAX_CORPUS_VECTOR_DIMENSION: usize = 4_096;
/// Maximum encoded corpus blob size in bytes (per design 16). Bounds the
/// post-encode storage write.
pub(crate) const MAX_CORPUS_BLOB_BYTES: usize = 1024 * 1024 * 1024;
/// Maximum mechanics-worker HTTP response body size in bytes for embed
/// jobs. Bounds transient JSON-in-memory pressure during parse — under
/// the corpus item-count + vector-dimension caps the worst-case JSON is
/// roughly 1 GiB (item count × vector dim × ~17 bytes/f32-as-JSON +
/// payload), but JSON encoding overhead and structural slack pushes the
/// realistic upper bound to ~2 GiB. Default sized at 2 GiB so this is
/// not the bottleneck for legitimate workloads while still bounding
/// adversarial responses.
pub(crate) const MAX_MECHANICS_RESPONSE_BYTES: usize = 2 * 1024 * 1024 * 1024;

/// Shared embedding-dataset route state.
#[derive(Clone)]
pub(crate) struct EmbedDatasetState {
    store: ApiStoreHandle,
    caps: EmbedDatasetCaps,
    dispatcher: Arc<dyn EmbedDatasetDispatcher>,
}

impl EmbedDatasetState {
    pub(crate) fn new(
        store: Arc<dyn ApiStore>,
        caps: EmbedDatasetCaps,
        dispatcher: Arc<dyn EmbedDatasetDispatcher>,
    ) -> Self {
        Self {
            store: ApiStoreHandle::new(store),
            caps,
            dispatcher,
        }
    }
}

/// Build embedding-dataset management routes.
pub fn router() -> Router {
    Router::new()
        .route(
            "/v1/embed-datasets",
            protected(post(create_dataset), atom::EMBED_DATASET_CREATE),
        )
        .route(
            "/v1/embed-datasets",
            protected(get(list_datasets), atom::EMBED_DATASET_READ),
        )
        .route(
            "/v1/embed-datasets/{id}",
            protected(get(read_dataset), atom::EMBED_DATASET_READ),
        )
        .route(
            "/v1/embed-datasets/{id}/update",
            protected(post(update_dataset), atom::EMBED_DATASET_UPDATE),
        )
        .route(
            "/v1/embed-datasets/{id}/retire",
            protected(post(retire_dataset), atom::EMBED_DATASET_RETIRE),
        )
        .route(
            "/v1/embed-datasets/{id}/source-items",
            protected(get(read_source_items), atom::EMBED_DATASET_READ),
        )
        .route(
            "/v1/embed-datasets/{id}/corpus",
            protected(get(read_corpus), atom::EMBED_DATASET_READ),
        )
}

fn protected(route: MethodRouter, permission: &'static str) -> MethodRouter {
    route
        .route_layer(from_fn(authorize))
        .route_layer(Extension(RequiredPermission(permission)))
}

async fn create_dataset(
    Extension(state): Extension<EmbedDatasetState>,
    Extension(context): Extension<RequestContext>,
    Json(request): Json<CreateDatasetRequest>,
) -> Result<(StatusCode, Json<CreateDatasetResponse>), ApiError> {
    let tenant = require_tenant_principal(&context)?;
    validate_embed_endpoint(&state.store, tenant, request.embed_endpoint_id).await?;
    let encoded_source_items = validate_and_encode_source_items(&request.items, state.caps)?;
    let item_count = item_count_i64(request.items.len())?;
    let dataset_id = state
        .store
        .create_entity_minting::<EmbeddingDataset>()
        .await
        .map_err(store_error)?;
    let display_name_hash =
        put_json(&state.store, &JsonValue::String(request.display_name)).await?;
    let source_items_hash = put_bytes(&state.store, &encoded_source_items).await?;
    let embed_endpoint_hash = put_json(
        &state.store,
        &JsonValue::String(request.embed_endpoint_id.to_string()),
    )
    .await?;

    // TODO(D5, ROADMAP §3.A): dispatch the asynchronous embed job after the
    // lowerer ephemeral-scope support lands.
    let revision = RevisionInput::new()
        .with_content("display_name", display_name_hash)
        .with_content("source_items", source_items_hash)
        .with_content("embed_endpoint_id", embed_endpoint_hash)
        .with_entity(
            "tenant",
            EntityRefValue::pinned(tenant.internal().as_uuid(), 0),
        )
        .with_scalar(
            "status",
            ScalarValue::I64(EmbeddingDatasetStatus::Created.as_i64()),
        )
        .with_scalar("is_retired", ScalarValue::Bool(false))
        .with_scalar("item_count", ScalarValue::I64(item_count));

    state
        .store
        .append_revision_typed::<EmbeddingDataset>(dataset_id, 0, &revision)
        .await
        .map_err(store_error)?;
    state.dispatcher.dispatch_embed_dataset(dataset_id, tenant);

    Ok((
        StatusCode::CREATED,
        Json(CreateDatasetResponse {
            dataset_id: dataset_id.public().as_uuid(),
        }),
    ))
}

async fn list_datasets(
    Extension(state): Extension<EmbedDatasetState>,
    Extension(context): Extension<RequestContext>,
    Query(params): Query<PaginationParams>,
) -> Result<Json<PaginatedResponse<DatasetSummaryResponse>>, ApiError> {
    let tenant = require_tenant_principal(&context)?;
    let mut rows = Vec::new();
    rows.extend(
        state
            .store
            .find_by_scalar_typed::<EmbeddingDataset>("is_retired", &ScalarValue::Bool(false))
            .await
            .map_err(store_error)?,
    );
    rows.extend(
        state
            .store
            .find_by_scalar_typed::<EmbeddingDataset>("is_retired", &ScalarValue::Bool(true))
            .await
            .map_err(store_error)?,
    );
    dedupe_rows(&mut rows);

    let items = dataset_items(&state.store, rows, tenant).await?;
    Ok(Json(paginate_items(items, &params)?))
}

async fn read_dataset(
    Extension(state): Extension<EmbedDatasetState>,
    Extension(context): Extension<RequestContext>,
    Path(id): Path<Uuid>,
) -> Result<Json<DatasetSummaryResponse>, ApiError> {
    let tenant = require_tenant_principal(&context)?;
    let dataset_id = resolve_public_dataset(&state.store, id).await?;
    let row = state
        .store
        .get_entity_typed::<EmbeddingDataset>(dataset_id)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::NotFound("embedding dataset not found".to_string()))?;
    let latest = latest_revision::<EmbeddingDataset>(&state.store, dataset_id).await?;
    ensure_revision_tenant(&latest, tenant, "embedding dataset")?;
    Ok(Json(
        dataset_summary_response(&state.store, &row, &latest).await?,
    ))
}

async fn update_dataset(
    Extension(state): Extension<EmbedDatasetState>,
    Extension(context): Extension<RequestContext>,
    Path(id): Path<Uuid>,
    Json(request): Json<UpdateDatasetRequest>,
) -> Result<Response, ApiError> {
    let tenant = require_tenant_principal(&context)?;
    let encoded_source_items = validate_and_encode_source_items(&request.items, state.caps)?;
    let item_count = item_count_i64(request.items.len())?;
    let dataset_id = resolve_public_dataset(&state.store, id).await?;
    let row = state
        .store
        .get_entity_typed::<EmbeddingDataset>(dataset_id)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::NotFound("embedding dataset not found".to_string()))?;
    let latest = latest_revision::<EmbeddingDataset>(&state.store, dataset_id).await?;
    ensure_revision_tenant(&latest, tenant, "embedding dataset")?;
    if status_from_revision(&latest)? == EmbeddingDatasetStatus::Embedding {
        return Err(ApiError::Conflict(
            "dataset is currently embedding".to_string(),
        ));
    }
    let source_items_hash = put_bytes(&state.store, &encoded_source_items).await?;
    let next_revision_seq = next_revision_seq(&latest, "embedding dataset")?;
    let tenant_ref = required_entity_ref(&latest, "tenant")?;

    // TODO(D5, ROADMAP §3.A): this transition intentionally marks the dataset
    // Embedding before the background embed job exists, matching the design's
    // update lifecycle and preserving the previous corpus for workflow reads.
    let mut revision = RevisionInput::new()
        .with_content(
            "display_name",
            required_content_hash(&latest, "display_name")?,
        )
        .with_content("source_items", source_items_hash)
        .with_content(
            "embed_endpoint_id",
            required_content_hash(&latest, "embed_endpoint_id")?,
        )
        .with_entity("tenant", tenant_ref)
        .with_scalar(
            "status",
            ScalarValue::I64(EmbeddingDatasetStatus::Embedding.as_i64()),
        )
        .with_scalar(
            "is_retired",
            ScalarValue::Bool(bool_scalar(&latest, "is_retired")?),
        )
        .with_scalar("item_count", ScalarValue::I64(item_count));
    if let Some(hash) = optional_content_hash(&latest, "corpus") {
        revision = revision.with_content("corpus", hash);
    }

    state
        .store
        .append_revision_typed::<EmbeddingDataset>(dataset_id, next_revision_seq, &revision)
        .await
        .map_err(store_error)?;
    state.dispatcher.dispatch_embed_dataset(dataset_id, tenant);
    let updated = latest_revision::<EmbeddingDataset>(&state.store, dataset_id).await?;
    Ok(Json(dataset_summary_response(&state.store, &row, &updated).await?).into_response())
}

async fn retire_dataset(
    Extension(state): Extension<EmbedDatasetState>,
    Extension(context): Extension<RequestContext>,
    Path(id): Path<Uuid>,
) -> Result<Json<RetireDatasetResponse>, ApiError> {
    let tenant = require_tenant_principal(&context)?;
    let dataset_id = resolve_public_dataset(&state.store, id).await?;
    let latest = latest_revision::<EmbeddingDataset>(&state.store, dataset_id).await?;
    ensure_revision_tenant(&latest, tenant, "embedding dataset")?;
    let next_revision_seq = next_revision_seq(&latest, "embedding dataset")?;
    let tenant_ref = required_entity_ref(&latest, "tenant")?;
    let mut revision = RevisionInput::new()
        .with_content(
            "display_name",
            required_content_hash(&latest, "display_name")?,
        )
        .with_content(
            "source_items",
            required_content_hash(&latest, "source_items")?,
        )
        .with_content(
            "embed_endpoint_id",
            required_content_hash(&latest, "embed_endpoint_id")?,
        )
        .with_entity("tenant", tenant_ref)
        .with_scalar(
            "status",
            ScalarValue::I64(status_from_revision(&latest)?.as_i64()),
        )
        .with_scalar("is_retired", ScalarValue::Bool(true))
        .with_scalar(
            "item_count",
            ScalarValue::I64(i64_scalar(&latest, "item_count")?),
        );
    if let Some(hash) = optional_content_hash(&latest, "corpus") {
        revision = revision.with_content("corpus", hash);
    }

    state
        .store
        .append_revision_typed::<EmbeddingDataset>(dataset_id, next_revision_seq, &revision)
        .await
        .map_err(store_error)?;

    Ok(Json(RetireDatasetResponse {
        dataset_id: id,
        is_retired: true,
    }))
}

async fn read_source_items(
    Extension(state): Extension<EmbedDatasetState>,
    Extension(context): Extension<RequestContext>,
    Path(id): Path<Uuid>,
) -> Result<Json<Vec<SourceItem>>, ApiError> {
    let tenant = require_tenant_principal(&context)?;
    let dataset_id = resolve_public_dataset(&state.store, id).await?;
    let latest = latest_revision::<EmbeddingDataset>(&state.store, dataset_id).await?;
    ensure_revision_tenant(&latest, tenant, "embedding dataset")?;
    let bytes = load_bytes(
        &state.store,
        required_content_hash(&latest, "source_items")?,
        "source items",
    )
    .await?;
    let items = decode_source_items(bytes.bytes())
        .map_err(|error| ApiError::Internal(format!("stored source items are invalid: {error}")))?;
    Ok(Json(items))
}

async fn read_corpus(
    Extension(state): Extension<EmbedDatasetState>,
    Extension(context): Extension<RequestContext>,
    Path(id): Path<Uuid>,
) -> Result<Json<JsonValue>, ApiError> {
    let tenant = require_tenant_principal(&context)?;
    let dataset_id = resolve_public_dataset(&state.store, id).await?;
    let latest = latest_revision::<EmbeddingDataset>(&state.store, dataset_id).await?;
    ensure_revision_tenant(&latest, tenant, "embedding dataset")?;
    let Some(hash) = optional_content_hash(&latest, "corpus") else {
        return Err(ApiError::NotFound("no corpus available".to_string()));
    };
    let bytes = load_bytes(&state.store, hash, "corpus").await?;
    let corpus = decode_corpus(bytes.bytes())
        .map_err(|error| ApiError::Internal(format!("stored corpus is invalid: {error}")))?;
    Ok(Json(corpus_items_to_json(&corpus)?))
}

#[derive(Deserialize)]
struct CreateDatasetRequest {
    display_name: String,
    embed_endpoint_id: Uuid,
    items: Vec<SourceItem>,
}

#[derive(Serialize)]
struct CreateDatasetResponse {
    dataset_id: Uuid,
}

#[derive(Deserialize)]
struct UpdateDatasetRequest {
    items: Vec<SourceItem>,
}

#[derive(Serialize)]
struct DatasetSummaryResponse {
    dataset_id: Uuid,
    display_name: String,
    status: &'static str,
    item_count: i64,
    embed_endpoint_id: Uuid,
    created_at: UnixMillis,
    updated_at: UnixMillis,
    is_retired: bool,
}

#[derive(Serialize)]
struct RetireDatasetResponse {
    dataset_id: Uuid,
    is_retired: bool,
}

fn require_tenant_principal(context: &RequestContext) -> Result<EntityId<crate::Tenant>, ApiError> {
    let tenant = tenant_scope(context)?;
    match context.auth.as_ref() {
        Some(AuthContext::Principal { tenant_id, .. }) if *tenant_id == tenant => Ok(tenant),
        Some(AuthContext::Principal { .. }) => Err(ApiError::Forbidden),
        Some(AuthContext::Ephemeral { .. }) => Err(ApiError::Forbidden),
        None => Err(ApiError::Unauthenticated),
    }
}

fn tenant_scope(context: &RequestContext) -> Result<EntityId<crate::Tenant>, ApiError> {
    match context.scope {
        RequestScope::Tenant(tenant) => Ok(tenant),
        RequestScope::Operator => Err(ApiError::Forbidden),
    }
}

async fn validate_embed_endpoint(
    store: &ApiStoreHandle,
    tenant: EntityId<crate::Tenant>,
    public: Uuid,
) -> Result<(), ApiError> {
    let identity = store
        .resolve_public(public)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::InvalidRequest(format!("endpoint {public} not found")))?;
    let endpoint_id = identity
        .typed::<TenantEndpointConfig>()
        .map_err(|_| ApiError::InvalidRequest(format!("{public} is not an endpoint config")))?;
    let row = store
        .get_entity(endpoint_id.internal().as_uuid())
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::InvalidRequest(format!("endpoint {public} not found")))?;
    if row.kind != TenantEndpointConfig::KIND {
        return Err(ApiError::InvalidRequest(format!(
            "{public} is not an endpoint config"
        )));
    }
    let latest = latest_revision::<TenantEndpointConfig>(store, endpoint_id).await?;
    ensure_revision_tenant(&latest, tenant, "endpoint config")?;
    if bool_scalar(&latest, "is_retired")? {
        return Err(ApiError::InvalidRequest(
            "endpoint config is retired".to_string(),
        ));
    }
    if endpoint_implementation(store, &latest).await? != "embed" {
        return Err(ApiError::InvalidRequest(
            "embed_endpoint_id must reference an embed endpoint".to_string(),
        ));
    }
    Ok(())
}

fn validate_and_encode_source_items(
    items: &[SourceItem],
    caps: EmbedDatasetCaps,
) -> Result<Vec<u8>, ApiError> {
    if items.len() > caps.max_items {
        return Err(ApiError::InvalidRequest(format!(
            "items cap exceeded: {} > {}",
            items.len(),
            caps.max_items
        )));
    }
    for (index, item) in items.iter().enumerate() {
        if item.text.len() > caps.max_text_bytes {
            return Err(ApiError::InvalidRequest(format!(
                "item {index} text cap exceeded: {} > {} bytes",
                item.text.len(),
                caps.max_text_bytes
            )));
        }
        if let Some(payload) = item.payload.as_ref() {
            let payload_len = serde_json::to_vec(payload)
                .map_err(|error| {
                    ApiError::InvalidRequest(format!(
                        "item {index} payload is invalid JSON: {error}"
                    ))
                })?
                .len();
            if payload_len > caps.max_payload_bytes {
                return Err(ApiError::InvalidRequest(format!(
                    "item {index} payload cap exceeded: {payload_len} > {} bytes",
                    caps.max_payload_bytes
                )));
            }
        }
    }
    let encoded = encode_source_items(items)
        .map_err(|error| ApiError::InvalidRequest(format!("source items are invalid: {error}")))?;
    if encoded.len() > caps.max_source_items_blob_bytes {
        return Err(ApiError::InvalidRequest(format!(
            "source_items blob cap exceeded: {} > {} bytes",
            encoded.len(),
            caps.max_source_items_blob_bytes
        )));
    }
    Ok(encoded)
}

async fn resolve_public_dataset(
    store: &ApiStoreHandle,
    public: Uuid,
) -> Result<EntityId<EmbeddingDataset>, ApiError> {
    let identity = store
        .resolve_public(public)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::NotFound("embedding dataset not found".to_string()))?;
    let typed = identity
        .typed::<EmbeddingDataset>()
        .map_err(|_| ApiError::NotFound("embedding dataset not found".to_string()))?;
    let _ = store
        .get_entity_typed::<EmbeddingDataset>(typed)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::NotFound("embedding dataset not found".to_string()))?;
    Ok(typed)
}

async fn latest_revision<T: Entity>(
    store: &ApiStoreHandle,
    id: EntityId<T>,
) -> Result<RevisionRow, ApiError> {
    store
        .get_latest_revision_typed::<T>(id)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::NotFound(format!("{} revision not found", T::NAME)))
}

fn ensure_revision_tenant(
    revision: &RevisionRow,
    tenant: EntityId<crate::Tenant>,
    entity_name: &'static str,
) -> Result<(), ApiError> {
    let tenant_ref = required_entity_ref(revision, "tenant")?;
    if tenant_ref.target_entity_id == tenant.internal().as_uuid() {
        Ok(())
    } else {
        Err(ApiError::NotFound(format!("{entity_name} not found")))
    }
}

async fn put_bytes(store: &ApiStoreHandle, bytes: &[u8]) -> Result<Sha256, ApiError> {
    let content = ContentValue::new(bytes.to_vec());
    let hash = content.digest();
    store.put(&content).await.map_err(store_error)?;
    Ok(hash)
}

async fn put_json(store: &ApiStoreHandle, value: &JsonValue) -> Result<Sha256, ApiError> {
    let canonical = CanonicalJson::from_value(value)
        .map_err(|error| ApiError::InvalidRequest(format!("invalid JSON: {error}")))?;
    let hash = store.put_typed(&canonical).await.map_err(store_error)?;
    Ok(hash.as_digest())
}

async fn load_json(store: &ApiStoreHandle, hash: Sha256) -> Result<JsonValue, ApiError> {
    let typed_hash = ContentHash::<CanonicalJson>::from_digest_unchecked(hash);
    let canonical = store
        .get_typed::<CanonicalJson>(typed_hash)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::Internal("missing JSON content blob".to_string()))?;
    canonical
        .to_deserializable()
        .map_err(|error| ApiError::Internal(format!("invalid stored JSON: {error}")))
}

async fn load_bytes(
    store: &ApiStoreHandle,
    hash: Sha256,
    label: &'static str,
) -> Result<ContentValue, ApiError> {
    store
        .get(hash)
        .await
        .map_err(store_error)?
        .ok_or_else(|| ApiError::Internal(format!("missing {label} content blob")))
}

fn required_content_hash(revision: &RevisionRow, attr: &'static str) -> Result<Sha256, ApiError> {
    revision
        .content_attrs
        .get(attr)
        .copied()
        .ok_or_else(|| ApiError::Internal(format!("missing content attribute {attr}")))
}

fn optional_content_hash(revision: &RevisionRow, attr: &'static str) -> Option<Sha256> {
    revision.content_attrs.get(attr).copied()
}

fn required_entity_ref(
    revision: &RevisionRow,
    attr: &'static str,
) -> Result<EntityRefValue, ApiError> {
    revision
        .entity_attrs
        .get(attr)
        .copied()
        .ok_or_else(|| ApiError::Internal(format!("missing entity attribute {attr}")))
}

fn bool_scalar(revision: &RevisionRow, attr: &'static str) -> Result<bool, ApiError> {
    match revision.scalar_attrs.get(attr) {
        Some(ScalarValue::Bool(value)) => Ok(*value),
        Some(ScalarValue::I64(_)) => Err(ApiError::Internal(format!(
            "invalid scalar type for {attr}: expected bool"
        ))),
        None => Err(ApiError::Internal(format!(
            "missing scalar attribute {attr}"
        ))),
    }
}

fn i64_scalar(revision: &RevisionRow, attr: &'static str) -> Result<i64, ApiError> {
    match revision.scalar_attrs.get(attr) {
        Some(ScalarValue::I64(value)) => Ok(*value),
        Some(ScalarValue::Bool(_)) => Err(ApiError::Internal(format!(
            "invalid scalar type for {attr}: expected i64"
        ))),
        None => Err(ApiError::Internal(format!(
            "missing scalar attribute {attr}"
        ))),
    }
}

fn status_from_revision(revision: &RevisionRow) -> Result<EmbeddingDatasetStatus, ApiError> {
    let value = i64_scalar(revision, "status")?;
    EmbeddingDatasetStatus::try_from(value)
        .map_err(|_| ApiError::Internal(format!("invalid embedding dataset status {value}")))
}

fn status_name(status: EmbeddingDatasetStatus) -> &'static str {
    match status {
        EmbeddingDatasetStatus::Created => "created",
        EmbeddingDatasetStatus::Embedding => "embedding",
        EmbeddingDatasetStatus::Ready => "ready",
        EmbeddingDatasetStatus::Failed => "failed",
    }
}

fn item_count_i64(count: usize) -> Result<i64, ApiError> {
    i64::try_from(count).map_err(|error| {
        ApiError::Internal(format!("source item count conversion failed: {error}"))
    })
}

fn next_revision_seq(revision: &RevisionRow, entity_name: &'static str) -> Result<u64, ApiError> {
    revision
        .revision_seq
        .checked_add(1)
        .ok_or_else(|| ApiError::Internal(format!("{entity_name} revision sequence overflow")))
}

async fn endpoint_implementation(
    store: &ApiStoreHandle,
    revision: &RevisionRow,
) -> Result<String, ApiError> {
    match load_json(store, required_content_hash(revision, "implementation")?).await? {
        JsonValue::String(value) => Ok(value),
        _ => Err(ApiError::Internal(
            "invalid stored endpoint implementation name".to_string(),
        )),
    }
}

async fn display_name(store: &ApiStoreHandle, revision: &RevisionRow) -> Result<String, ApiError> {
    match load_json(store, required_content_hash(revision, "display_name")?).await? {
        JsonValue::String(value) => Ok(value),
        _ => Err(ApiError::Internal(
            "invalid stored dataset display name".to_string(),
        )),
    }
}

async fn embed_endpoint_id(
    store: &ApiStoreHandle,
    revision: &RevisionRow,
) -> Result<Uuid, ApiError> {
    match load_json(store, required_content_hash(revision, "embed_endpoint_id")?).await? {
        JsonValue::String(value) => value
            .parse()
            .map_err(|error| ApiError::Internal(format!("invalid stored endpoint UUID: {error}"))),
        _ => Err(ApiError::Internal(
            "invalid stored dataset endpoint UUID".to_string(),
        )),
    }
}

async fn dataset_items(
    store: &ApiStoreHandle,
    rows: Vec<philharmonic_store::EntityRow>,
    tenant: EntityId<crate::Tenant>,
) -> Result<Vec<(CursorKey, DatasetSummaryResponse)>, ApiError> {
    let mut items = Vec::new();
    for row in rows {
        let dataset_id = row.identity.typed::<EmbeddingDataset>().map_err(|error| {
            ApiError::Internal(format!("invalid embedding dataset identity: {error}"))
        })?;
        let latest = latest_revision::<EmbeddingDataset>(store, dataset_id).await?;
        if required_entity_ref(&latest, "tenant")?.target_entity_id != tenant.internal().as_uuid() {
            continue;
        }
        items.push((
            CursorKey::new(row.created_at, row.identity.public),
            dataset_summary_response(store, &row, &latest).await?,
        ));
    }
    Ok(items)
}

async fn dataset_summary_response(
    store: &ApiStoreHandle,
    row: &philharmonic_store::EntityRow,
    latest: &RevisionRow,
) -> Result<DatasetSummaryResponse, ApiError> {
    Ok(DatasetSummaryResponse {
        dataset_id: row.identity.public,
        display_name: display_name(store, latest).await?,
        status: status_name(status_from_revision(latest)?),
        item_count: i64_scalar(latest, "item_count")?,
        embed_endpoint_id: embed_endpoint_id(store, latest).await?,
        created_at: row.created_at,
        updated_at: latest.created_at,
        is_retired: bool_scalar(latest, "is_retired")?,
    })
}

fn corpus_items_to_json(items: &[CorpusItem]) -> Result<JsonValue, ApiError> {
    let mut values = Vec::with_capacity(items.len());
    for item in items {
        let mut object = JsonMap::new();
        object.insert("id".to_string(), JsonValue::String(item.id.clone()));
        let mut vector = Vec::with_capacity(item.vector.len());
        for value in &item.vector {
            let number = serde_json::Number::from_f64(f64::from(*value)).ok_or_else(|| {
                ApiError::Internal(format!(
                    "corpus item {} has non-finite vector value",
                    item.id
                ))
            })?;
            vector.push(JsonValue::Number(number));
        }
        object.insert("vector".to_string(), JsonValue::Array(vector));
        if let Some(payload) = item.payload.as_ref() {
            object.insert("payload".to_string(), payload.clone());
        }
        values.push(JsonValue::Object(object));
    }
    Ok(JsonValue::Array(values))
}

fn paginate_items<T>(
    mut items: Vec<(CursorKey, T)>,
    params: &PaginationParams,
) -> Result<PaginatedResponse<T>, ApiError> {
    let cursor = decode_cursor(params.cursor.as_deref()).map_err(pagination_error)?;
    items.sort_by(|(a, _), (b, _)| {
        (b.created_at, b.id.as_u128()).cmp(&(a.created_at, a.id.as_u128()))
    });
    if let Some(cursor) = cursor {
        items.retain(|(key, _)| cursor_before(*key, cursor));
    }

    let limit = page_size(params.limit).map_err(pagination_error)?;
    let overfetch = limit
        .checked_add(1)
        .ok_or(ApiError::Internal("pagination limit overflow".to_string()))?;
    let mut page = items.into_iter().take(overfetch).collect::<Vec<_>>();
    let next_cursor = if page.len() > limit {
        page.truncate(limit);
        page.last()
            .map(|(key, _)| encode_cursor(*key))
            .transpose()
            .map_err(pagination_error)?
    } else {
        None
    };

    Ok(PaginatedResponse {
        items: page.into_iter().map(|(_, item)| item).collect(),
        next_cursor,
    })
}

fn cursor_before(key: CursorKey, cursor: CursorKey) -> bool {
    (key.created_at, key.id.as_u128()) < (cursor.created_at, cursor.id.as_u128())
}

fn dedupe_rows(rows: &mut Vec<philharmonic_store::EntityRow>) {
    let mut seen = HashSet::new();
    rows.retain(|row| seen.insert(row.identity.internal));
}

fn pagination_error(error: PaginationError) -> ApiError {
    match error {
        PaginationError::InvalidCursor => ApiError::InvalidRequest(error.to_string()),
        PaginationError::LimitConversion => ApiError::Internal(error.to_string()),
    }
}

fn store_error(error: philharmonic_store::StoreError) -> ApiError {
    ApiError::Internal(error.to_string())
}
