//! `Actor` — per-DID metadata accessor for Lua scripts.
//!
//! The `Actor` global is bound to the current authenticated caller. It can
//! also be called like a function to construct a fresh actor table for any
//! DID:
//!
//! ```lua
//! Actor:has("beta")              -- the current caller
//! Actor:get("role")
//! Actor:set("tier", "pro")
//!
//! local mod = Actor("did:plc:xyz")
//! mod:set("role", "moderator")
//! local meta = mod:metadata()
//! ```
//!
//! Values are JSON-encoded so any Lua value (string, number, bool, table)
//! round-trips cleanly. When there is no authenticated caller, the global
//! `Actor` still exists but its `did` is nil — read methods return nil/false
//! and write methods raise an error.
//!
//! Records live in the `actor_metadata` table.

use mlua::{Lua, LuaSerdeExt, Result as LuaResult};
use serde_json::Value;
use std::sync::Arc;

use crate::AppState;
use crate::db::{adapt_sql, now_rfc3339};

/// Maximum length of a metadata key.
const MAX_KEY_LEN: usize = 256;
/// Maximum length of a metadata value after JSON encoding.
const MAX_VALUE_LEN: usize = 64 * 1024;

/// Validate a metadata key. Mirrors the constraints in `kv` style APIs.
fn check_key(key: &str) -> LuaResult<()> {
    if key.is_empty() {
        return Err(mlua::Error::runtime("actor metadata key must not be empty"));
    }
    if key.len() > MAX_KEY_LEN {
        return Err(mlua::Error::runtime(format!(
            "actor metadata key length ({}) exceeds maximum ({})",
            key.len(),
            MAX_KEY_LEN,
        )));
    }
    Ok(())
}

/// Pull `did` from the actor's table, raising a clear error if it's missing.
fn require_did(this: &mlua::Table) -> LuaResult<String> {
    let did: Option<String> = this.raw_get("did")?;
    did.filter(|s| !s.is_empty()).ok_or_else(|| {
        mlua::Error::runtime(
            "Actor has no DID — there is no authenticated caller in this context. \
             Use Actor(did) to operate on a specific DID instead.",
        )
    })
}

/// Register the `Actor` global. The global is bound to `caller_did` when
/// present; otherwise it is bound to nil and read methods become no-ops.
pub fn register_actor_api(
    lua: &Lua,
    state: Arc<AppState>,
    caller_did: Option<String>,
) -> LuaResult<()> {
    // -- methods table (shared via __index for every actor instance) --
    let methods = lua.create_table()?;

    // :get(key) -> any | nil
    {
        let state = state.clone();
        let f = lua.create_async_function(move |lua, (this, key): (mlua::Table, String)| {
            let state = state.clone();
            async move {
                check_key(&key)?;
                let did: Option<String> = this.raw_get("did")?;
                let did = match did {
                    Some(d) if !d.is_empty() => d,
                    _ => return Ok(mlua::Value::Nil),
                };

                let backend = state.db_backend;
                let sql = adapt_sql(
                    "SELECT value FROM actor_metadata WHERE did = ? AND key = ?",
                    backend,
                );
                let row: Option<(String,)> = sqlx::query_as(&sql)
                    .bind(&did)
                    .bind(&key)
                    .fetch_optional(&state.db)
                    .await
                    .map_err(|e| mlua::Error::runtime(format!("Actor:get failed: {e}")))?;

                match row {
                    Some((value_str,)) => {
                        let value: Value = serde_json::from_str(&value_str).map_err(|e| {
                            mlua::Error::runtime(format!(
                                "Actor:get: stored metadata is not valid JSON: {e}"
                            ))
                        })?;
                        lua.to_value(&value)
                    }
                    None => Ok(mlua::Value::Nil),
                }
            }
        })?;
        methods.set("get", f)?;
    }

    // :has(key) -> bool
    {
        let state = state.clone();
        let f = lua.create_async_function(move |_, (this, key): (mlua::Table, String)| {
            let state = state.clone();
            async move {
                check_key(&key)?;
                let did: Option<String> = this.raw_get("did")?;
                let did = match did {
                    Some(d) if !d.is_empty() => d,
                    _ => return Ok(false),
                };

                let backend = state.db_backend;
                let sql = adapt_sql(
                    "SELECT 1 FROM actor_metadata WHERE did = ? AND key = ?",
                    backend,
                );
                let row: Option<(i64,)> = sqlx::query_as(&sql)
                    .bind(&did)
                    .bind(&key)
                    .fetch_optional(&state.db)
                    .await
                    .map_err(|e| mlua::Error::runtime(format!("Actor:has failed: {e}")))?;
                Ok(row.is_some())
            }
        })?;
        methods.set("has", f)?;
    }

    // :metadata() -> table
    {
        let state = state.clone();
        let f = lua.create_async_function(move |lua, this: mlua::Table| {
            let state = state.clone();
            async move {
                let did: Option<String> = this.raw_get("did")?;
                let result = lua.create_table()?;
                let did = match did {
                    Some(d) if !d.is_empty() => d,
                    _ => return Ok(result),
                };

                let backend = state.db_backend;
                let sql = adapt_sql(
                    "SELECT key, value FROM actor_metadata WHERE did = ? ORDER BY key ASC",
                    backend,
                );
                let rows: Vec<(String, String)> = sqlx::query_as(&sql)
                    .bind(&did)
                    .fetch_all(&state.db)
                    .await
                    .map_err(|e| mlua::Error::runtime(format!("Actor:metadata failed: {e}")))?;

                for (key, value_str) in rows {
                    let value: Value = serde_json::from_str(&value_str).unwrap_or(Value::Null);
                    result.set(key, lua.to_value(&value)?)?;
                }
                Ok(result)
            }
        })?;
        methods.set("metadata", f)?;
    }

    // :set(key, value) -> nil
    {
        let state = state.clone();
        let f = lua.create_async_function(
            move |lua, (this, key, value): (mlua::Table, String, mlua::Value)| {
                let state = state.clone();
                async move {
                    let did = require_did(&this)?;
                    check_key(&key)?;

                    let value_json: Value = lua.from_value(value).map_err(|e| {
                        mlua::Error::runtime(format!("Actor:set: failed to convert value: {e}"))
                    })?;
                    let value_str = serde_json::to_string(&value_json).map_err(|e| {
                        mlua::Error::runtime(format!("Actor:set: failed to encode value: {e}"))
                    })?;
                    if value_str.len() > MAX_VALUE_LEN {
                        return Err(mlua::Error::runtime(format!(
                            "Actor:set: value size ({}) exceeds maximum ({})",
                            value_str.len(),
                            MAX_VALUE_LEN,
                        )));
                    }

                    let backend = state.db_backend;
                    let now = now_rfc3339();
                    let sql = adapt_sql(
                        "INSERT INTO actor_metadata (did, key, value, created_at, updated_at) \
                         VALUES (?, ?, ?, ?, ?) \
                         ON CONFLICT (did, key) DO UPDATE \
                            SET value = ?, updated_at = ?",
                        backend,
                    );
                    sqlx::query(&sql)
                        .bind(&did)
                        .bind(&key)
                        .bind(&value_str)
                        .bind(&now)
                        .bind(&now)
                        .bind(&value_str)
                        .bind(&now)
                        .execute(&state.db)
                        .await
                        .map_err(|e| mlua::Error::runtime(format!("Actor:set failed: {e}")))?;

                    Ok(())
                }
            },
        )?;
        methods.set("set", f)?;
    }

    // :delete(key) -> bool (true if a row was removed)
    {
        let state = state.clone();
        let f = lua.create_async_function(move |_, (this, key): (mlua::Table, String)| {
            let state = state.clone();
            async move {
                let did = require_did(&this)?;
                check_key(&key)?;
                let backend = state.db_backend;
                let sql = adapt_sql(
                    "DELETE FROM actor_metadata WHERE did = ? AND key = ?",
                    backend,
                );
                let result = sqlx::query(&sql)
                    .bind(&did)
                    .bind(&key)
                    .execute(&state.db)
                    .await
                    .map_err(|e| mlua::Error::runtime(format!("Actor:delete failed: {e}")))?;
                Ok(result.rows_affected() > 0)
            }
        })?;
        methods.set("delete", f)?;
    }

    // :clear() -> integer (number of rows removed)
    {
        let state = state.clone();
        let f = lua.create_async_function(move |_, this: mlua::Table| {
            let state = state.clone();
            async move {
                let did = require_did(&this)?;
                let backend = state.db_backend;
                let sql = adapt_sql("DELETE FROM actor_metadata WHERE did = ?", backend);
                let result = sqlx::query(&sql)
                    .bind(&did)
                    .execute(&state.db)
                    .await
                    .map_err(|e| mlua::Error::runtime(format!("Actor:clear failed: {e}")))?;
                Ok(result.rows_affected() as i64)
            }
        })?;
        methods.set("clear", f)?;
    }

    // -- shared metatable for every actor instance --
    let mt = lua.create_table()?;

    // __index: methods first, then raw fields (so `actor.did` still works).
    {
        let methods_for_index = methods.clone();
        let f = lua.create_function(move |_, (this, key): (mlua::Table, String)| {
            let m: mlua::Value = methods_for_index.raw_get(key.as_str())?;
            if !m.is_nil() {
                return Ok(m);
            }
            this.raw_get::<mlua::Value>(key.as_str())
        })?;
        mt.set("__index", f)?;
    }

    // __tostring for diagnostics.
    {
        let f = lua.create_function(|_, this: mlua::Table| {
            let did: Option<String> = this.raw_get("did")?;
            Ok(match did {
                Some(d) if !d.is_empty() => format!("Actor[{d}]"),
                _ => "Actor[<unauthenticated>]".to_string(),
            })
        })?;
        mt.set("__tostring", f)?;
    }

    // __call: Actor(did) returns a fresh actor bound to `did`. Works on
    // both the global Actor and on any actor instance (e.g. obtained
    // earlier via Actor(did)).
    {
        let mt_for_call = mt.clone();
        let f = lua.create_function(move |lua, (_self, did): (mlua::Table, String)| {
            if did.is_empty() {
                return Err(mlua::Error::runtime("Actor(did): did must be non-empty"));
            }
            let t = lua.create_table()?;
            t.raw_set("did", did)?;
            t.set_metatable(Some(mt_for_call.clone()))?;
            Ok(t)
        })?;
        mt.set("__call", f)?;
    }

    // -- the global `Actor` itself: an actor bound to caller_did (or nil). --
    let actor_global = lua.create_table()?;
    match caller_did {
        Some(d) if !d.is_empty() => actor_global.raw_set("did", d)?,
        _ => actor_global.raw_set("did", mlua::Value::Nil)?,
    }
    actor_global.set_metatable(Some(mt))?;

    lua.globals().set("Actor", actor_global)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn check_key_rejects_empty() {
        let err = check_key("").unwrap_err().to_string();
        assert!(err.contains("must not be empty"), "got: {err}");
    }

    #[test]
    fn check_key_rejects_too_long() {
        let key = "x".repeat(MAX_KEY_LEN + 1);
        let err = check_key(&key).unwrap_err().to_string();
        assert!(err.contains("exceeds maximum"), "got: {err}");
    }

    #[test]
    fn check_key_accepts_normal() {
        assert!(check_key("role").is_ok());
        assert!(check_key("limits.rate").is_ok());
    }

    #[test]
    fn require_did_rejects_missing() {
        let lua = Lua::new();
        let t = lua.create_table().unwrap();
        let err = require_did(&t).unwrap_err().to_string();
        assert!(err.contains("no DID"), "got: {err}");
    }

    #[test]
    fn require_did_rejects_empty_string() {
        let lua = Lua::new();
        let t = lua.create_table().unwrap();
        t.raw_set("did", "").unwrap();
        let err = require_did(&t).unwrap_err().to_string();
        assert!(err.contains("no DID"), "got: {err}");
    }

    #[test]
    fn require_did_returns_present() {
        let lua = Lua::new();
        let t = lua.create_table().unwrap();
        t.raw_set("did", "did:plc:abc").unwrap();
        assert_eq!(require_did(&t).unwrap(), "did:plc:abc");
    }
}