use crate::domain::model::{ListQuery, Upstream};
use crate::domain::repo::{RepositoryError, UpstreamRepository};
use async_trait::async_trait;
use dashmap::DashMap;
use toolkit_macros::domain_model;
use uuid::Uuid;

/// In-memory upstream repository backed by `DashMap`.
#[domain_model]
pub struct InMemoryUpstreamRepo {
    /// Primary store: id -> Upstream.
    store: DashMap<Uuid, Upstream>,
    /// Alias index: (tenant_id, alias) -> upstream_id.
    alias_index: DashMap<(Uuid, String), Uuid>,
}

impl InMemoryUpstreamRepo {
    #[must_use]
    pub fn new() -> Self {
        Self {
            store: DashMap::new(),
            alias_index: DashMap::new(),
        }
    }
}

impl Default for InMemoryUpstreamRepo {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl UpstreamRepository for InMemoryUpstreamRepo {
    async fn create(&self, upstream: Upstream) -> Result<Upstream, RepositoryError> {
        let alias_key = (upstream.tenant_id, upstream.alias.clone());

        // Atomic alias uniqueness check via entry API.
        match self.alias_index.entry(alias_key) {
            dashmap::mapref::entry::Entry::Occupied(_) => {
                return Err(RepositoryError::Conflict {
                    entity: "upstream",
                    resource: upstream.alias.clone(),
                    detail: format!("alias '{}' already exists for tenant", upstream.alias),
                });
            }
            dashmap::mapref::entry::Entry::Vacant(entry) => {
                entry.insert(upstream.id);
            }
        }

        self.store.insert(upstream.id, upstream.clone());
        Ok(upstream)
    }

    async fn get_by_id(&self, tenant_id: Uuid, id: Uuid) -> Result<Upstream, RepositoryError> {
        self.store
            .get(&id)
            .filter(|u| u.tenant_id == tenant_id)
            .map(|u| u.clone())
            .ok_or(RepositoryError::NotFound {
                entity: "upstream",
                id,
            })
    }

    async fn get_by_alias(
        &self,
        tenant_id: Uuid,
        alias: &str,
    ) -> Result<Upstream, RepositoryError> {
        let id = self
            .alias_index
            .get(&(tenant_id, alias.to_string()))
            .map(|r| *r.value())
            .ok_or(RepositoryError::NotFound {
                entity: "upstream",
                id: Uuid::nil(),
            })?;
        self.get_by_id(tenant_id, id).await
    }

    async fn list(
        &self,
        tenant_id: Uuid,
        query: &ListQuery,
    ) -> Result<Vec<Upstream>, RepositoryError> {
        let mut all: Vec<Upstream> = self
            .store
            .iter()
            .filter(|e| e.value().tenant_id == tenant_id)
            .map(|e| e.value().clone())
            .collect();

        all.sort_by_key(|u| u.id);

        let skip = query.skip as usize;
        let top = query.top as usize;
        Ok(all.into_iter().skip(skip).take(top).collect())
    }

    async fn update(&self, upstream: Upstream) -> Result<Upstream, RepositoryError> {
        let id = upstream.id;
        let tenant_id = upstream.tenant_id;

        // Get the old upstream to remove old alias if changed.
        let old = self
            .store
            .get(&id)
            .filter(|u| u.tenant_id == tenant_id)
            .map(|u| u.clone())
            .ok_or(RepositoryError::NotFound {
                entity: "upstream",
                id,
            })?;

        // If alias changed, swap in the alias index.
        // Note: we avoid using entry() + remove() on the same DashMap because
        // holding a shard lock from entry() while remove() tries to lock
        // another key can deadlock if both keys hash to the same shard.
        if old.alias != upstream.alias {
            let new_alias_key = (tenant_id, upstream.alias.clone());
            if self.alias_index.contains_key(&new_alias_key) {
                return Err(RepositoryError::Conflict {
                    entity: "upstream",
                    resource: upstream.alias.clone(),
                    detail: format!("alias '{}' already exists for tenant", upstream.alias),
                });
            }
            self.alias_index.remove(&(tenant_id, old.alias.clone()));
            self.alias_index.insert(new_alias_key, id);
        }

        self.store.insert(id, upstream.clone());
        Ok(upstream)
    }

    async fn delete(&self, tenant_id: Uuid, id: Uuid) -> Result<(), RepositoryError> {
        // Atomically remove first, then verify tenant ownership.
        let (_, upstream) = self.store.remove(&id).ok_or(RepositoryError::NotFound {
            entity: "upstream",
            id,
        })?;

        if upstream.tenant_id != tenant_id {
            // Wrong tenant — put it back and report not-found.
            self.store.insert(id, upstream);
            return Err(RepositoryError::NotFound {
                entity: "upstream",
                id,
            });
        }

        self.alias_index.remove(&(tenant_id, upstream.alias));
        Ok(())
    }

    async fn list_by_alias_for_tenants(
        &self,
        alias: &str,
        tenant_ids: &std::collections::HashSet<Uuid>,
    ) -> Result<Vec<Upstream>, RepositoryError> {
        Ok(self
            .alias_index
            .iter()
            .filter(|e| e.key().1 == alias && tenant_ids.contains(&e.key().0))
            .filter_map(|e| self.store.get(e.value()).map(|u| u.value().clone()))
            .collect())
    }
}

#[cfg(test)]
mod tests {
    use crate::domain::model::{Endpoint, Scheme, Server};

    use super::*;

    fn make_upstream(tenant_id: Uuid, alias: &str) -> Upstream {
        Upstream {
            id: Uuid::new_v4(),
            tenant_id,
            alias: alias.into(),
            server: Server {
                endpoints: vec![Endpoint {
                    scheme: Scheme::Https,
                    host: "api.openai.com".into(),
                    port: 443,
                }],
            },
            protocol: "gts.cf.core.oagw.protocol.v1~cf.core.oagw.http.v1".into(),
            enabled: true,
            auth: None,
            headers: None,
            plugins: None,
            rate_limit: None,
            cors: None,
            tags: vec![],
        }
    }

    #[tokio::test]
    async fn create_and_get_round_trip() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();
        let u = make_upstream(tenant, "openai");
        let id = u.id;

        let created = repo.create(u).await.unwrap();
        assert_eq!(created.alias, "openai");

        let fetched = repo.get_by_id(tenant, id).await.unwrap();
        assert_eq!(fetched.id, id);
        assert_eq!(fetched.alias, "openai");
    }

    #[tokio::test]
    async fn get_by_alias() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();
        let u = make_upstream(tenant, "openai");
        repo.create(u.clone()).await.unwrap();

        let fetched = repo.get_by_alias(tenant, "openai").await.unwrap();
        assert_eq!(fetched.id, u.id);
    }

    #[tokio::test]
    async fn alias_uniqueness_same_tenant() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();

        repo.create(make_upstream(tenant, "openai")).await.unwrap();
        let err = repo.create(make_upstream(tenant, "openai")).await;
        assert!(matches!(err, Err(RepositoryError::Conflict { .. })));
    }

    #[tokio::test]
    async fn alias_allowed_different_tenant() {
        let repo = InMemoryUpstreamRepo::new();
        let t1 = Uuid::new_v4();
        let t2 = Uuid::new_v4();

        repo.create(make_upstream(t1, "openai")).await.unwrap();
        repo.create(make_upstream(t2, "openai")).await.unwrap();
    }

    #[tokio::test]
    async fn update_preserves_id() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();
        let mut u = make_upstream(tenant, "openai");
        let id = u.id;
        repo.create(u.clone()).await.unwrap();

        u.alias = "openai-v2".into();
        let updated = repo.update(u).await.unwrap();
        assert_eq!(updated.id, id);
        assert_eq!(updated.alias, "openai-v2");

        // Old alias should not resolve.
        assert!(repo.get_by_alias(tenant, "openai").await.is_err());
        // New alias should resolve.
        assert!(repo.get_by_alias(tenant, "openai-v2").await.is_ok());
    }

    #[tokio::test]
    async fn update_to_duplicate_alias_returns_conflict_and_preserves_original() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();

        let u1 = make_upstream(tenant, "openai");
        let mut u2 = make_upstream(tenant, "anthropic");
        repo.create(u1.clone()).await.unwrap();
        repo.create(u2.clone()).await.unwrap();

        // Try to rename u2's alias to "openai" (already taken by u1).
        u2.alias = "openai".into();
        let result = repo.update(u2).await;
        assert!(matches!(result, Err(RepositoryError::Conflict { .. })));

        // Original alias mappings are intact.
        let fetched_u1 = repo.get_by_alias(tenant, "openai").await.unwrap();
        assert_eq!(fetched_u1.id, u1.id);

        let fetched_u2 = repo.get_by_alias(tenant, "anthropic").await.unwrap();
        assert_eq!(fetched_u2.alias, "anthropic");
    }

    #[tokio::test]
    async fn delete_removes_alias_index() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();
        let u = make_upstream(tenant, "openai");
        let id = u.id;
        repo.create(u).await.unwrap();

        repo.delete(tenant, id).await.unwrap();
        assert!(repo.get_by_id(tenant, id).await.is_err());
        assert!(repo.get_by_alias(tenant, "openai").await.is_err());
    }

    #[tokio::test]
    async fn list_with_pagination() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();

        for i in 0..5 {
            repo.create(make_upstream(tenant, &format!("svc-{i}")))
                .await
                .unwrap();
        }

        let all = repo
            .list(tenant, &ListQuery { top: 50, skip: 0 })
            .await
            .unwrap();
        assert_eq!(all.len(), 5);

        let page = repo
            .list(tenant, &ListQuery { top: 2, skip: 1 })
            .await
            .unwrap();
        assert_eq!(page.len(), 2);
    }

    #[tokio::test]
    async fn cross_tenant_delete_returns_not_found_and_preserves_upstream() {
        let repo = InMemoryUpstreamRepo::new();
        let owner = Uuid::new_v4();
        let attacker = Uuid::new_v4();

        let u = make_upstream(owner, "openai");
        let id = u.id;
        repo.create(u).await.unwrap();

        // Different tenant cannot delete it.
        let result = repo.delete(attacker, id).await;
        assert!(matches!(result, Err(RepositoryError::NotFound { .. })));

        // Upstream remains accessible to the owner.
        let fetched = repo.get_by_id(owner, id).await.unwrap();
        assert_eq!(fetched.id, id);
        assert_eq!(fetched.alias, "openai");

        // Alias index is also intact.
        let by_alias = repo.get_by_alias(owner, "openai").await.unwrap();
        assert_eq!(by_alias.id, id);
    }

    #[tokio::test]
    async fn list_pagination_is_deterministic() {
        let repo = InMemoryUpstreamRepo::new();
        let tenant = Uuid::new_v4();

        for i in 0..5 {
            repo.create(make_upstream(tenant, &format!("svc-{i}")))
                .await
                .unwrap();
        }

        let query = ListQuery { top: 3, skip: 0 };
        let first = repo.list(tenant, &query).await.unwrap();
        let second = repo.list(tenant, &query).await.unwrap();

        // Same results both times.
        assert_eq!(first.len(), 3);
        let first_ids: Vec<Uuid> = first.iter().map(|u| u.id).collect();
        let second_ids: Vec<Uuid> = second.iter().map(|u| u.id).collect();
        assert_eq!(first_ids, second_ids);

        // IDs are sorted.
        for w in first_ids.windows(2) {
            assert!(w[0] < w[1], "IDs must be in ascending order");
        }
    }

    #[tokio::test]
    async fn cross_tenant_isolation() {
        let repo = InMemoryUpstreamRepo::new();
        let t1 = Uuid::new_v4();
        let t2 = Uuid::new_v4();

        let u = make_upstream(t1, "openai");
        let id = u.id;
        repo.create(u).await.unwrap();

        // Different tenant cannot see it.
        assert!(repo.get_by_id(t2, id).await.is_err());
    }

    #[tokio::test]
    async fn list_by_alias_for_tenants_filters_by_tenant_set() {
        let repo = InMemoryUpstreamRepo::new();
        let t1 = Uuid::new_v4();
        let t2 = Uuid::new_v4();
        let t3 = Uuid::new_v4();

        repo.create(make_upstream(t1, "shared")).await.unwrap();
        repo.create(make_upstream(t2, "shared")).await.unwrap();
        repo.create(make_upstream(t3, "shared")).await.unwrap();

        let filter: std::collections::HashSet<Uuid> = [t1, t2].into();
        let results = repo
            .list_by_alias_for_tenants("shared", &filter)
            .await
            .unwrap();
        assert_eq!(results.len(), 2);
        assert!(results.iter().all(|u| u.alias == "shared"));
        assert!(results.iter().all(|u| u.tenant_id != t3));
    }

    #[tokio::test]
    async fn list_by_alias_for_tenants_returns_empty_for_unknown_alias() {
        let repo = InMemoryUpstreamRepo::new();
        let t = Uuid::new_v4();
        repo.create(make_upstream(t, "openai")).await.unwrap();

        let filter: std::collections::HashSet<Uuid> = [t].into();
        let results = repo
            .list_by_alias_for_tenants("nonexistent", &filter)
            .await
            .unwrap();
        assert!(results.is_empty());
    }

    #[tokio::test]
    async fn list_by_alias_for_tenants_returns_empty_for_empty_tenant_set() {
        let repo = InMemoryUpstreamRepo::new();
        let t = Uuid::new_v4();
        repo.create(make_upstream(t, "openai")).await.unwrap();

        let filter: std::collections::HashSet<Uuid> = std::collections::HashSet::new();
        let results = repo
            .list_by_alias_for_tenants("openai", &filter)
            .await
            .unwrap();
        assert!(results.is_empty());
    }
}
