use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
};

use axum::{
    body::Body,
    http::{Request, StatusCode},
};
use philharmonic_api::{ApiStore, EmbedDatasetCaps, EmbedDatasetDispatcher, RequestScope};
use philharmonic_policy::{
    CorpusItem, EmbeddingDataset, EmbeddingDatasetStatus, Principal, PrincipalKind, RoleDefinition,
    RoleMembership, Tenant, TenantEndpointConfig, TenantStatus, TokenHash, atom, encode_corpus,
    encode_source_items, generate_api_token,
};
use philharmonic_store::{ContentStore, EntityRefValue, RevisionRow};
use philharmonic_types::{
    CanonicalJson, ContentValue, EntityId, JsonValue, ScalarValue, Sha256, UnixMillis, Uuid,
};
use serde_json::json;
use tower::ServiceExt;

mod common;

struct Fixture {
    router: axum::Router,
    store: Arc<common::MockStore>,
    principal_token: String,
    tenant: EntityId<Tenant>,
}

async fn fixture(permissions: &[&str]) -> Fixture {
    let store = common::MockStore::new();
    let tenant = active_tenant(&store);
    let (principal_token, token_hash) = generate_api_token();
    let principal = principal(&store, tenant, token_hash_content_hash(token_hash), false);
    let role = role_with_permissions(&store, tenant, permissions).await;
    membership(&store, tenant, principal, role);
    let router = router(store.clone(), tenant);

    Fixture {
        router,
        store,
        principal_token: principal_token.to_string(),
        tenant,
    }
}

fn router(store: Arc<dyn ApiStore>, tenant: EntityId<Tenant>) -> axum::Router {
    router_with_caps(store, tenant, EmbedDatasetCaps::default())
}

fn router_with_caps(
    store: Arc<dyn ApiStore>,
    tenant: EntityId<Tenant>,
    caps: EmbedDatasetCaps,
) -> axum::Router {
    common::builder(
        Arc::new(common::FixedResolver::new(RequestScope::Tenant(tenant))),
        store,
        common::test_api_verifying_key_registry(),
    )
    .embed_dataset_caps(caps)
    .build()
    .unwrap()
    .into_router()
}

fn router_with_dispatcher(
    store: Arc<dyn ApiStore>,
    tenant: EntityId<Tenant>,
    dispatcher: Arc<dyn EmbedDatasetDispatcher>,
) -> axum::Router {
    common::builder(
        Arc::new(common::FixedResolver::new(RequestScope::Tenant(tenant))),
        store,
        common::test_api_verifying_key_registry(),
    )
    .embed_dataset_dispatcher(dispatcher)
    .build()
    .unwrap()
    .into_router()
}

#[derive(Default)]
struct RecordingDispatcher {
    calls: Mutex<Vec<(EntityId<EmbeddingDataset>, EntityId<Tenant>)>>,
}

impl RecordingDispatcher {
    fn calls(&self) -> Vec<(EntityId<EmbeddingDataset>, EntityId<Tenant>)> {
        self.calls.lock().unwrap().clone()
    }
}

impl EmbedDatasetDispatcher for RecordingDispatcher {
    fn dispatch_embed_dataset(
        &self,
        dataset_id: EntityId<EmbeddingDataset>,
        tenant_id: EntityId<Tenant>,
    ) {
        self.calls.lock().unwrap().push((dataset_id, tenant_id));
    }
}

fn active_tenant(store: &common::MockStore) -> EntityId<Tenant> {
    let tenant = common::new_typed_id::<Tenant>();
    store.insert_entity(tenant);
    store.insert_revision(RevisionRow {
        entity_id: tenant.internal().as_uuid(),
        revision_seq: 0,
        created_at: UnixMillis(1),
        content_attrs: HashMap::new(),
        entity_attrs: HashMap::new(),
        scalar_attrs: HashMap::from([(
            "status".to_string(),
            ScalarValue::I64(TenantStatus::Active.as_i64()),
        )]),
    });
    tenant
}

fn principal(
    store: &common::MockStore,
    tenant: EntityId<Tenant>,
    credential_hash: Sha256,
    is_retired: bool,
) -> EntityId<Principal> {
    let principal = common::new_typed_id::<Principal>();
    store.insert_entity(principal);
    store.insert_revision(RevisionRow {
        entity_id: principal.internal().as_uuid(),
        revision_seq: 0,
        created_at: UnixMillis(2),
        content_attrs: HashMap::from([("credential_hash".to_string(), credential_hash)]),
        entity_attrs: HashMap::from([(
            "tenant".to_string(),
            EntityRefValue::pinned(tenant.internal().as_uuid(), 0),
        )]),
        scalar_attrs: HashMap::from([
            (
                "kind".to_string(),
                ScalarValue::I64(PrincipalKind::User.as_i64()),
            ),
            ("epoch".to_string(), ScalarValue::I64(0)),
            ("is_retired".to_string(), ScalarValue::Bool(is_retired)),
        ]),
    });
    principal
}

async fn role_with_permissions(
    store: &common::MockStore,
    tenant: EntityId<Tenant>,
    permissions: &[&str],
) -> EntityId<RoleDefinition> {
    let role = common::new_typed_id::<RoleDefinition>();
    let permissions_hash = put_content(store, &serde_json::to_vec(permissions).unwrap()).await;
    store.insert_entity(role);
    store.insert_revision(RevisionRow {
        entity_id: role.internal().as_uuid(),
        revision_seq: 0,
        created_at: UnixMillis(3),
        content_attrs: HashMap::from([("permissions".to_string(), permissions_hash)]),
        entity_attrs: HashMap::from([(
            "tenant".to_string(),
            EntityRefValue::pinned(tenant.internal().as_uuid(), 0),
        )]),
        scalar_attrs: HashMap::from([("is_retired".to_string(), ScalarValue::Bool(false))]),
    });
    role
}

fn membership(
    store: &common::MockStore,
    tenant: EntityId<Tenant>,
    principal: EntityId<Principal>,
    role: EntityId<RoleDefinition>,
) {
    let membership = common::new_typed_id::<RoleMembership>();
    store.insert_entity(membership);
    store.insert_revision(RevisionRow {
        entity_id: membership.internal().as_uuid(),
        revision_seq: 0,
        created_at: UnixMillis(4),
        content_attrs: HashMap::new(),
        entity_attrs: HashMap::from([
            (
                "tenant".to_string(),
                EntityRefValue::pinned(tenant.internal().as_uuid(), 0),
            ),
            (
                "principal".to_string(),
                EntityRefValue::pinned(principal.internal().as_uuid(), 0),
            ),
            (
                "role".to_string(),
                EntityRefValue::pinned(role.internal().as_uuid(), 0),
            ),
        ]),
        scalar_attrs: HashMap::from([("is_retired".to_string(), ScalarValue::Bool(false))]),
    });
}

async fn put_content(store: &common::MockStore, bytes: &[u8]) -> Sha256 {
    let value = ContentValue::new(bytes.to_vec());
    let hash = value.digest();
    store.put(&value).await.unwrap();
    hash
}

async fn put_json(store: &common::MockStore, value: &JsonValue) -> Sha256 {
    let canonical = CanonicalJson::from_value(value).unwrap();
    put_content(store, canonical.as_bytes()).await
}

fn token_hash_content_hash(token_hash: TokenHash) -> Sha256 {
    ContentValue::new(token_hash.0.to_vec()).digest()
}

fn json_request(method: &str, uri: &str, token: &str, body: JsonValue) -> Request<Body> {
    Request::builder()
        .method(method)
        .uri(uri)
        .header("Authorization", format!("Bearer {token}"))
        .header("Content-Type", "application/json")
        .body(Body::from(serde_json::to_vec(&body).unwrap()))
        .unwrap()
}

fn empty_request(method: &str, uri: &str, token: &str) -> Request<Body> {
    Request::builder()
        .method(method)
        .uri(uri)
        .header("Authorization", format!("Bearer {token}"))
        .body(Body::empty())
        .unwrap()
}

async fn response_json(response: axum::response::Response) -> JsonValue {
    let body = axum::body::to_bytes(response.into_body(), usize::MAX)
        .await
        .unwrap();
    serde_json::from_slice(&body).unwrap()
}

async fn seed_endpoint(
    store: &common::MockStore,
    tenant: EntityId<Tenant>,
    implementation: &str,
    is_retired: bool,
) -> EntityId<TenantEndpointConfig> {
    let endpoint = common::new_typed_id::<TenantEndpointConfig>();
    store.insert_entity(endpoint);
    let display_hash = put_json(store, &json!("Embed endpoint")).await;
    let encrypted_hash = put_content(store, b"not-read-by-embed-dataset-routes").await;
    let implementation_hash = put_json(store, &json!(implementation)).await;
    store.insert_revision(RevisionRow {
        entity_id: endpoint.internal().as_uuid(),
        revision_seq: 0,
        created_at: UnixMillis(5),
        content_attrs: HashMap::from([
            ("display_name".to_string(), display_hash),
            ("encrypted_config".to_string(), encrypted_hash),
            ("implementation".to_string(), implementation_hash),
        ]),
        entity_attrs: HashMap::from([(
            "tenant".to_string(),
            EntityRefValue::pinned(tenant.internal().as_uuid(), 0),
        )]),
        scalar_attrs: HashMap::from([
            ("key_version".to_string(), ScalarValue::I64(1)),
            ("is_retired".to_string(), ScalarValue::Bool(is_retired)),
        ]),
    });
    endpoint
}

async fn seed_dataset(
    store: &common::MockStore,
    tenant: EntityId<Tenant>,
    endpoint_id: Uuid,
    status: EmbeddingDatasetStatus,
    is_retired: bool,
    corpus: Option<Vec<CorpusItem>>,
) -> EntityId<EmbeddingDataset> {
    let dataset = common::new_typed_id::<EmbeddingDataset>();
    store.insert_entity(dataset);
    let display_hash = put_json(store, &json!("Seeded dataset")).await;
    let source_items = vec![philharmonic_policy::SourceItem {
        id: "doc-1".to_string(),
        text: "hello".to_string(),
        payload: Some(json!({ "rank": 1 })),
    }];
    let source_hash = put_content(store, &encode_source_items(&source_items).unwrap()).await;
    let endpoint_hash = put_json(store, &json!(endpoint_id.to_string())).await;
    let mut content_attrs = HashMap::from([
        ("display_name".to_string(), display_hash),
        ("source_items".to_string(), source_hash),
        ("embed_endpoint_id".to_string(), endpoint_hash),
    ]);
    if let Some(corpus) = corpus {
        content_attrs.insert(
            "corpus".to_string(),
            put_content(store, &encode_corpus(&corpus).unwrap()).await,
        );
    }
    store.insert_revision(RevisionRow {
        entity_id: dataset.internal().as_uuid(),
        revision_seq: 0,
        created_at: UnixMillis(6),
        content_attrs,
        entity_attrs: HashMap::from([(
            "tenant".to_string(),
            EntityRefValue::pinned(tenant.internal().as_uuid(), 0),
        )]),
        scalar_attrs: HashMap::from([
            ("status".to_string(), ScalarValue::I64(status.as_i64())),
            ("is_retired".to_string(), ScalarValue::Bool(is_retired)),
            ("item_count".to_string(), ScalarValue::I64(1)),
        ]),
    });
    dataset
}

async fn create_dataset(router: axum::Router, token: &str, endpoint_id: Uuid) -> Uuid {
    let response = router
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            token,
            json!({
                "display_name": "Knowledge Base",
                "embed_endpoint_id": endpoint_id,
                "items": [{ "id": "doc-1", "text": "Hello", "payload": { "kind": "faq" } }]
            }),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::CREATED);
    response_json(response).await["dataset_id"]
        .as_str()
        .unwrap()
        .parse()
        .unwrap()
}

#[tokio::test]
async fn create_list_read_source_items_and_retire_flow() {
    let fixture = fixture(&[
        atom::EMBED_DATASET_CREATE,
        atom::EMBED_DATASET_READ,
        atom::EMBED_DATASET_RETIRE,
    ])
    .await;
    let endpoint = seed_endpoint(&fixture.store, fixture.tenant, "embed", false).await;
    let dataset_id = create_dataset(
        fixture.router.clone(),
        &fixture.principal_token,
        endpoint.public().as_uuid(),
    )
    .await;

    let response = fixture
        .router
        .clone()
        .oneshot(empty_request(
            "GET",
            "/v1/embed-datasets",
            &fixture.principal_token,
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);
    assert_eq!(
        response_json(response).await["items"][0]["dataset_id"],
        dataset_id.to_string()
    );

    let response = fixture
        .router
        .clone()
        .oneshot(empty_request(
            "GET",
            &format!("/v1/embed-datasets/{dataset_id}"),
            &fixture.principal_token,
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);
    assert_eq!(response_json(response).await["status"], "created");

    let response = fixture
        .router
        .clone()
        .oneshot(empty_request(
            "GET",
            &format!("/v1/embed-datasets/{dataset_id}/source-items"),
            &fixture.principal_token,
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);
    assert_eq!(response_json(response).await[0]["id"], "doc-1");

    let response = fixture
        .router
        .oneshot(empty_request(
            "POST",
            &format!("/v1/embed-datasets/{dataset_id}/retire"),
            &fixture.principal_token,
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);
    assert_eq!(response_json(response).await["is_retired"], true);
}

#[tokio::test]
async fn update_while_embedding_returns_conflict() {
    let fixture = fixture(&[atom::EMBED_DATASET_UPDATE]).await;
    let endpoint = seed_endpoint(&fixture.store, fixture.tenant, "embed", false).await;
    let dataset = seed_dataset(
        &fixture.store,
        fixture.tenant,
        endpoint.public().as_uuid(),
        EmbeddingDatasetStatus::Embedding,
        false,
        None,
    )
    .await;

    let response = fixture
        .router
        .oneshot(json_request(
            "POST",
            &format!("/v1/embed-datasets/{}/update", dataset.public().as_uuid()),
            &fixture.principal_token,
            json!({ "items": [{ "id": "doc-2", "text": "New" }] }),
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::CONFLICT);
    let body = response_json(response).await;
    assert_eq!(body["error"]["code"], "conflict");
    assert_eq!(body["error"]["message"], "dataset is currently embedding");
}

#[tokio::test]
async fn create_dispatches_embed_job_after_revision_write() {
    let store = common::MockStore::new();
    let tenant = active_tenant(&store);
    let (principal_token, token_hash) = generate_api_token();
    let principal = principal(&store, tenant, token_hash_content_hash(token_hash), false);
    let role = role_with_permissions(&store, tenant, &[atom::EMBED_DATASET_CREATE]).await;
    membership(&store, tenant, principal, role);
    let dispatcher = Arc::new(RecordingDispatcher::default());
    let router = router_with_dispatcher(store.clone(), tenant, dispatcher.clone());
    let endpoint = seed_endpoint(&store, tenant, "embed", false).await;

    let dataset_id = create_dataset(router, &principal_token, endpoint.public().as_uuid()).await;

    let calls = dispatcher.calls();
    assert_eq!(calls.len(), 1);
    assert_eq!(calls[0].0.public().as_uuid(), dataset_id);
    assert_eq!(calls[0].1, tenant);
}

#[tokio::test]
async fn update_dispatches_embed_job_after_embedding_revision_write() {
    let store = common::MockStore::new();
    let tenant = active_tenant(&store);
    let (principal_token, token_hash) = generate_api_token();
    let principal = principal(&store, tenant, token_hash_content_hash(token_hash), false);
    let role = role_with_permissions(&store, tenant, &[atom::EMBED_DATASET_UPDATE]).await;
    membership(&store, tenant, principal, role);
    let dispatcher = Arc::new(RecordingDispatcher::default());
    let router = router_with_dispatcher(store.clone(), tenant, dispatcher.clone());
    let endpoint = seed_endpoint(&store, tenant, "embed", false).await;
    let corpus = vec![CorpusItem {
        id: "doc-1".to_string(),
        vector: vec![1.0],
        payload: None,
    }];
    let dataset = seed_dataset(
        &store,
        tenant,
        endpoint.public().as_uuid(),
        EmbeddingDatasetStatus::Ready,
        false,
        Some(corpus),
    )
    .await;

    let response = router
        .oneshot(json_request(
            "POST",
            &format!("/v1/embed-datasets/{}/update", dataset.public().as_uuid()),
            &principal_token,
            json!({ "items": [{ "id": "doc-2", "text": "New" }] }),
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::OK);
    let calls = dispatcher.calls();
    assert_eq!(calls, vec![(dataset, tenant)]);
}

#[tokio::test]
async fn corpus_without_corpus_slot_returns_not_found() {
    let fixture = fixture(&[atom::EMBED_DATASET_READ]).await;
    let endpoint = seed_endpoint(&fixture.store, fixture.tenant, "embed", false).await;
    let dataset = seed_dataset(
        &fixture.store,
        fixture.tenant,
        endpoint.public().as_uuid(),
        EmbeddingDatasetStatus::Created,
        false,
        None,
    )
    .await;

    let response = fixture
        .router
        .oneshot(empty_request(
            "GET",
            &format!("/v1/embed-datasets/{}/corpus", dataset.public().as_uuid()),
            &fixture.principal_token,
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn create_rejects_item_text_payload_and_endpoint_errors() {
    let fixture = fixture(&[atom::EMBED_DATASET_CREATE]).await;
    let endpoint = seed_endpoint(&fixture.store, fixture.tenant, "embed", false).await;

    let too_many = (0..=10_000)
        .map(|index| json!({ "id": format!("doc-{index}"), "text": "x" }))
        .collect::<Vec<_>>();
    let response = fixture
        .router
        .clone()
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &fixture.principal_token,
            json!({ "display_name": "Too many", "embed_endpoint_id": endpoint.public().as_uuid(), "items": too_many }),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    let response = fixture
        .router
        .clone()
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &fixture.principal_token,
            json!({
                "display_name": "Text cap",
                "embed_endpoint_id": endpoint.public().as_uuid(),
                "items": [{ "id": "doc-1", "text": "x".repeat(65_537) }]
            }),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    let response = fixture
        .router
        .clone()
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &fixture.principal_token,
            json!({
                "display_name": "Payload cap",
                "embed_endpoint_id": endpoint.public().as_uuid(),
                "items": [{ "id": "doc-1", "text": "x", "payload": { "blob": "x".repeat(65_537) } }]
            }),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    let llm_endpoint = seed_endpoint(&fixture.store, fixture.tenant, "llm", false).await;
    let response = fixture
        .router
        .clone()
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &fixture.principal_token,
            json!({
                "display_name": "Wrong impl",
                "embed_endpoint_id": llm_endpoint.public().as_uuid(),
                "items": [{ "id": "doc-1", "text": "x" }]
            }),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    let response = fixture
        .router
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &fixture.principal_token,
            json!({
                "display_name": "Missing endpoint",
                "embed_endpoint_id": Uuid::new_v4(),
                "items": [{ "id": "doc-1", "text": "x" }]
            }),
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn create_uses_configured_item_cap() {
    let store = common::MockStore::new();
    let tenant = active_tenant(&store);
    let (principal_token, token_hash) = generate_api_token();
    let principal = principal(&store, tenant, token_hash_content_hash(token_hash), false);
    let role = role_with_permissions(&store, tenant, &[atom::EMBED_DATASET_CREATE]).await;
    membership(&store, tenant, principal, role);
    let router = router_with_caps(
        store.clone(),
        tenant,
        EmbedDatasetCaps {
            max_items: 5,
            ..EmbedDatasetCaps::default()
        },
    );
    let endpoint = seed_endpoint(&store, tenant, "embed", false).await;
    let items = (0..=5)
        .map(|index| json!({ "id": format!("doc-{index}"), "text": "x" }))
        .collect::<Vec<_>>();

    let response = router
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &principal_token,
            json!({ "display_name": "Too many", "embed_endpoint_id": endpoint.public().as_uuid(), "items": items }),
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    assert_eq!(
        response_json(response).await["error"]["message"],
        "items cap exceeded: 6 > 5"
    );
}

#[tokio::test]
async fn create_uses_configured_text_cap() {
    let store = common::MockStore::new();
    let tenant = active_tenant(&store);
    let (principal_token, token_hash) = generate_api_token();
    let principal = principal(&store, tenant, token_hash_content_hash(token_hash), false);
    let role = role_with_permissions(&store, tenant, &[atom::EMBED_DATASET_CREATE]).await;
    membership(&store, tenant, principal, role);
    let router = router_with_caps(
        store.clone(),
        tenant,
        EmbedDatasetCaps {
            max_text_bytes: 5,
            ..EmbedDatasetCaps::default()
        },
    );
    let endpoint = seed_endpoint(&store, tenant, "embed", false).await;

    let response = router
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &principal_token,
            json!({
                "display_name": "Text cap",
                "embed_endpoint_id": endpoint.public().as_uuid(),
                "items": [{ "id": "doc-1", "text": "123456" }]
            }),
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    assert_eq!(
        response_json(response).await["error"]["message"],
        "item 0 text cap exceeded: 6 > 5 bytes"
    );
}

#[tokio::test]
async fn create_uses_configured_payload_cap() {
    let store = common::MockStore::new();
    let tenant = active_tenant(&store);
    let (principal_token, token_hash) = generate_api_token();
    let principal = principal(&store, tenant, token_hash_content_hash(token_hash), false);
    let role = role_with_permissions(&store, tenant, &[atom::EMBED_DATASET_CREATE]).await;
    membership(&store, tenant, principal, role);
    let router = router_with_caps(
        store.clone(),
        tenant,
        EmbedDatasetCaps {
            max_payload_bytes: 10,
            ..EmbedDatasetCaps::default()
        },
    );
    let endpoint = seed_endpoint(&store, tenant, "embed", false).await;

    let response = router
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &principal_token,
            json!({
                "display_name": "Payload cap",
                "embed_endpoint_id": endpoint.public().as_uuid(),
                "items": [{ "id": "doc-1", "text": "x", "payload": { "blob": "abcdef" } }]
            }),
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    assert_eq!(
        response_json(response).await["error"]["message"],
        "item 0 payload cap exceeded: 17 > 10 bytes"
    );
}

#[tokio::test]
async fn create_uses_configured_source_items_blob_cap() {
    let store = common::MockStore::new();
    let tenant = active_tenant(&store);
    let (principal_token, token_hash) = generate_api_token();
    let principal = principal(&store, tenant, token_hash_content_hash(token_hash), false);
    let role = role_with_permissions(&store, tenant, &[atom::EMBED_DATASET_CREATE]).await;
    membership(&store, tenant, principal, role);
    let router = router_with_caps(
        store.clone(),
        tenant,
        EmbedDatasetCaps {
            max_source_items_blob_bytes: 20,
            ..EmbedDatasetCaps::default()
        },
    );
    let endpoint = seed_endpoint(&store, tenant, "embed", false).await;

    let response = router
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &principal_token,
            json!({
                "display_name": "Blob cap",
                "embed_endpoint_id": endpoint.public().as_uuid(),
                "items": [{ "id": "doc-1", "text": "this encoded blob exceeds twenty bytes" }]
            }),
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    assert!(
        response_json(response).await["error"]["message"]
            .as_str()
            .unwrap()
            .contains(" > 20 bytes")
    );
}

#[tokio::test]
async fn cross_tenant_read_returns_not_found() {
    let fixture = fixture(&[atom::EMBED_DATASET_READ]).await;
    let other_tenant = active_tenant(&fixture.store);
    let endpoint = seed_endpoint(&fixture.store, other_tenant, "embed", false).await;
    let dataset = seed_dataset(
        &fixture.store,
        other_tenant,
        endpoint.public().as_uuid(),
        EmbeddingDatasetStatus::Created,
        false,
        None,
    )
    .await;

    let response = fixture
        .router
        .oneshot(empty_request(
            "GET",
            &format!("/v1/embed-datasets/{}", dataset.public().as_uuid()),
            &fixture.principal_token,
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn create_without_permission_is_forbidden() {
    let fixture = fixture(&[atom::EMBED_DATASET_READ]).await;
    let endpoint = seed_endpoint(&fixture.store, fixture.tenant, "embed", false).await;

    let response = fixture
        .router
        .oneshot(json_request(
            "POST",
            "/v1/embed-datasets",
            &fixture.principal_token,
            json!({
                "display_name": "Forbidden",
                "embed_endpoint_id": endpoint.public().as_uuid(),
                "items": [{ "id": "doc-1", "text": "x" }]
            }),
        ))
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::FORBIDDEN);
}
