//! `KvServiceImpl` — wires the `KvRepo` underneath the
//! `picloud_shared::KvService` trait that scripts see via the Rhai
//! bridge.
//!
//! Layers added here (vs the raw repo):
//!
//! 1. Empty-collection rejection at the SDK boundary
//!    (`docs/sdk-shape.md`).
//! 2. **Script-as-gate authz**: when `cx.principal.is_some()` we run
//!    `authz::require(...)`; when it's `None` (public unauthenticated
//!    HTTP — the common case for public routes) we skip the check.
//!    Cross-app isolation isn't affected — every query is keyed by
//!    `cx.app_id`, never an argument.
//! 3. `ServiceEvent` emission after each mutation (`insert` / `update`
//!    / `delete`), via the injected `KvWriter` — which for the host is
//!    transactional, so a mutation and the trigger fan-out it produces
//!    commit together. See `crate::atomic_write`.

use std::sync::Arc;

use async_trait::async_trait;
use picloud_shared::{KvError, KvListPage, KvService, SdkCallCx, ServiceEventEmitter};
use sqlx::PgPool;

use crate::atomic_write::{BestEffortKvWriter, KvWriter, PostgresKvWriter};
use crate::authz::{self, AuthzRepo, Capability};
use crate::kv_repo::{KvRepo, KvRepoError};

/// Default per-key JSON-encoded value cap (256 KB). Override with
/// `PICLOUD_KV_MAX_VALUE_BYTES`.
pub const DEFAULT_KV_MAX_VALUE_BYTES: usize = 256 * 1024;

/// Read `PICLOUD_KV_MAX_VALUE_BYTES`; invalid values fall back to the
/// conservative default with a warning. Public so the picloud binary can
/// thread it through `KvServiceImpl::new`.
#[must_use]
pub fn kv_max_value_bytes_from_env() -> usize {
    if let Ok(v) = std::env::var("PICLOUD_KV_MAX_VALUE_BYTES") {
        match v.trim().parse::<usize>() {
            Ok(n) if n > 0 => return n,
            _ => tracing::warn!(
                value = %v,
                "ignoring invalid PICLOUD_KV_MAX_VALUE_BYTES (want a positive integer)"
            ),
        }
    }
    DEFAULT_KV_MAX_VALUE_BYTES
}

pub struct KvServiceImpl {
    /// Reads only. Mutations go through `writer`, which owns the write AND the
    /// trigger fan-out so the two can share a transaction.
    repo: Arc<dyn KvRepo>,
    authz: Arc<dyn AuthzRepo>,
    writer: Arc<dyn KvWriter>,
    max_value_bytes: usize,
}

impl KvServiceImpl {
    #[must_use]
    pub fn new(
        repo: Arc<dyn KvRepo>,
        authz: Arc<dyn AuthzRepo>,
        events: Arc<dyn ServiceEventEmitter>,
    ) -> Self {
        Self::with_max_value_bytes(repo, authz, events, DEFAULT_KV_MAX_VALUE_BYTES)
    }

    #[must_use]
    pub fn with_max_value_bytes(
        repo: Arc<dyn KvRepo>,
        authz: Arc<dyn AuthzRepo>,
        events: Arc<dyn ServiceEventEmitter>,
        max_value_bytes: usize,
    ) -> Self {
        Self {
            writer: Arc::new(BestEffortKvWriter::new(repo.clone(), events)),
            repo,
            authz,
            max_value_bytes,
        }
    }

    /// Swap the best-effort writer for the transactional one: the row write and
    /// its trigger fan-out then commit together, so an outbox failure rolls the
    /// write back instead of silently losing the event. The host always calls
    /// this; the in-memory unit tests do not (they have no Postgres).
    #[must_use]
    pub fn with_atomic_writes(mut self, pool: PgPool, max_rows: u64) -> Self {
        self.writer = Arc::new(PostgresKvWriter::new(pool, max_rows));
        self
    }

    /// Encode `value` and enforce the per-key byte cap. Runs before authz so an
    /// anonymous public script can't push an oversized payload at Postgres.
    fn check_value_size(&self, value: &serde_json::Value) -> Result<(), KvError> {
        let encoded_len = serde_json::to_vec(value)
            .map(|v| v.len())
            .map_err(|e| KvError::Backend(format!("encode value: {e}")))?;
        if encoded_len > self.max_value_bytes {
            return Err(KvError::ValueTooLarge {
                limit: self.max_value_bytes,
                actual: encoded_len,
            });
        }
        Ok(())
    }

    async fn check_read(&self, cx: &SdkCallCx) -> Result<(), KvError> {
        authz::script_gate(
            &*self.authz,
            cx,
            Capability::AppKvRead(cx.app_id),
            || KvError::Forbidden,
            KvError::Backend,
        )
        .await
    }

    async fn check_write(&self, cx: &SdkCallCx) -> Result<(), KvError> {
        authz::script_gate(
            &*self.authz,
            cx,
            Capability::AppKvWrite(cx.app_id),
            || KvError::Forbidden,
            KvError::Backend,
        )
        .await
    }
}

fn validate_collection(collection: &str) -> Result<(), KvError> {
    if collection.is_empty() {
        return Err(KvError::InvalidCollection);
    }
    Ok(())
}

impl From<KvRepoError> for KvError {
    fn from(e: KvRepoError) -> Self {
        Self::Backend(e.to_string())
    }
}

#[async_trait]
impl KvService for KvServiceImpl {
    async fn get(
        &self,
        cx: &SdkCallCx,
        collection: &str,
        key: &str,
    ) -> Result<Option<serde_json::Value>, KvError> {
        validate_collection(collection)?;
        self.check_read(cx).await?;
        Ok(self.repo.get(cx.app_id, collection, key).await?)
    }

    async fn set(
        &self,
        cx: &SdkCallCx,
        collection: &str,
        key: &str,
        value: serde_json::Value,
    ) -> Result<(), KvError> {
        validate_collection(collection)?;
        self.check_value_size(&value)?;
        self.check_write(cx).await?;
        self.writer.set(cx, collection, key, value).await?;
        Ok(())
    }

    async fn set_if(
        &self,
        cx: &SdkCallCx,
        collection: &str,
        key: &str,
        expected: Option<serde_json::Value>,
        new: serde_json::Value,
    ) -> Result<bool, KvError> {
        validate_collection(collection)?;
        self.check_value_size(&new)?;
        self.check_write(cx).await?;
        self.writer.set_if(cx, collection, key, expected, new).await
    }

    async fn delete(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result<bool, KvError> {
        validate_collection(collection)?;
        self.check_write(cx).await?;
        Ok(self.writer.delete(cx, collection, key).await?.is_some())
    }

    async fn has(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result<bool, KvError> {
        validate_collection(collection)?;
        self.check_read(cx).await?;
        Ok(self.repo.has(cx.app_id, collection, key).await?)
    }

    async fn list(
        &self,
        cx: &SdkCallCx,
        collection: &str,
        cursor: Option<&str>,
        limit: u32,
    ) -> Result<KvListPage, KvError> {
        validate_collection(collection)?;
        self.check_read(cx).await?;
        Ok(self.repo.list(cx.app_id, collection, cursor, limit).await?)
    }
}

// ----------------------------------------------------------------------------
// Tests — in-memory KvRepo so unit tests don't need Postgres.
// ----------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::authz::{AuthzError, AuthzRepo};
    use async_trait::async_trait;
    use picloud_shared::{
        AdminUserId, AppId, AppRole, ExecutionId, InstanceRole, NoopEventEmitter, Principal,
        RequestId, ScriptId, UserId,
    };
    use std::collections::{BTreeMap, HashMap};
    use tokio::sync::Mutex;

    #[derive(Default)]
    struct InMemoryKvRepo {
        data: Mutex<BTreeMap<(AppId, String, String), serde_json::Value>>,
    }

    #[async_trait]
    impl KvRepo for InMemoryKvRepo {
        async fn get(
            &self,
            app_id: AppId,
            collection: &str,
            key: &str,
        ) -> Result<Option<serde_json::Value>, KvRepoError> {
            Ok(self
                .data
                .lock()
                .await
                .get(&(app_id, collection.to_string(), key.to_string()))
                .cloned())
        }

        async fn set(
            &self,
            app_id: AppId,
            collection: &str,
            key: &str,
            value: serde_json::Value,
        ) -> Result<Option<serde_json::Value>, KvRepoError> {
            Ok(self
                .data
                .lock()
                .await
                .insert((app_id, collection.to_string(), key.to_string()), value))
        }

        async fn set_if(
            &self,
            app_id: AppId,
            collection: &str,
            key: &str,
            expected: Option<serde_json::Value>,
            new: serde_json::Value,
        ) -> Result<bool, KvRepoError> {
            // Holding the map lock makes the compare-and-set atomic in-memory.
            let mut data = self.data.lock().await;
            let k = (app_id, collection.to_string(), key.to_string());
            let matches = match (&expected, data.get(&k)) {
                (None, None) => true,                 // insert-if-absent
                (Some(exp), Some(cur)) => exp == cur, // swap-if-equal
                _ => false,
            };
            if matches {
                data.insert(k, new);
            }
            Ok(matches)
        }

        async fn delete(
            &self,
            app_id: AppId,
            collection: &str,
            key: &str,
        ) -> Result<Option<serde_json::Value>, KvRepoError> {
            Ok(self
                .data
                .lock()
                .await
                .remove(&(app_id, collection.to_string(), key.to_string())))
        }

        async fn has(
            &self,
            app_id: AppId,
            collection: &str,
            key: &str,
        ) -> Result<bool, KvRepoError> {
            Ok(self.data.lock().await.contains_key(&(
                app_id,
                collection.to_string(),
                key.to_string(),
            )))
        }

        async fn list(
            &self,
            app_id: AppId,
            collection: &str,
            cursor: Option<&str>,
            limit: u32,
        ) -> Result<KvListPage, KvRepoError> {
            let data = self.data.lock().await;
            let last_key = cursor.map(std::string::ToString::to_string);
            let mut keys: Vec<String> = data
                .iter()
                .filter(|((a, c, _), _)| *a == app_id && c == collection)
                .map(|((_, _, k), _)| k.clone())
                .filter(|k| last_key.as_ref().is_none_or(|lk| k > lk))
                .collect();
            keys.sort();
            let take = (limit as usize).max(1);
            let next_cursor = if keys.len() > take {
                keys.truncate(take);
                keys.last().cloned()
            } else {
                None
            };
            Ok(KvListPage { keys, next_cursor })
        }
    }

    /// AuthzRepo that always denies — used to confirm the service
    /// short-circuits on cx.principal.is_some() with a denial, and
    /// that it does NOT call into authz when cx.principal is None.
    #[derive(Default)]
    struct DenyingAuthzRepo;

    #[async_trait]
    impl AuthzRepo for DenyingAuthzRepo {
        async fn membership(
            &self,
            _user_id: UserId,
            _app_id: AppId,
        ) -> Result<Option<AppRole>, AuthzError> {
            Ok(None)
        }
    }

    fn anon_cx(app_id: AppId) -> SdkCallCx {
        SdkCallCx {
            app_id,
            script_id: ScriptId::new(),
            principal: None,
            execution_id: ExecutionId::new(),
            request_id: RequestId::new(),
            trigger_depth: 0,
            root_execution_id: ExecutionId::new(),
            is_dead_letter_handler: false,
            event: None,
        }
    }

    fn owner_cx(app_id: AppId) -> SdkCallCx {
        SdkCallCx {
            app_id,
            script_id: ScriptId::new(),
            principal: Some(Principal {
                user_id: AdminUserId::new(),
                instance_role: InstanceRole::Owner,
                scopes: None,
                app_binding: None,
            }),
            execution_id: ExecutionId::new(),
            request_id: RequestId::new(),
            trigger_depth: 0,
            root_execution_id: ExecutionId::new(),
            is_dead_letter_handler: false,
            event: None,
        }
    }

    fn member_no_role_cx(app_id: AppId) -> SdkCallCx {
        SdkCallCx {
            app_id,
            script_id: ScriptId::new(),
            principal: Some(Principal {
                user_id: AdminUserId::new(),
                instance_role: InstanceRole::Member,
                scopes: None,
                app_binding: None,
            }),
            execution_id: ExecutionId::new(),
            request_id: RequestId::new(),
            trigger_depth: 0,
            root_execution_id: ExecutionId::new(),
            is_dead_letter_handler: false,
            event: None,
        }
    }

    fn svc() -> KvServiceImpl {
        KvServiceImpl::new(
            Arc::new(InMemoryKvRepo::default()),
            Arc::new(DenyingAuthzRepo),
            Arc::new(NoopEventEmitter),
        )
    }

    #[tokio::test]
    async fn set_then_get_round_trips() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        kv.set(&cx, "widgets", "k1", serde_json::json!({"n": 1}))
            .await
            .unwrap();
        let v = kv.get(&cx, "widgets", "k1").await.unwrap();
        assert_eq!(v, Some(serde_json::json!({"n": 1})));
    }

    #[tokio::test]
    async fn get_missing_returns_none() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        let v = kv.get(&cx, "widgets", "nope").await.unwrap();
        assert_eq!(v, None);
    }

    #[tokio::test]
    async fn has_returns_bool() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        assert!(!kv.has(&cx, "widgets", "k1").await.unwrap());
        kv.set(&cx, "widgets", "k1", serde_json::json!(true))
            .await
            .unwrap();
        assert!(kv.has(&cx, "widgets", "k1").await.unwrap());
    }

    #[tokio::test]
    async fn delete_returns_was_present() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        assert!(!kv.delete(&cx, "widgets", "missing").await.unwrap());
        kv.set(&cx, "widgets", "k1", serde_json::json!(1))
            .await
            .unwrap();
        assert!(kv.delete(&cx, "widgets", "k1").await.unwrap());
        // Idempotent — second delete returns false.
        assert!(!kv.delete(&cx, "widgets", "k1").await.unwrap());
    }

    #[tokio::test]
    async fn empty_collection_rejected() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        let err = kv.get(&cx, "", "k1").await.unwrap_err();
        assert!(matches!(err, KvError::InvalidCollection));
    }

    #[tokio::test]
    async fn set_if_is_compare_and_swap() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        // Insert-if-absent: expected () swaps only when the key is missing.
        assert!(
            kv.set_if(&cx, "cas", "k", None, serde_json::json!(1))
                .await
                .unwrap(),
            "insert-if-absent must succeed on a missing key"
        );
        assert!(
            !kv.set_if(&cx, "cas", "k", None, serde_json::json!(2))
                .await
                .unwrap(),
            "insert-if-absent must fail once the key exists"
        );
        assert_eq!(
            kv.get(&cx, "cas", "k").await.unwrap(),
            Some(serde_json::json!(1))
        );

        // Swap-if-equal: wrong expected → no-op; correct expected → swaps.
        assert!(
            !kv.set_if(
                &cx,
                "cas",
                "k",
                Some(serde_json::json!(99)),
                serde_json::json!(3)
            )
            .await
            .unwrap(),
            "a mismatched expected must not swap"
        );
        assert_eq!(
            kv.get(&cx, "cas", "k").await.unwrap(),
            Some(serde_json::json!(1))
        );
        assert!(
            kv.set_if(
                &cx,
                "cas",
                "k",
                Some(serde_json::json!(1)),
                serde_json::json!(3)
            )
            .await
            .unwrap(),
            "a matching expected must swap"
        );
        assert_eq!(
            kv.get(&cx, "cas", "k").await.unwrap(),
            Some(serde_json::json!(3))
        );
    }

    /// Load-bearing: a script with `cx.app_id = A` must NOT see
    /// entries inserted under `cx.app_id = B`. This is the cross-app
    /// isolation boundary; getting this wrong is a security
    /// vulnerability.
    #[tokio::test]
    async fn cross_app_isolation_via_cx_app_id() {
        let kv = svc();
        let app_a = AppId::new();
        let app_b = AppId::new();
        let cx_a = anon_cx(app_a);
        let cx_b = anon_cx(app_b);

        kv.set(&cx_a, "shared", "k", serde_json::json!("from-a"))
            .await
            .unwrap();
        kv.set(&cx_b, "shared", "k", serde_json::json!("from-b"))
            .await
            .unwrap();

        assert_eq!(
            kv.get(&cx_a, "shared", "k").await.unwrap(),
            Some(serde_json::json!("from-a"))
        );
        assert_eq!(
            kv.get(&cx_b, "shared", "k").await.unwrap(),
            Some(serde_json::json!("from-b"))
        );
    }

    /// Script-as-gate: an `anon_cx` (principal = None) skips the
    /// capability check entirely. Even with a denying authz repo,
    /// the write succeeds.
    #[tokio::test]
    async fn anonymous_cx_skips_authz() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        kv.set(&cx, "widgets", "k", serde_json::json!(1))
            .await
            .unwrap();
        // No panic, no Forbidden.
    }

    /// Authenticated principal with no role on the app: the
    /// `DenyingAuthzRepo` returns no membership, so the capability
    /// check denies. Set must surface KvError::Forbidden.
    #[tokio::test]
    async fn authed_cx_with_no_role_is_forbidden() {
        let kv = svc();
        let cx = member_no_role_cx(AppId::new());
        let err = kv
            .set(&cx, "widgets", "k", serde_json::json!(1))
            .await
            .unwrap_err();
        assert!(matches!(err, KvError::Forbidden));
    }

    /// Owner principal: instance-role grants kick in inside `authz::can`
    /// (Owner -> implicit AppAdmin which covers KvWrite).
    #[tokio::test]
    async fn owner_principal_can_write() {
        let kv = svc();
        let cx = owner_cx(AppId::new());
        kv.set(&cx, "widgets", "k", serde_json::json!(1))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn list_cursor_pagination() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        for i in 0..5 {
            kv.set(
                &cx,
                "widgets",
                &format!("k{i:02}"),
                serde_json::json!({"i": i}),
            )
            .await
            .unwrap();
        }
        // page 1 — 2 keys
        let p1 = kv.list(&cx, "widgets", None, 2).await.unwrap();
        assert_eq!(p1.keys, vec!["k00".to_string(), "k01".to_string()]);
        assert!(p1.next_cursor.is_some());
        // page 2 — 2 keys
        let p2 = kv
            .list(&cx, "widgets", p1.next_cursor.as_deref(), 2)
            .await
            .unwrap();
        assert_eq!(p2.keys, vec!["k02".to_string(), "k03".to_string()]);
        // final page — 1 key, no cursor
        let p3 = kv
            .list(&cx, "widgets", p2.next_cursor.as_deref(), 2)
            .await
            .unwrap();
        assert_eq!(p3.keys, vec!["k04".to_string()]);
        assert!(p3.next_cursor.is_none());
    }

    /// Pinning the v1.1.0 contract: services hold the emitter as a
    /// dyn Arc and call `emit().await` unconditionally. This test
    /// proves the call site doesn't blow up against the noop impl —
    /// the outbox emitter (v1.1.1) drops in transparently.
    #[tokio::test]
    async fn noop_emitter_does_not_block_mutations() {
        let kv = svc();
        let cx = anon_cx(AppId::new());
        kv.set(&cx, "widgets", "k", serde_json::json!(1))
            .await
            .unwrap();
        kv.delete(&cx, "widgets", "k").await.unwrap();
        // Reaching here means emit() returned Ok and didn't panic.
        // Suppress unused-import warning when run alone:
        let _ = HashMap::<String, String>::new();
    }
}