use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use chrono::prelude::FixedOffset;
use chrono::{DateTime, Duration, Utc};
use rand::RngCore;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashSet};

use athena_auth_core::adapters::DatabaseAdapter;
use athena_auth_core::entity::{AuthAccount, AuthApiKey, AuthSession, AuthUser};
use athena_auth_core::{
    AuthContext, AuthEmailRenderContext, AuthError, AuthResult, DEFAULT_EMAIL_TRANSPORT_ID,
    ListUsersParams, PASSWORD_HASH_KEY, build_auth_email_message,
    default_event_type_for_template_key, merge_email_metadata,
};
use athena_auth_core::{
    CreateAccount, CreateApiKey, CreateAthenaClient, CreateEmailSendFailure, CreateEmailTemplate,
    CreateOAuthClient, CreateSentEmail, ListAdminAuditLogsParams, ListEmailEventTypesParams,
    ListEmailSendFailuresParams, ListEmailTemplatesParams, ListSentEmailsParams,
    ReplaceGrantAssignmentsForSource, SessionCreateOptions, SessionOps, UpdateEmailSendFailure,
    UpdateEmailTemplate, UpdateSentEmail, UpdateUser, grant_by_key,
    permissions_map_from_assignments, render_stored_auth_email_template, resolve_grant_assignments,
};
use validator::ValidateEmail;

use crate::plugins::StatusResponse;
use crate::plugins::api_key_shared::{
    build_stored_api_key_metadata, normalize_api_key_permissions_input, stored_api_key_grant_keys,
    stored_api_key_right_modules, stored_api_key_user_metadata,
};
use crate::plugins::grants::{
    grant_catalog_entries, grant_keys_from_assignments, merged_native_right_catalog,
    native_right_assignment_views, native_right_catalog_entries, native_right_keys_for_grant_keys,
    native_right_modules_for_grant_keys, native_rights_for_grant_keys, normalize_grant_inputs,
    resolved_grant_assignment_views, resolved_native_right_assignment_views,
    right_module_inputs_from_native_modules, right_modules_for_grant_keys,
    unknown_native_right_keys_for_grant_keys,
};
use crate::plugins::helpers::{
    resolve_user_name_from_email, session_with_server_version, sessions_with_server_version,
};

use super::AdminConfig;
use super::types::*;

const MAX_ADMIN_JSON_BYTES: usize = 32 * 1024;
const MAX_ADMIN_JSON_DEPTH: usize = 8;
const MAX_TEMPLATE_VARIABLES: usize = 64;
const MAX_TEMPLATE_VARIABLE_LENGTH: usize = 128;
const MAX_TEMPLATE_BINDINGS: usize = 64;
const MAX_TEMPLATE_BINDING_NAME_LENGTH: usize = 128;
const MAX_TEMPLATE_BINDING_SOURCE_LENGTH: usize = 256;
const MAX_TEMPLATE_ATTACHMENTS: usize = 10;
const MAX_TEMPLATE_ATTACHMENT_URL_LENGTH: usize = 2048;
const MAX_TEMPLATE_ATTACHMENT_FILENAME_LENGTH: usize = 255;
const ADMIN_EMAIL_TEMPLATE_SEND_FLOW: &str = "admin.email_template.send";

pub(crate) struct SendEmailTemplateCoreResult {
    pub status: u16,
    pub response: SendEmailTemplateResponse,
}

// ---------------------------------------------------------------------------
// Core functions -- framework-agnostic business logic
// ---------------------------------------------------------------------------

fn parse_optional_rfc3339_utc(
    value: Option<&str>,
    field_name: &str,
) -> AuthResult<Option<DateTime<Utc>>> {
    let Some(raw_value) = value.map(str::trim).filter(|value| !value.is_empty()) else {
        return Ok(None);
    };

    let parsed: DateTime<FixedOffset> =
        DateTime::parse_from_rfc3339(raw_value).map_err(|error| {
            AuthError::bad_request(format!(
                "Invalid {field_name}: expected RFC3339 timestamp ({error})"
            ))
        })?;

    Ok(Some(parsed.with_timezone(&Utc)))
}

// TODO: This should probably have a bit more of a global utils
fn normalize_non_empty_string(value: &str, field_name: &str) -> AuthResult<String> {
    let normalized: &str = value.trim();
    if normalized.is_empty() {
        return Err(AuthError::bad_request(format!("{field_name} is required")));
    }
    Ok(normalized.to_string())
}

// TODO: This should probably go to a more global utils
fn normalize_string_list(values: &[String], field_name: &str) -> AuthResult<Vec<String>> {
    let mut out: Vec<String> = Vec::new();
    for value in values {
        let normalized = value.trim();
        if normalized.is_empty() {
            continue;
        }
        if !out.iter().any(|existing| existing == normalized) {
            out.push(normalized.to_string());
        }
    }

    if out.is_empty() {
        return Err(AuthError::bad_request(format!(
            "{field_name} must contain at least one value"
        )));
    }

    Ok(out)
}

// TODO: This should probably have a bit more of a global utils
fn normalize_required_trimmed(value: &str, field_name: &str) -> AuthResult<String> {
    let trimmed: &str = value.trim();
    if trimmed.is_empty() {
        return Err(AuthError::bad_request(format!("{field_name} is required")));
    }
    Ok(trimmed.to_string())
}

// TODO: This should probably have a bit more of a global utils
fn normalize_optional_trimmed(value: Option<&str>) -> Option<String> {
    value
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(std::string::ToString::to_string)
}

fn normalize_optional_non_empty(
    value: Option<&str>,
    field_name: &str,
) -> AuthResult<Option<String>> {
    match value {
        Some(raw) => {
            let trimmed: &str = raw.trim();
            if trimmed.is_empty() {
                Err(AuthError::bad_request(format!(
                    "{field_name} cannot be empty when provided"
                )))
            } else {
                Ok(Some(trimmed.to_string()))
            }
        }
        None => Ok(None),
    }
}

fn normalize_optional_event_type(
    value: Option<&str>,
    field_name: &str,
) -> AuthResult<Option<String>> {
    let normalized: Option<String> = normalize_optional_non_empty(value, field_name)?;
    let Some(event_type) = normalized else {
        return Ok(None);
    };

    if event_type.len() > 128 {
        return Err(AuthError::bad_request(format!(
            "{field_name} must be at most 128 characters"
        )));
    }

    if !event_type.contains('.') {
        return Err(AuthError::bad_request(format!(
            "{field_name} must contain at least one '.' segment separator"
        )));
    }

    let valid: bool = event_type
        .chars()
        .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '.' || ch == '-');
    if !valid {
        return Err(AuthError::bad_request(format!(
            "{field_name} may only contain lowercase letters, digits, '.', and '-'"
        )));
    }

    Ok(Some(event_type))
}

fn normalize_required_email(value: &str, field_name: &str) -> AuthResult<String> {
    let normalized = normalize_required_trimmed(value, field_name)?;
    if !normalized.validate_email() {
        return Err(AuthError::bad_request(format!("Invalid {field_name}")));
    }
    Ok(normalized)
}

fn normalize_optional_email(value: Option<&str>, field_name: &str) -> AuthResult<Option<String>> {
    match value {
        Some(raw) => {
            let trimmed: &str = raw.trim();
            if trimmed.is_empty() {
                return Err(AuthError::bad_request(format!(
                    "{field_name} cannot be empty when provided"
                )));
            }
            if !trimmed.validate_email() {
                return Err(AuthError::bad_request(format!("Invalid {field_name}")));
            }
            Ok(Some(trimmed.to_string()))
        }
        None => Ok(None),
    }
}

fn normalize_nullable_trimmed_patch(value: Option<Option<String>>) -> Option<Option<String>> {
    value.map(|inner| {
        inner.and_then(|text| {
            let trimmed = text.trim();
            if trimmed.is_empty() {
                None
            } else {
                Some(trimmed.to_string())
            }
        })
    })
}

fn json_depth(value: &serde_json::Value) -> usize {
    match value {
        serde_json::Value::Array(items) => 1 + items.iter().map(json_depth).max().unwrap_or(0),
        serde_json::Value::Object(map) => 1 + map.values().map(json_depth).max().unwrap_or(0),
        _ => 1,
    }
}

fn validate_json_value_bounds(value: &serde_json::Value, field_name: &str) -> AuthResult<()> {
    let serialized = serde_json::to_vec(value).map_err(AuthError::from)?;
    if serialized.len() > MAX_ADMIN_JSON_BYTES {
        return Err(AuthError::bad_request(format!(
            "{field_name} exceeds max size of {} bytes",
            MAX_ADMIN_JSON_BYTES
        )));
    }

    let depth = json_depth(value);
    if depth > MAX_ADMIN_JSON_DEPTH {
        return Err(AuthError::bad_request(format!(
            "{field_name} exceeds max nesting depth of {}",
            MAX_ADMIN_JSON_DEPTH
        )));
    }

    Ok(())
}

fn normalize_metadata_object(
    value: Option<&serde_json::Value>,
    field_name: &str,
) -> AuthResult<Option<serde_json::Value>> {
    match value {
        Some(metadata) => {
            if !metadata.is_object() {
                return Err(AuthError::bad_request(format!(
                    "{field_name} must be a JSON object"
                )));
            }
            validate_json_value_bounds(metadata, field_name)?;
            Ok(Some(metadata.clone()))
        }
        None => Ok(None),
    }
}

fn json_value_to_render_string(value: &serde_json::Value, field_name: &str) -> AuthResult<String> {
    match value {
        serde_json::Value::Null => Err(AuthError::bad_request(format!(
            "{field_name} values must not be null"
        ))),
        serde_json::Value::String(text) => Ok(text.clone()),
        serde_json::Value::Bool(value) => Ok(value.to_string()),
        serde_json::Value::Number(value) => Ok(value.to_string()),
        serde_json::Value::Array(_) | serde_json::Value::Object(_) => {
            serde_json::to_string(value).map_err(AuthError::from)
        }
    }
}

fn normalize_render_variables(
    value: Option<&serde_json::Value>,
) -> AuthResult<BTreeMap<String, String>> {
    let Some(raw_value) = value else {
        return Ok(BTreeMap::new());
    };

    validate_json_value_bounds(raw_value, "render_variables")?;

    let object = raw_value
        .as_object()
        .ok_or_else(|| AuthError::bad_request("render_variables must be a JSON object"))?;

    if object.len() > MAX_TEMPLATE_VARIABLES {
        return Err(AuthError::bad_request(format!(
            "render_variables cannot contain more than {} entries",
            MAX_TEMPLATE_VARIABLES
        )));
    }

    let mut normalized = BTreeMap::new();
    for (raw_key, raw_value) in object {
        let key = raw_key.trim();
        if key.is_empty() {
            return Err(AuthError::bad_request(
                "render_variables keys must not be blank",
            ));
        }
        if key.len() > MAX_TEMPLATE_VARIABLE_LENGTH {
            return Err(AuthError::bad_request(format!(
                "each render_variables key must be at most {} characters",
                MAX_TEMPLATE_VARIABLE_LENGTH
            )));
        }
        let valid_key = key
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '.' || ch == '-');
        if !valid_key {
            return Err(AuthError::bad_request(
                "each render_variables key may only contain letters, digits, '_', '.', and '-'",
            ));
        }

        let field_name = format!("render_variables.{key}");
        normalized.insert(
            key.to_string(),
            json_value_to_render_string(raw_value, &field_name)?,
        );
    }

    Ok(normalized)
}

fn normalize_template_variables(
    value: Option<&serde_json::Value>,
) -> AuthResult<Option<serde_json::Value>> {
    let Some(variables_value) = value else {
        return Ok(None);
    };

    let variables = variables_value
        .as_array()
        .ok_or_else(|| AuthError::bad_request("variables must be an array of non-empty strings"))?;

    if variables.len() > MAX_TEMPLATE_VARIABLES {
        return Err(AuthError::bad_request(format!(
            "variables cannot contain more than {} entries",
            MAX_TEMPLATE_VARIABLES
        )));
    }

    let mut seen = HashSet::new();
    let mut normalized: Vec<serde_json::Value> = Vec::with_capacity(variables.len());

    for variable in variables {
        let raw = variable.as_str().ok_or_else(|| {
            AuthError::bad_request("variables must be an array of non-empty strings")
        })?;
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            return Err(AuthError::bad_request(
                "variables must not contain blank entries",
            ));
        }
        if trimmed.len() > MAX_TEMPLATE_VARIABLE_LENGTH {
            return Err(AuthError::bad_request(format!(
                "each variable entry must be at most {} characters",
                MAX_TEMPLATE_VARIABLE_LENGTH
            )));
        }
        if !seen.insert(trimmed.to_string()) {
            return Err(AuthError::bad_request(
                "variables must not contain duplicate entries",
            ));
        }
        normalized.push(serde_json::Value::String(trimmed.to_string()));
    }

    let normalized_value = serde_json::Value::Array(normalized);
    validate_json_value_bounds(&normalized_value, "variables")?;
    Ok(Some(normalized_value))
}

fn normalize_template_variable_bindings(
    value: Option<&serde_json::Value>,
) -> AuthResult<Option<serde_json::Value>> {
    let Some(bindings_value) = value else {
        return Ok(None);
    };

    let bindings = bindings_value.as_array().ok_or_else(|| {
        AuthError::bad_request(
            "variableBindings must be an array of { name, source, required? } objects",
        )
    })?;

    if bindings.len() > MAX_TEMPLATE_BINDINGS {
        return Err(AuthError::bad_request(format!(
            "variableBindings cannot contain more than {} entries",
            MAX_TEMPLATE_BINDINGS
        )));
    }

    let mut seen = HashSet::new();
    let mut normalized = Vec::with_capacity(bindings.len());

    for binding in bindings {
        let object = binding.as_object().ok_or_else(|| {
            AuthError::bad_request(
                "variableBindings must be an array of { name, source, required? } objects",
            )
        })?;

        let name = object
            .get("name")
            .and_then(serde_json::Value::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| {
                AuthError::bad_request("each variableBindings entry must include a non-empty name")
            })?;
        if name.len() > MAX_TEMPLATE_BINDING_NAME_LENGTH {
            return Err(AuthError::bad_request(format!(
                "each variableBindings name must be at most {} characters",
                MAX_TEMPLATE_BINDING_NAME_LENGTH
            )));
        }
        let valid_name = name
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '.' || ch == '-');
        if !valid_name {
            return Err(AuthError::bad_request(
                "each variableBindings name may only contain letters, digits, '_', '.', and '-'",
            ));
        }

        let source = object
            .get("source")
            .and_then(serde_json::Value::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| {
                AuthError::bad_request(
                    "each variableBindings entry must include a non-empty source",
                )
            })?;
        if source.len() > MAX_TEMPLATE_BINDING_SOURCE_LENGTH {
            return Err(AuthError::bad_request(format!(
                "each variableBindings source must be at most {} characters",
                MAX_TEMPLATE_BINDING_SOURCE_LENGTH
            )));
        }
        let segments: Vec<&str> = source.split('.').map(str::trim).collect();
        if segments.len() != 3 || segments.iter().any(|segment| segment.is_empty()) {
            return Err(AuthError::bad_request(
                "each variableBindings source must use schema.table.column format",
            ));
        }
        for segment in &segments {
            let valid_segment = segment
                .chars()
                .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == '-');
            if !valid_segment {
                return Err(AuthError::bad_request(
                    "each variableBindings source segment may only contain letters, digits, '_' and '-'",
                ));
            }
        }

        let required = match object.get("required") {
            Some(serde_json::Value::Bool(value)) => *value,
            Some(_) => {
                return Err(AuthError::bad_request(
                    "each variableBindings required field must be a boolean when provided",
                ));
            }
            None => false,
        };

        if !seen.insert(name.to_string()) {
            return Err(AuthError::bad_request(
                "variableBindings must not contain duplicate names",
            ));
        }

        normalized.push(serde_json::json!({
            "name": name,
            "source": source,
            "required": required
        }));
    }

    let normalized_value = serde_json::Value::Array(normalized);
    validate_json_value_bounds(&normalized_value, "variableBindings")?;
    Ok(Some(normalized_value))
}

fn normalize_template_attachments(
    value: Option<&serde_json::Value>,
) -> AuthResult<Option<serde_json::Value>> {
    let Some(attachments_value) = value else {
        return Ok(None);
    };

    let raw_items: Vec<&serde_json::Value> = match attachments_value {
        serde_json::Value::String(_) | serde_json::Value::Object(_) => vec![attachments_value],
        serde_json::Value::Array(items) => items.iter().collect(),
        _ => {
            return Err(AuthError::bad_request(
                "attachments must be a URL string, an object, or an array of those values",
            ));
        }
    };

    if raw_items.len() > MAX_TEMPLATE_ATTACHMENTS {
        return Err(AuthError::bad_request(format!(
            "attachments cannot contain more than {} entries",
            MAX_TEMPLATE_ATTACHMENTS
        )));
    }

    let mut seen = HashSet::new();
    let mut normalized = Vec::with_capacity(raw_items.len());

    for item in raw_items {
        let (raw_file_url, raw_filename) = match item {
            serde_json::Value::String(file_url) => (file_url.as_str(), None),
            serde_json::Value::Object(object) => {
                let file_url = object
                    .get("file_url")
                    .or_else(|| object.get("fileUrl"))
                    .and_then(serde_json::Value::as_str)
                    .ok_or_else(|| {
                        AuthError::bad_request(
                            "each attachments object must include a non-empty file_url",
                        )
                    })?;
                let filename = object.get("filename").and_then(serde_json::Value::as_str);
                (file_url, filename)
            }
            _ => {
                return Err(AuthError::bad_request(
                    "attachments must be a URL string, an object, or an array of those values",
                ));
            }
        };

        let file_url = raw_file_url.trim();
        if file_url.is_empty() {
            return Err(AuthError::bad_request(
                "each attachments entry must include a non-empty file_url",
            ));
        }
        if file_url.len() > MAX_TEMPLATE_ATTACHMENT_URL_LENGTH {
            return Err(AuthError::bad_request(format!(
                "each attachments file_url must be at most {} characters",
                MAX_TEMPLATE_ATTACHMENT_URL_LENGTH
            )));
        }
        let parsed_url = reqwest::Url::parse(file_url).map_err(|error| {
            AuthError::bad_request(format!(
                "each attachments file_url must be a valid URL ({error})"
            ))
        })?;
        match parsed_url.scheme() {
            "http" | "https" => {}
            _ => {
                return Err(AuthError::bad_request(
                    "each attachments file_url must use http or https",
                ));
            }
        }

        let filename = match raw_filename {
            Some(value) => {
                let trimmed = value.trim();
                if trimmed.is_empty() {
                    return Err(AuthError::bad_request(
                        "each attachments filename must not be blank when provided",
                    ));
                }
                if trimmed.len() > MAX_TEMPLATE_ATTACHMENT_FILENAME_LENGTH {
                    return Err(AuthError::bad_request(format!(
                        "each attachments filename must be at most {} characters",
                        MAX_TEMPLATE_ATTACHMENT_FILENAME_LENGTH
                    )));
                }
                Some(trimmed.to_string())
            }
            None => None,
        };

        let dedupe_key = format!(
            "{}::{}",
            parsed_url.as_str(),
            filename.as_deref().unwrap_or_default()
        );
        if !seen.insert(dedupe_key) {
            return Err(AuthError::bad_request(
                "attachments must not contain duplicate file_url and filename pairs",
            ));
        }

        normalized.push(serde_json::json!({
            "file_url": parsed_url.to_string(),
            "filename": filename,
        }));
    }

    let normalized_value = serde_json::Value::Array(normalized);
    validate_json_value_bounds(&normalized_value, "attachments")?;
    Ok(Some(normalized_value))
}

fn grant_catalog_modules(
    grant_catalog: &[athena_auth_core::GrantDescriptor],
    native_right_catalog: &[athena_rights::AthenaRightDescriptor],
    native_right_domain_contracts: &std::collections::BTreeMap<String, serde_json::Value>,
) -> Vec<GrantCatalogModuleResponse> {
    let catalog_entries = grant_catalog_entries(grant_catalog, native_right_catalog);
    let native_right_entries = native_right_catalog_entries(grant_catalog, native_right_catalog);
    let mut modules = BTreeMap::<String, GrantCatalogModuleResponse>::new();

    for native_right in native_right_catalog {
        modules
            .entry(native_right.source.clone())
            .or_insert_with(|| GrantCatalogModuleResponse {
                key: native_right.source.clone(),
                grant_catalog: Vec::new(),
                rights_catalog: Vec::new(),
                native_right_catalog: Vec::new(),
                native_rights: Vec::new(),
                integration_contract: native_right_domain_contracts
                    .get(&native_right.source)
                    .cloned(),
            })
            .native_rights
            .push(native_right.clone());
    }

    for entry in native_right_entries {
        let source_key = entry.native_right.source.clone();
        modules
            .entry(source_key.clone())
            .or_insert_with(|| GrantCatalogModuleResponse {
                key: source_key,
                grant_catalog: Vec::new(),
                rights_catalog: Vec::new(),
                native_right_catalog: Vec::new(),
                native_rights: Vec::new(),
                integration_contract: native_right_domain_contracts
                    .get(&entry.native_right.source)
                    .cloned(),
            })
            .native_right_catalog
            .push(entry);
    }

    let mut custom_grants = Vec::new();
    for entry in catalog_entries {
        match entry
            .native_right
            .as_ref()
            .map(|right| right.source.as_str())
        {
            Some(source_key) => {
                if let Some(module) = modules.get_mut(source_key) {
                    module.grant_catalog.push(entry.clone());
                } else {
                    custom_grants.push(entry);
                }
            }
            None => custom_grants.push(entry),
        }
    }

    let mut modules: Vec<GrantCatalogModuleResponse> = modules
        .into_values()
        .map(|mut module| {
            module.rights_catalog = module.grant_catalog.clone();
            module
        })
        .collect();

    if !custom_grants.is_empty() {
        modules.push(GrantCatalogModuleResponse {
            key: "custom".to_string(),
            rights_catalog: custom_grants.clone(),
            grant_catalog: custom_grants,
            native_right_catalog: Vec::new(),
            native_rights: Vec::new(),
            integration_contract: None,
        });
    }

    modules
}

fn native_right_domain_integration_contracts(
    modules: &[GrantCatalogModuleResponse],
) -> BTreeMap<String, serde_json::Value> {
    modules
        .iter()
        .filter_map(|module| {
            module
                .integration_contract
                .as_ref()
                .map(|contract| (module.key.clone(), contract.clone()))
        })
        .collect()
}

fn normalize_attachment_failure_mode(value: Option<&str>) -> AuthResult<Option<String>> {
    let Some(raw_value) = value else {
        return Ok(None);
    };
    let normalized = raw_value.trim().to_ascii_lowercase();
    if normalized.is_empty() {
        return Err(AuthError::bad_request(
            "attachment_failure_mode cannot be empty when provided",
        ));
    }
    match normalized.as_str() {
        "fail" | "skip" => Ok(Some(normalized)),
        _ => Err(AuthError::bad_request(
            "attachment_failure_mode must be either 'fail' or 'skip'",
        )),
    }
}

fn parse_attachment_specs(
    value: Option<&serde_json::Value>,
) -> AuthResult<Vec<athena_auth_core::TemplateAttachmentSpec>> {
    let Some(serde_json::Value::Array(items)) = value else {
        return Ok(Vec::new());
    };

    items
        .iter()
        .map(|item| {
            let object = item.as_object().ok_or_else(|| {
                AuthError::bad_request("attachments entries must be objects after normalization")
            })?;
            let file_url = object
                .get("file_url")
                .and_then(serde_json::Value::as_str)
                .ok_or_else(|| {
                    AuthError::bad_request(
                        "attachments entries must include file_url after normalization",
                    )
                })?;
            let filename = object
                .get("filename")
                .and_then(serde_json::Value::as_str)
                .map(std::string::ToString::to_string);

            Ok(athena_auth_core::TemplateAttachmentSpec {
                file_url: file_url.to_string(),
                filename,
            })
        })
        .collect()
}

fn merge_attachment_specs(
    template_attachments: Vec<athena_auth_core::TemplateAttachmentSpec>,
    request_attachments: Vec<athena_auth_core::TemplateAttachmentSpec>,
) -> Vec<athena_auth_core::TemplateAttachmentSpec> {
    let mut merged = Vec::with_capacity(template_attachments.len() + request_attachments.len());
    let mut seen = HashSet::new();

    for attachment in template_attachments.into_iter().chain(request_attachments) {
        let dedupe_key = format!(
            "{}::{}",
            attachment.file_url,
            attachment.filename.as_deref().unwrap_or_default()
        );
        if seen.insert(dedupe_key) {
            merged.push(attachment);
        }
    }

    merged
}

async fn ensure_email_event_type_exists<DB: DatabaseAdapter>(
    ctx: &AuthContext<DB>,
    event_type: &str,
) -> AuthResult<()> {
    let exists = ctx
        .database
        .get_email_event_type(event_type)
        .await?
        .is_some();
    if exists {
        Ok(())
    } else {
        Err(AuthError::bad_request(format!(
            "Unknown eventType '{event_type}'"
        )))
    }
}

fn resolved_template_event_type(template: &athena_auth_core::EmailTemplate) -> Option<String> {
    template
        .event_type
        .clone()
        .or_else(|| default_event_type_for_template_key(&template.template_key).map(str::to_string))
}

fn admin_email_template_send_metadata(
    body: &SendEmailTemplateRequest,
    template: &athena_auth_core::EmailTemplate,
) -> serde_json::Value {
    let mut metadata = serde_json::Map::new();
    metadata.insert(
        "template_id".to_string(),
        serde_json::Value::String(template.id.clone()),
    );
    metadata.insert(
        "template_key".to_string(),
        serde_json::Value::String(template.template_key.clone()),
    );
    if let Some(event_type) = resolved_template_event_type(template) {
        metadata.insert(
            "event_type".to_string(),
            serde_json::Value::String(event_type),
        );
    }
    metadata.insert(
        "render_anchors".to_string(),
        serde_json::json!({
            "user_id": normalize_optional_trimmed(body.user_id.as_deref()),
            "organization_id": normalize_optional_trimmed(body.organization_id.as_deref()),
            "session_token_present": body
                .session_token
                .as_deref()
                .map(str::trim)
                .is_some_and(|value| !value.is_empty()),
        }),
    );
    if let Some(render_variables) = body.render_variables.clone() {
        metadata.insert("render_variables".to_string(), render_variables);
    }
    if let Some(attachments) = body.attachments.clone() {
        metadata.insert("attachments".to_string(), attachments);
    }
    if let Some(attachment_failure_mode) =
        normalize_optional_trimmed(body.attachment_failure_mode.as_deref())
    {
        metadata.insert(
            "attachment_failure_mode".to_string(),
            serde_json::Value::String(attachment_failure_mode),
        );
    }

    serde_json::Value::Object(metadata)
}

async fn build_admin_email_template_send_failure<DB: DatabaseAdapter>(
    ctx: &AuthContext<DB>,
    body: &SendEmailTemplateRequest,
    template: &athena_auth_core::EmailTemplate,
    provider: Option<&str>,
    status: u16,
    error_message: impl Into<String>,
    delivery_metadata: Option<serde_json::Value>,
) -> SendEmailTemplateCoreResult {
    let error_message = error_message.into();
    let metadata = merge_email_metadata(
        merge_email_metadata(
            body.metadata.clone(),
            Some(admin_email_template_send_metadata(body, template)),
        ),
        delivery_metadata.clone(),
    );

    let email_send_failure_id = match ctx
        .database
        .create_email_send_failure_record(CreateEmailSendFailure {
            user_id: normalize_optional_trimmed(body.user_id.as_deref()),
            recipient_email: body.recipient_email.trim().to_string(),
            flow: ADMIN_EMAIL_TEMPLATE_SEND_FLOW.to_string(),
            provider: provider.map(str::to_string),
            error_message: error_message.clone(),
            metadata: metadata.clone(),
        })
        .await
    {
        Ok(row) => Some(row.id),
        Err(log_err) => {
            tracing::error!(
                template_id = %template.id,
                recipient_email = %body.recipient_email,
                flow = ADMIN_EMAIL_TEMPLATE_SEND_FLOW,
                error = %log_err,
                "failed to persist admin email template send failure"
            );
            None
        }
    };

    SendEmailTemplateCoreResult {
        status,
        response: SendEmailTemplateResponse {
            success: false,
            flow: ADMIN_EMAIL_TEMPLATE_SEND_FLOW.to_string(),
            template_id: template.id.clone(),
            template_key: template.template_key.clone(),
            event_type: resolved_template_event_type(template),
            recipient_email: body.recipient_email.trim().to_string(),
            subject: None,
            attachment_count: 0,
            delivery_metadata: metadata,
            email_send_failure_id,
            error: Some(error_message),
        },
    }
}

pub(crate) async fn set_role_core<DB: DatabaseAdapter>(
    body: &SetRoleRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<UserResponse<DB::User>> {
    let _target = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    let update = UpdateUser {
        role: Some(body.role.clone()),
        ..Default::default()
    };

    let updated_user = ctx.database.update_user(&body.user_id, update).await?;
    if let Err(cache_error) = ctx.cache_session_user(updated_user.id(), &updated_user) {
        tracing::warn!(
            user_id = %updated_user.id(),
            error = %cache_error,
            "failed to refresh session user cache after set-role"
        );
    }
    Ok(UserResponse { user: updated_user })
}

pub(crate) async fn create_user_core<DB: DatabaseAdapter>(
    body: &CreateUserRequest,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<UserResponse<DB::User>> {
    if ctx.database.get_user_by_email(&body.email).await?.is_some() {
        return Err(AuthError::conflict("A user with this email already exists"));
    }

    if body.password.len() < ctx.config.password.min_length {
        return Err(AuthError::bad_request(format!(
            "Password must be at least {} characters long",
            ctx.config.password.min_length
        )));
    }

    let password_hash: String = athena_auth_core::hash_password_with_config(
        None,
        Some(&ctx.config.password.argon2_config),
        &body.password,
    )
    .await?;

    // This is seeded from the configuration of the AdminPlugin()
    let role: String = body
        .role
        .clone()
        .unwrap_or_else(|| config.default_user_role.clone());

    let metadata_value: serde_json::Value = body.data.clone().unwrap_or(serde_json::json!({}));
    let metadata: serde_json::Value = if let serde_json::Value::Object(mut obj) = metadata_value {
        obj.insert(
            PASSWORD_HASH_KEY.to_string(),
            serde_json::json!(password_hash),
        );
        serde_json::Value::Object(obj)
    } else {
        let mut obj = serde_json::Map::new();
        obj.insert(
            PASSWORD_HASH_KEY.to_string(),
            serde_json::json!(password_hash),
        );
        serde_json::Value::Object(obj)
    };

    let resolved_name = resolve_user_name_from_email(&body.email, body.name.as_deref())?;

    let create_user = athena_auth_core::CreateUser::new()
        .with_email(&body.email)
        .with_name(resolved_name)
        .with_role(role)
        .with_email_verified(true)
        .with_metadata(metadata);

    let user = ctx.database.create_user(create_user).await?;

    ctx.database
        .create_account(CreateAccount {
            user_id: user.id().to_string(),
            account_id: user.id().to_string(),
            provider_id: "credential".to_string(),
            access_token: None,
            refresh_token: None,
            id_token: None,
            access_token_expires_at: None,
            refresh_token_expires_at: None,
            scope: None,
            password: Some(password_hash),
        })
        .await?;

    Ok(UserResponse { user })
}

pub(crate) async fn list_users_core<DB: DatabaseAdapter>(
    query: &ListUsersQueryParams,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<ListUsersResponse<DB::User>> {
    let limit = query
        .limit
        .unwrap_or(config.default_page_limit)
        .min(config.max_page_limit);
    let offset = query.offset.unwrap_or(0);

    let params: ListUsersParams = ListUsersParams {
        limit: Some(limit),
        offset: Some(offset),
        search_field: query.search_field.clone(),
        search_value: query.search_value.clone(),
        search_operator: query.search_operator.clone(),
        sort_by: query.sort_by.clone(),
        sort_direction: query.sort_direction.clone(),
        filter_field: query.filter_field.clone(),
        filter_value: query.filter_value.clone(),
        filter_operator: query.filter_operator.clone(),
    };

    let (users, total) = ctx.database.list_users(params).await?;
    Ok(ListUsersResponse {
        users,
        total,
        limit,
        offset,
    })
}

pub(crate) async fn list_user_sessions_core<DB: DatabaseAdapter>(
    body: &UserIdRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<ListSessionsResponse<DB::Session>> {
    let _target = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    let session_manager: athena_auth_core::SessionManager<DB> = ctx.session_manager();
    let sessions: Vec<<DB as SessionOps>::Session> =
        session_manager.list_user_sessions(&body.user_id).await?;
    Ok(ListSessionsResponse {
        sessions: sessions_with_server_version(sessions),
    })
}

pub(crate) async fn ban_user_core<DB: DatabaseAdapter>(
    body: &BanUserRequest,
    admin_user_id: &str,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<UserResponse<DB::User>> {
    if body.user_id == admin_user_id {
        return Err(AuthError::bad_request("You cannot ban yourself"));
    }

    let target = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    if !config.allow_ban_admin && target.role().unwrap_or("user") == config.admin_role {
        return Err(AuthError::forbidden("Cannot ban an admin user"));
    }

    let ban_expires = body
        .ban_expires_in
        .and_then(Duration::try_seconds)
        .map(|d| Utc::now() + d);

    let update = UpdateUser {
        banned: Some(true),
        ban_reason: body.ban_reason.clone(),
        ban_expires,
        ..Default::default()
    };

    let updated_user = ctx.database.update_user(&body.user_id, update).await?;
    if let Err(cache_error) = ctx.cache_session_user(updated_user.id(), &updated_user) {
        tracing::warn!(
            user_id = %updated_user.id(),
            error = %cache_error,
            "failed to refresh session user cache after ban-user"
        );
    }

    ctx.session_manager()
        .revoke_all_user_sessions(&body.user_id)
        .await?;

    Ok(UserResponse { user: updated_user })
}

pub(crate) async fn unban_user_core<DB: DatabaseAdapter>(
    body: &UserIdRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<UserResponse<DB::User>> {
    let _target = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    let update = UpdateUser {
        banned: Some(false),
        ban_reason: None,
        ban_expires: None,
        ..Default::default()
    };

    let updated_user = ctx.database.update_user(&body.user_id, update).await?;
    if let Err(cache_error) = ctx.cache_session_user(updated_user.id(), &updated_user) {
        tracing::warn!(
            user_id = %updated_user.id(),
            error = %cache_error,
            "failed to refresh session user cache after unban-user"
        );
    }
    Ok(UserResponse { user: updated_user })
}

pub(crate) async fn impersonate_user_core<DB: DatabaseAdapter>(
    body: &UserIdRequest,
    admin_user_id: &str,
    ip_address: Option<&str>,
    user_agent: Option<&str>,
    ctx: &AuthContext<DB>,
) -> AuthResult<(SessionUserResponse<DB::Session, DB::User>, String)> {
    if body.user_id == admin_user_id {
        return Err(AuthError::bad_request("Cannot impersonate yourself"));
    }

    let target = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    let session = ctx
        .session_manager()
        .create_session_for_user_id(
            target.id(),
            SessionCreateOptions {
                ip_address: ip_address.map(str::to_string),
                user_agent: user_agent.map(str::to_string),
                impersonated_by: Some(admin_user_id.to_string()),
                active_organization_id: None,
                update_last_sign_in_at: false,
            },
        )
        .await?;
    let token = session.token().to_string();
    let response = SessionUserResponse {
        session: session_with_server_version(session),
        user: target,
    };

    Ok((response, token))
}

pub(crate) async fn stop_impersonating_core<DB: DatabaseAdapter>(
    session: &DB::Session,
    session_token: &str,
    ip_address: Option<&str>,
    user_agent: Option<&str>,
    ctx: &AuthContext<DB>,
) -> AuthResult<(SessionUserResponse<DB::Session, DB::User>, String)> {
    let admin_id = session
        .impersonated_by()
        .ok_or_else(|| AuthError::bad_request("Current session is not an impersonation session"))?
        .to_string();

    ctx.session_manager().delete_session(session_token).await?;

    let admin_user = ctx
        .database
        .get_user_by_id(&admin_id)
        .await?
        .ok_or(AuthError::UserNotFound)?;

    let admin_session = ctx
        .session_manager()
        .create_session_for_user_id(
            &admin_id,
            SessionCreateOptions {
                ip_address: ip_address.map(str::to_string),
                user_agent: user_agent.map(str::to_string),
                impersonated_by: None,
                active_organization_id: None,
                update_last_sign_in_at: false,
            },
        )
        .await?;
    let token = admin_session.token().to_string();
    let response = SessionUserResponse {
        session: session_with_server_version(admin_session),
        user: admin_user,
    };

    Ok((response, token))
}

pub(crate) async fn revoke_user_session_core<DB: DatabaseAdapter>(
    body: &RevokeSessionRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SuccessResponse> {
    ctx.session_manager()
        .delete_session(&body.session_token)
        .await?;
    Ok(SuccessResponse { success: true })
}

pub(crate) async fn revoke_user_sessions_core<DB: DatabaseAdapter>(
    body: &UserIdRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SuccessResponse> {
    let _target = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    ctx.session_manager()
        .revoke_all_user_sessions(&body.user_id)
        .await?;

    Ok(SuccessResponse { success: true })
}

pub(crate) async fn remove_user_core<DB: DatabaseAdapter>(
    body: &UserIdRequest,
    admin_user_id: &str,
    ctx: &AuthContext<DB>,
) -> AuthResult<SuccessResponse> {
    if body.user_id == admin_user_id {
        return Err(AuthError::bad_request("You cannot remove yourself"));
    }

    let _target = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    ctx.database.delete_user_sessions(&body.user_id).await?;

    let accounts = ctx.database.get_user_accounts(&body.user_id).await?;
    for account in &accounts {
        ctx.database.delete_account(account.id()).await?;
    }

    ctx.database.delete_user(&body.user_id).await?;
    ctx.invalidate_session_user_cache(&body.user_id);
    Ok(SuccessResponse { success: true })
}

pub(crate) async fn set_user_password_core<DB: DatabaseAdapter>(
    body: &SetUserPasswordRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<StatusResponse> {
    if body.new_password.len() < ctx.config.password.min_length {
        return Err(AuthError::bad_request(format!(
            "Password must be at least {} characters long",
            ctx.config.password.min_length
        )));
    }

    let user = ctx
        .database
        .get_user_by_id(&body.user_id)
        .await?
        .ok_or_else(|| AuthError::not_found("User not found"))?;

    let password_hash = athena_auth_core::hash_password_with_config(
        None,
        Some(&ctx.config.password.argon2_config),
        &body.new_password,
    )
    .await?;

    let mut metadata = user.metadata().clone();
    if let Some(obj) = metadata.as_object_mut() {
        obj.insert(
            PASSWORD_HASH_KEY.to_string(),
            serde_json::json!(password_hash),
        );
    } else {
        return Err(AuthError::bad_request(
            "User metadata must be a JSON object to store password hash",
        ));
    }

    let update = UpdateUser {
        metadata: Some(metadata),
        ..Default::default()
    };
    let updated_user = ctx.database.update_user(&body.user_id, update).await?;
    if let Err(cache_error) = ctx.cache_session_user(updated_user.id(), &updated_user) {
        tracing::warn!(
            user_id = %updated_user.id(),
            error = %cache_error,
            "failed to refresh session user cache after set-user-password"
        );
    }

    let accounts = ctx.database.get_user_accounts(&body.user_id).await?;
    let has_credential = accounts.iter().any(|a| a.provider_id() == "credential");

    if has_credential {
        for account in &accounts {
            if account.provider_id() == "credential" {
                let account_update = athena_auth_core::UpdateAccount {
                    password: Some(password_hash.clone()),
                    ..Default::default()
                };
                ctx.database
                    .update_account(account.id(), account_update)
                    .await?;
                break;
            }
        }
    } else {
        ctx.database
            .create_account(CreateAccount {
                user_id: body.user_id.clone(),
                account_id: body.user_id.clone(),
                provider_id: "credential".to_string(),
                access_token: None,
                refresh_token: None,
                id_token: None,
                access_token_expires_at: None,
                refresh_token_expires_at: None,
                scope: None,
                password: Some(password_hash.clone()),
            })
            .await?;
    }

    Ok(StatusResponse { status: true })
}

pub(crate) async fn has_permission_core<DB: DatabaseAdapter>(
    body: &HasPermissionRequest,
    user: &DB::User,
    config: &AdminConfig,
) -> AuthResult<PermissionResponse> {
    let _permissions = body.permissions.clone().or(body.permission.clone());

    let is_admin = user.role().unwrap_or("user") == config.admin_role;

    let (success, error) = if is_admin {
        (true, None)
    } else {
        (
            false,
            Some("User does not have the required permissions".to_string()),
        )
    };

    Ok(PermissionResponse { success, error })
}

pub(crate) async fn list_grants_core(config: &AdminConfig) -> AuthResult<GrantCatalogResponse> {
    let native_right_catalog = config.merged_native_right_catalog();
    let effective_grant_catalog = config.effective_grant_catalog();
    let catalog_entries = grant_catalog_entries(&effective_grant_catalog, &native_right_catalog);
    let catalog_modules = grant_catalog_modules(
        &effective_grant_catalog,
        &native_right_catalog,
        &config.native_right_domain_contracts,
    );
    Ok(GrantCatalogResponse {
        grant_catalog: catalog_entries.clone(),
        rights_catalog: catalog_entries,
        native_right_catalog: native_right_catalog_entries(
            &effective_grant_catalog,
            &native_right_catalog,
        ),
        integration_contracts: native_right_domain_integration_contracts(&catalog_modules),
        catalog_modules,
        native_rights: native_right_catalog,
    })
}

pub(crate) async fn list_grant_assignments_core<DB: DatabaseAdapter>(
    query: &ListGrantAssignmentsQueryParams,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<SubjectGrantAssignmentsResponse> {
    let subject_kind = normalize_non_empty_string(&query.subject_kind, "subjectKind")?;
    let subject_id = normalize_non_empty_string(&query.subject_id, "subjectId")?;
    let assignments = ctx
        .database
        .list_grant_assignments_for_subject(&subject_kind, &subject_id)
        .await?;
    Ok(build_subject_grant_assignments_response(
        subject_kind,
        subject_id,
        assignments,
        &config.grant_catalog,
        &config.native_right_catalog,
    ))
}

fn build_subject_grant_assignments_response(
    subject_kind: String,
    subject_id: String,
    assignments: Vec<athena_auth_core::GrantAssignment>,
    grant_catalog: &[athena_auth_core::GrantDescriptor],
    native_right_catalog: &[athena_rights::AthenaRightDescriptor],
) -> SubjectGrantAssignmentsResponse {
    let native_right_catalog = merged_native_right_catalog(native_right_catalog);
    let effective_grant_catalog =
        crate::plugins::grants::merged_grant_catalog(grant_catalog, &native_right_catalog);
    let grant_keys = grant_keys_from_assignments(assignments.iter());
    let resolved_grants = resolved_grant_assignment_views(
        resolve_grant_assignments(assignments.iter(), &effective_grant_catalog),
        &native_right_catalog,
    );
    let native_rights =
        native_rights_for_grant_keys(grant_keys.iter().map(String::as_str), &native_right_catalog);
    let native_right_modules = native_right_modules_for_grant_keys(
        grant_keys.iter().map(String::as_str),
        &native_right_catalog,
    );
    let right_modules = right_module_inputs_from_native_modules(&native_right_modules);
    let native_right_assignments =
        native_right_assignment_views(assignments.iter(), &native_right_catalog);
    let resolved_native_rights = resolved_native_right_assignment_views(
        assignments.iter(),
        &effective_grant_catalog,
        &native_right_catalog,
    );
    let permissions =
        permissions_map_from_assignments(assignments.iter(), &effective_grant_catalog);
    let right_keys = native_right_keys_for_grant_keys(
        grant_keys.iter().map(String::as_str),
        &native_right_catalog,
    );
    let unknown_grant_keys: Vec<String> = assignments
        .iter()
        .filter(|assignment| {
            grant_by_key(&effective_grant_catalog, &assignment.grant_key).is_none()
        })
        .map(|assignment| assignment.grant_key.clone())
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect();
    let unknown_right_keys = unknown_native_right_keys_for_grant_keys(
        grant_keys.iter().map(String::as_str),
        &native_right_catalog,
    );

    SubjectGrantAssignmentsResponse {
        subject_kind,
        subject_id,
        grant_keys,
        right_keys,
        right_modules,
        native_right_modules,
        right_assignments: assignments.clone(),
        grant_assignments: assignments,
        resolved_rights: resolved_grants.clone(),
        resolved_grants,
        resolved_native_rights,
        native_rights,
        native_right_assignments,
        permissions,
        unknown_right_keys,
        unknown_grant_keys,
    }
}

pub(crate) async fn replace_grant_assignments_for_source_core<DB: DatabaseAdapter>(
    body: &ReplaceGrantAssignmentsForSourceRequest,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<ReplaceGrantAssignmentsForSourceResponse> {
    let effective_grant_catalog = config.effective_grant_catalog();
    let subject_kind = normalize_non_empty_string(&body.subject_kind, "subjectKind")?;
    let subject_id = normalize_non_empty_string(&body.subject_id, "subjectId")?;
    let source_kind = normalize_non_empty_string(&body.source_kind, "sourceKind")?;
    let source_ref = normalize_non_empty_string(&body.source_ref, "sourceRef")?;
    let normalized_input_keys = normalize_grant_inputs(
        body.grant_keys.iter().map(String::as_str),
        body.right_keys.iter().map(String::as_str),
        body.right_modules.iter(),
    );
    let applied_grant_keys = normalized_input_keys
        .iter()
        .map(|value| normalize_non_empty_string(value, "grantKeys"))
        .collect::<AuthResult<BTreeSet<_>>>()?
        .into_iter()
        .collect::<Vec<_>>();
    let native_right_catalog = merged_native_right_catalog(&config.native_right_catalog);
    let applied_right_keys = native_right_keys_for_grant_keys(
        applied_grant_keys.iter().map(String::as_str),
        &native_right_catalog,
    );
    let applied_native_right_modules = native_right_modules_for_grant_keys(
        applied_grant_keys.iter().map(String::as_str),
        &native_right_catalog,
    );
    let applied_right_modules = right_modules_for_grant_keys(
        applied_grant_keys.iter().map(String::as_str),
        &native_right_catalog,
    );
    let metadata = normalize_metadata_object(body.metadata.as_ref(), "metadata")?
        .unwrap_or_else(|| serde_json::json!({}));

    let replaced_assignments = ctx
        .database
        .replace_grant_assignments_for_source(ReplaceGrantAssignmentsForSource {
            subject_kind: subject_kind.clone(),
            subject_id: subject_id.clone(),
            source_kind: source_kind.clone(),
            source_ref: source_ref.clone(),
            grant_keys: applied_grant_keys.clone(),
            right_modules: body.right_modules.clone(),
            metadata,
        })
        .await?;
    let replaced_native_right_assignments =
        native_right_assignment_views(replaced_assignments.iter(), &native_right_catalog);
    let subject_assignments = ctx
        .database
        .list_grant_assignments_for_subject(&subject_kind, &subject_id)
        .await?;
    let subject_state = build_subject_grant_assignments_response(
        subject_kind.clone(),
        subject_id.clone(),
        subject_assignments,
        &effective_grant_catalog,
        &config.native_right_catalog,
    );

    Ok(ReplaceGrantAssignmentsForSourceResponse {
        subject_kind,
        subject_id,
        source_kind,
        source_ref,
        applied_right_keys,
        applied_right_modules,
        applied_native_right_modules,
        applied_grant_keys,
        replaced_native_right_assignments,
        replaced_assignments,
        subject_state,
    })
}

pub(crate) async fn sync_billing_grant_assignments_core<DB: DatabaseAdapter>(
    body: &SyncBillingGrantAssignmentsRequest,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<ReplaceGrantAssignmentsForSourceResponse> {
    replace_grant_assignments_for_source_core(
        &ReplaceGrantAssignmentsForSourceRequest {
            subject_kind: body.subject_kind.clone(),
            subject_id: body.subject_id.clone(),
            source_kind: body.source_kind.clone(),
            source_ref: body.source_ref.clone(),
            grant_keys: body.grant_keys.clone(),
            right_keys: body.right_keys.clone(),
            right_modules: body.right_modules.clone(),
            metadata: body.metadata.clone(),
        },
        config,
        ctx,
    )
    .await
}

pub(crate) async fn sync_billing_right_assignments_core<DB: DatabaseAdapter>(
    body: &SyncBillingRightAssignmentsRequest,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<ReplaceNativeRightAssignmentsForSourceResponse> {
    let request: ReplaceNativeRightAssignmentsForSourceRequest =
        ReplaceGrantAssignmentsForSourceRequest {
            subject_kind: body.subject_kind.clone(),
            subject_id: body.subject_id.clone(),
            source_kind: body.source_kind.clone(),
            source_ref: body.source_ref.clone(),
            grant_keys: body.grant_keys.clone(),
            right_keys: body.right_keys.clone(),
            right_modules: body.right_modules.clone(),
            metadata: body.metadata.clone(),
        };
    let response = replace_grant_assignments_for_source_core(&request, config, ctx).await?;
    Ok(response)
}

pub(crate) async fn sync_grant_assignments_core<DB: DatabaseAdapter>(
    body: &SyncGrantAssignmentsRequest,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<ReplaceGrantAssignmentsForSourceResponse> {
    replace_grant_assignments_for_source_core(
        &ReplaceGrantAssignmentsForSourceRequest {
            subject_kind: body.subject_kind.clone(),
            subject_id: body.subject_id.clone(),
            source_kind: body.source_kind.clone(),
            source_ref: body.source_ref.clone(),
            grant_keys: body.grant_keys.clone(),
            right_keys: body.right_keys.clone(),
            right_modules: body.right_modules.clone(),
            metadata: body.metadata.clone(),
        },
        config,
        ctx,
    )
    .await
}

pub(crate) async fn sync_native_right_assignments_core<DB: DatabaseAdapter>(
    body: &SyncNativeRightAssignmentsRequest,
    config: &AdminConfig,
    ctx: &AuthContext<DB>,
) -> AuthResult<ReplaceNativeRightAssignmentsForSourceResponse> {
    let request: ReplaceNativeRightAssignmentsForSourceRequest =
        ReplaceGrantAssignmentsForSourceRequest {
            subject_kind: body.subject_kind.clone(),
            subject_id: body.subject_id.clone(),
            source_kind: body.source_kind.clone(),
            source_ref: body.source_ref.clone(),
            grant_keys: body.grant_keys.clone(),
            right_keys: body.right_keys.clone(),
            right_modules: body.right_modules.clone(),
            metadata: body.metadata.clone(),
        };
    let response = replace_grant_assignments_for_source_core(&request, config, ctx).await?;
    Ok(response)
}

fn hash_api_key(raw_key: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(raw_key.as_bytes());
    let digest = hasher.finalize();
    URL_SAFE_NO_PAD.encode(digest)
}

fn generate_admin_key() -> (String, String, String) {
    let mut bytes = vec![0u8; 32];
    rand::rngs::OsRng.fill_bytes(&mut bytes);
    let raw = URL_SAFE_NO_PAD.encode(&bytes);
    let full_key = format!("adm_{}", raw);
    let start = raw.chars().take(6).collect::<String>();
    let key_hash = hash_api_key(&full_key);
    (full_key, key_hash, start)
}

pub(crate) async fn create_admin_api_key_core<DB: DatabaseAdapter>(
    body: &CreateAdminApiKeyRequest,
    config: &AdminConfig,
    admin_user: &DB::User,
    ctx: &AuthContext<DB>,
) -> AuthResult<CreateAdminApiKeyResponse> {
    let effective_grant_catalog = config.effective_grant_catalog();
    if let Some(client_name) = &body.athena_client_name {
        let exists = ctx
            .database
            .list_athena_clients()
            .await?
            .into_iter()
            .any(|client| client.client_name == *client_name && client.is_active);
        if !exists {
            return Err(AuthError::bad_request(
                "athenaClientName does not exist or is not active",
            ));
        }
    }

    let (raw_key, key_hash, start) = generate_admin_key();
    let now = Utc::now();
    let expires_at = body
        .expires_in
        .map(|ms| now + chrono::Duration::milliseconds(ms))
        .map(|dt| dt.to_rfc3339());

    let base_permissions = body
        .permissions
        .clone()
        .unwrap_or_else(|| serde_json::json!({ "admin": ["*"] }));
    let permissions = normalize_api_key_permissions_input(
        Some(&base_permissions),
        body.grant_keys.as_ref(),
        body.right_keys.as_ref(),
        body.right_modules.as_ref(),
        &effective_grant_catalog,
    )
    .map(|value| value.to_string());

    let metadata = {
        let mut object = match body.metadata.clone() {
            Some(serde_json::Value::Object(map)) => map,
            _ => serde_json::Map::new(),
        };
        if let Some(client_name) = &body.athena_client_name {
            object.insert(
                "athenaClientName".to_string(),
                serde_json::Value::String(client_name.clone()),
            );
        }
        build_stored_api_key_metadata(
            None,
            Some(&serde_json::Value::Object(object)),
            body.grant_keys.as_ref(),
            body.right_keys.as_ref(),
            body.right_modules.as_ref(),
        )
    };

    let created = ctx
        .database
        .create_api_key(CreateApiKey {
            user_id: admin_user.id().to_string(),
            name: body.name.clone(),
            prefix: Some("adm_".to_string()),
            key_hash,
            start: Some(start),
            expires_at,
            remaining: None,
            rate_limit_enabled: false,
            rate_limit_time_window: None,
            rate_limit_max: None,
            refill_interval: None,
            refill_amount: None,
            permissions,
            metadata,
            enabled: true,
        })
        .await?;

    let grant_keys = stored_api_key_grant_keys(created.metadata());
    let right_modules = stored_api_key_right_modules(created.metadata());
    let native_right_catalog = merged_native_right_catalog(&config.native_right_catalog);
    let right_keys = native_right_keys_for_grant_keys(
        grant_keys.iter().map(String::as_str),
        &native_right_catalog,
    );
    let response = CreateAdminApiKeyResponse {
        key: raw_key,
        api_key: AdminApiKeyView {
            id: created.id().to_string(),
            name: created.name().map(|s| s.to_string()),
            user_id: created.user_id().to_string(),
            prefix: created.prefix().map(|s| s.to_string()),
            enabled: created.enabled(),
            expires_at: created.expires_at().map(|s| s.to_string()),
            created_at: created.created_at().to_string(),
            permissions: created
                .permissions()
                .and_then(|value| serde_json::from_str(value).ok()),
            right_modules,
            native_right_modules: native_right_modules_for_grant_keys(
                grant_keys.iter().map(String::as_str),
                &native_right_catalog,
            ),
            native_rights: native_rights_for_grant_keys(
                grant_keys.iter().map(String::as_str),
                &native_right_catalog,
            ),
            right_keys,
            grant_keys,
            metadata: stored_api_key_user_metadata(created.metadata()),
        },
    };

    Ok(response)
}

pub(crate) async fn create_athena_client_core<DB: DatabaseAdapter>(
    body: &CreateAthenaClientRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<athena_auth_core::AthenaClient> {
    ctx.database
        .create_athena_client(CreateAthenaClient {
            client_name: body.client_name.clone(),
            description: body.description.clone(),
            is_active: body.is_active,
            pg_uri_env_var: body.pg_uri_env_var.clone(),
            metadata: body.metadata.clone(),
        })
        .await
}

pub(crate) async fn list_athena_clients_core<DB: DatabaseAdapter>(
    ctx: &AuthContext<DB>,
) -> AuthResult<AthenaClientsResponse> {
    let clients = ctx.database.list_athena_clients().await?;
    Ok(AthenaClientsResponse {
        athena_clients: clients,
    })
}

pub(crate) async fn create_oauth_client_core<DB: DatabaseAdapter>(
    body: &CreateOAuthClientRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<athena_auth_core::OAuthClient> {
    let provider = normalize_non_empty_string(&body.provider, "provider")?.to_ascii_lowercase();
    let redirect_uris = normalize_string_list(&body.redirect_uris, "redirectUris")?;
    let scopes = body
        .scopes
        .as_ref()
        .map(|values| normalize_string_list(values, "scopes"))
        .transpose()?;

    ctx.database
        .create_oauth_client(CreateOAuthClient {
            client_name: normalize_non_empty_string(&body.client_name, "clientName")?,
            provider,
            description: body.description.clone(),
            response_type: body.response_type.clone(),
            grant_type: body.grant_type.clone(),
            token_authentication_method: body.token_authentication_method.clone(),
            redirect_uris,
            client_url: body.client_url.clone(),
            scopes,
            is_active: body.is_active,
            metadata: body.metadata.clone(),
        })
        .await
}

pub(crate) async fn list_oauth_clients_core<DB: DatabaseAdapter>(
    ctx: &AuthContext<DB>,
) -> AuthResult<OAuthClientsResponse> {
    let clients = ctx.database.list_oauth_clients().await?;
    Ok(OAuthClientsResponse {
        oauth_clients: clients,
    })
}

pub(crate) async fn list_audit_logs_core<DB: DatabaseAdapter>(
    query: &ListAuditLogsQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<AuditLogsResponse> {
    let limit = query.limit.unwrap_or(100).min(500);
    let offset = query.offset.unwrap_or(0);
    let created_after = parse_optional_rfc3339_utc(query.created_after.as_deref(), "createdAfter")?;
    let created_before =
        parse_optional_rfc3339_utc(query.created_before.as_deref(), "createdBefore")?;
    let params = ListAdminAuditLogsParams {
        limit: Some(limit),
        offset: Some(offset),
        actor_user_id: query.actor_user_id.clone(),
        action: query.action.clone(),
        target_type: query.target_type.clone(),
        target_id: query.target_id.clone(),
        success: query.success,
        created_after,
        created_before,
    };
    let total = ctx.database.count_admin_audit_logs(params.clone()).await?;
    let logs = ctx
        .database
        .list_admin_audit_logs_with_params(params)
        .await?;
    Ok(AuditLogsResponse {
        total,
        audit_logs: logs,
        limit,
        offset,
    })
}

pub(crate) async fn list_sent_emails_core<DB: DatabaseAdapter>(
    query: &ListSentEmailsQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<SentEmailsResponse> {
    let limit = query.limit.unwrap_or(100).min(1000);
    let offset = query.offset.unwrap_or(0);
    let created_after = parse_optional_rfc3339_utc(query.created_after.as_deref(), "createdAfter")?;
    let created_before =
        parse_optional_rfc3339_utc(query.created_before.as_deref(), "createdBefore")?;
    let params = ListSentEmailsParams {
        limit: Some(limit),
        offset: Some(offset),
        recipient_email: query.recipient_email.clone(),
        provider: query.provider.clone(),
        flow: query.flow.clone(),
        subject_contains: query.subject_contains.clone(),
        created_after,
        created_before,
    };
    let total = ctx.database.count_sent_emails(params.clone()).await?;
    let emails = ctx.database.list_sent_emails_with_params(params).await?;
    Ok(SentEmailsResponse {
        total,
        emails,
        limit,
        offset,
    })
}

pub(crate) async fn get_sent_email_core<DB: DatabaseAdapter>(
    query: &GetSentEmailQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<SentEmailResponse> {
    let email = ctx
        .database
        .get_sent_email_by_id(query.id.trim())
        .await?
        .ok_or_else(|| AuthError::not_found("Sent email not found"))?;
    Ok(SentEmailResponse { email })
}

pub(crate) async fn create_sent_email_core<DB: DatabaseAdapter>(
    body: &CreateSentEmailRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SentEmailResponse> {
    let recipient_email = normalize_required_email(&body.recipient_email, "recipientEmail")?;
    let subject = normalize_required_trimmed(&body.subject, "subject")?;
    let from_address = normalize_required_email(&body.from_address, "fromAddress")?;
    let provider = normalize_required_trimmed(&body.provider, "provider")?;
    let metadata = normalize_metadata_object(body.metadata.as_ref(), "metadata")?;
    let email = ctx
        .database
        .create_sent_email_record(CreateSentEmail {
            recipient_email,
            subject,
            from_address,
            from_name: normalize_optional_trimmed(body.from_name.as_deref()),
            text_body: normalize_optional_trimmed(body.text_body.as_deref()),
            html_body: normalize_optional_trimmed(body.html_body.as_deref()),
            provider,
            flow: normalize_optional_trimmed(body.flow.as_deref()),
            metadata,
        })
        .await?;
    Ok(SentEmailResponse { email })
}

pub(crate) async fn update_sent_email_core<DB: DatabaseAdapter>(
    body: &UpdateSentEmailRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SentEmailResponse> {
    let recipient_email =
        normalize_optional_email(body.recipient_email.as_deref(), "recipientEmail")?;
    let subject = normalize_optional_non_empty(body.subject.as_deref(), "subject")?;
    let from_address = normalize_optional_email(body.from_address.as_deref(), "fromAddress")?;
    let provider = normalize_optional_non_empty(body.provider.as_deref(), "provider")?;
    let metadata = normalize_metadata_object(body.metadata.as_ref(), "metadata")?;
    let email = ctx
        .database
        .update_sent_email(
            &body.id,
            UpdateSentEmail {
                recipient_email,
                subject,
                from_address,
                from_name: normalize_nullable_trimmed_patch(body.from_name.clone()),
                text_body: normalize_nullable_trimmed_patch(body.text_body.clone()),
                html_body: normalize_nullable_trimmed_patch(body.html_body.clone()),
                provider,
                flow: normalize_nullable_trimmed_patch(body.flow.clone()),
                metadata,
            },
        )
        .await?;

    Ok(SentEmailResponse { email })
}

pub(crate) async fn delete_sent_email_core<DB: DatabaseAdapter>(
    body: &DeleteSentEmailRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SuccessResponse> {
    ctx.database.delete_sent_email(body.id.trim()).await?;
    Ok(SuccessResponse { success: true })
}

pub(crate) async fn list_email_send_failures_core<DB: DatabaseAdapter>(
    query: &ListEmailSendFailuresQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailSendFailuresResponse> {
    let limit = query.limit.unwrap_or(100).min(1000);
    let offset = query.offset.unwrap_or(0);
    let created_after = parse_optional_rfc3339_utc(query.created_after.as_deref(), "createdAfter")?;
    let created_before =
        parse_optional_rfc3339_utc(query.created_before.as_deref(), "createdBefore")?;

    let params = ListEmailSendFailuresParams {
        limit: Some(limit),
        offset: Some(offset),
        recipient_email: query.recipient_email.clone(),
        provider: query.provider.clone(),
        flow: query.flow.clone(),
        resolved: query.resolved,
        created_after,
        created_before,
    };
    let total = ctx
        .database
        .count_email_send_failures(params.clone())
        .await?;

    let rows = ctx.database.list_email_send_failures(params).await?;

    Ok(EmailSendFailuresResponse {
        total,
        email_send_failures: rows,
        limit,
        offset,
    })
}

pub(crate) async fn create_email_send_failure_core<DB: DatabaseAdapter>(
    body: &CreateEmailSendFailureRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailSendFailureResponse> {
    let recipient_email = normalize_required_email(&body.recipient_email, "recipientEmail")?;
    let flow = normalize_required_trimmed(&body.flow, "flow")?;
    let provider = normalize_optional_non_empty(body.provider.as_deref(), "provider")?;
    let error_message = normalize_required_trimmed(&body.error_message, "errorMessage")?;
    let metadata = normalize_metadata_object(body.metadata.as_ref(), "metadata")?;
    let email_send_failure = ctx
        .database
        .create_email_send_failure_record(CreateEmailSendFailure {
            user_id: normalize_optional_trimmed(body.user_id.as_deref()),
            recipient_email,
            flow,
            provider,
            error_message,
            metadata,
        })
        .await?;
    Ok(EmailSendFailureResponse { email_send_failure })
}

pub(crate) async fn get_email_send_failure_core<DB: DatabaseAdapter>(
    query: &GetEmailSendFailureQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailSendFailureResponse> {
    let email_send_failure = ctx
        .database
        .get_email_send_failure_by_id(query.id.trim())
        .await?
        .ok_or_else(|| AuthError::not_found("Email send failure not found"))?;
    Ok(EmailSendFailureResponse { email_send_failure })
}

pub(crate) async fn update_email_send_failure_core<DB: DatabaseAdapter>(
    body: &UpdateEmailSendFailureRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailSendFailureResponse> {
    let metadata = normalize_metadata_object(body.metadata.as_ref(), "metadata")?;
    let email_send_failure = ctx
        .database
        .update_email_send_failure(
            &body.id,
            UpdateEmailSendFailure {
                resolved: body.resolved,
                resolution_note: normalize_nullable_trimmed_patch(body.resolution_note.clone()),
                metadata,
            },
        )
        .await?;

    Ok(EmailSendFailureResponse { email_send_failure })
}

pub(crate) async fn delete_email_send_failure_core<DB: DatabaseAdapter>(
    body: &DeleteEmailSendFailureRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SuccessResponse> {
    ctx.database
        .delete_email_send_failure(body.id.trim())
        .await?;
    Ok(SuccessResponse { success: true })
}

pub(crate) async fn create_email_template_core<DB: DatabaseAdapter>(
    body: &CreateEmailTemplateRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailTemplateApiResponse> {
    let template_key = normalize_required_trimmed(&body.template_key, "template_key")?;
    let mut event_type = normalize_optional_event_type(body.event_type.as_deref(), "event_type")?;
    if event_type.is_none() {
        event_type = default_event_type_for_template_key(&template_key).map(str::to_string);
    }
    if let Some(ref value) = event_type {
        ensure_email_event_type_exists(ctx, value).await?;
    }
    let subject_template = normalize_required_trimmed(&body.subject_template, "subject_template")?;
    let locale = normalize_optional_non_empty(body.locale.as_deref(), "locale")?;
    let variables = normalize_template_variables(body.variables.as_ref())?;
    let variable_bindings = normalize_template_variable_bindings(body.variable_bindings.as_ref())?;
    let attachments = normalize_template_attachments(body.attachments.as_ref())?;
    let attachment_failure_mode =
        normalize_attachment_failure_mode(body.attachment_failure_mode.as_deref())?;
    let metadata = normalize_metadata_object(body.metadata.as_ref(), "metadata")?;
    let template = ctx
        .database
        .create_email_template(CreateEmailTemplate {
            template_key,
            event_type,
            locale,
            subject_template,
            text_template: normalize_optional_trimmed(body.text_template.as_deref()),
            html_template: normalize_optional_trimmed(body.html_template.as_deref()),
            variables,
            variable_bindings,
            attachments,
            attachment_failure_mode,
            is_active: body.is_active,
            metadata,
        })
        .await?;

    Ok(EmailTemplateApiResponse::from(template))
}

pub(crate) async fn list_email_templates_core<DB: DatabaseAdapter>(
    query: &ListEmailTemplatesQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailTemplatesResponse> {
    let limit = query.limit.unwrap_or(100).min(1000);
    let offset = query.offset.unwrap_or(0);
    let event_type = normalize_optional_event_type(query.event_type.as_deref(), "event_type")?;
    let params = ListEmailTemplatesParams {
        limit: Some(limit),
        offset: Some(offset),
        template_key: query.template_key.clone(),
        event_type,
        locale: query.locale.clone(),
        is_active: query.is_active,
    };
    let total = ctx.database.count_email_templates(params.clone()).await?;
    let templates = ctx.database.list_email_templates(params).await?;

    Ok(EmailTemplatesResponse {
        total,
        email_templates: templates
            .into_iter()
            .map(EmailTemplateApiResponse::from)
            .collect(),
        limit,
        offset,
    })
}

pub(crate) async fn get_email_template_core<DB: DatabaseAdapter>(
    query: &GetEmailTemplateQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailTemplateResponse> {
    let email_template = ctx
        .database
        .get_email_template_by_id(query.id.trim())
        .await?
        .ok_or_else(|| AuthError::not_found("Email template not found"))?;
    Ok(EmailTemplateResponse {
        email_template: EmailTemplateApiResponse::from(email_template),
    })
}

pub(crate) async fn update_email_template_core<DB: DatabaseAdapter>(
    body: &UpdateEmailTemplateRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailTemplateApiResponse> {
    let template_key = normalize_optional_non_empty(body.template_key.as_deref(), "template_key")?;
    let event_type = normalize_nullable_trimmed_patch(body.event_type.clone());
    if let Some(Some(ref value)) = event_type {
        normalize_optional_event_type(Some(value.as_str()), "event_type")?;
        ensure_email_event_type_exists(ctx, value).await?;
    }
    let locale = normalize_optional_non_empty(body.locale.as_deref(), "locale")?;
    let subject_template =
        normalize_optional_non_empty(body.subject_template.as_deref(), "subject_template")?;
    let variables = normalize_template_variables(body.variables.as_ref())?;
    let variable_bindings = normalize_template_variable_bindings(body.variable_bindings.as_ref())?;
    let attachments = match body.attachments.as_ref() {
        Some(Some(value)) => Some(Some(
            normalize_template_attachments(Some(value))?
                .unwrap_or_else(|| serde_json::Value::Array(Vec::new())),
        )),
        Some(None) => Some(None),
        None => None,
    };
    let attachment_failure_mode = match body.attachment_failure_mode.as_ref() {
        Some(Some(value)) => Some(Some(
            normalize_attachment_failure_mode(Some(value.as_str()))?
                .unwrap_or_else(|| "fail".to_string()),
        )),
        Some(None) => Some(None),
        None => None,
    };
    let metadata = normalize_metadata_object(body.metadata.as_ref(), "metadata")?;
    let template = ctx
        .database
        .update_email_template(
            &body.id,
            UpdateEmailTemplate {
                template_key,
                event_type,
                locale,
                subject_template,
                text_template: normalize_nullable_trimmed_patch(body.text_template.clone()),
                html_template: normalize_nullable_trimmed_patch(body.html_template.clone()),
                variables,
                variable_bindings,
                attachments,
                attachment_failure_mode,
                is_active: body.is_active,
                metadata,
            },
        )
        .await?;

    Ok(EmailTemplateApiResponse::from(template))
}

pub(crate) async fn delete_email_template_core<DB: DatabaseAdapter>(
    body: &DeleteEmailTemplateRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SuccessResponse> {
    ctx.database.delete_email_template(&body.id).await?;
    Ok(SuccessResponse { success: true })
}

pub(crate) async fn send_email_template_core<DB: DatabaseAdapter>(
    body: &SendEmailTemplateRequest,
    ctx: &AuthContext<DB>,
) -> AuthResult<SendEmailTemplateCoreResult> {
    let template_id = normalize_required_trimmed(&body.template_id, "template_id")?;
    let recipient_email = normalize_required_email(&body.recipient_email, "recipient_email")?;
    let user_id = normalize_optional_non_empty(body.user_id.as_deref(), "user_id")?;
    let organization_id =
        normalize_optional_non_empty(body.organization_id.as_deref(), "organization_id")?;
    let session_token =
        normalize_optional_non_empty(body.session_token.as_deref(), "session_token")?;
    let render_variables = normalize_render_variables(body.render_variables.as_ref())?;
    let request_attachment_value = normalize_template_attachments(body.attachments.as_ref())?;
    let request_attachments = parse_attachment_specs(request_attachment_value.as_ref())?;
    let request_attachment_failure_mode =
        normalize_attachment_failure_mode(body.attachment_failure_mode.as_deref())?;

    let template = ctx
        .database
        .get_email_template_by_id(&template_id)
        .await?
        .ok_or_else(|| AuthError::not_found("Email template not found"))?;

    if !template.is_active {
        return Err(AuthError::bad_request(
            "Email template is inactive and cannot be sent",
        ));
    }

    let provider = match ctx.email_provider() {
        Ok(provider) => provider,
        Err(_error) => {
            return Ok(build_admin_email_template_send_failure(
                ctx,
                body,
                &template,
                None,
                503,
                "No email provider configured",
                None,
            )
            .await);
        }
    };

    let mut render_context = AuthEmailRenderContext::from_variables(render_variables);
    if let Some(value) = user_id {
        render_context = render_context.with_user_id(value);
    }
    if let Some(value) = organization_id {
        render_context = render_context.with_organization_id(value);
    }
    if let Some(value) = session_token {
        render_context = render_context.with_session_token(value);
    }

    let mut bodies = match render_stored_auth_email_template(ctx, &template, render_context).await {
        Ok(bodies) => bodies,
        Err(error) => {
            return Ok(build_admin_email_template_send_failure(
                ctx,
                body,
                &template,
                Some(DEFAULT_EMAIL_TRANSPORT_ID),
                400,
                error.to_string(),
                None,
            )
            .await);
        }
    };
    bodies.attachments = merge_attachment_specs(bodies.attachments, request_attachments);
    if let Some(value) = request_attachment_failure_mode.as_deref() {
        bodies.attachment_failure_mode =
            athena_auth_core::AttachmentFailureMode::parse(value).unwrap_or_default();
    }

    let message = match build_auth_email_message(ctx, &recipient_email, &bodies).await {
        Ok(message) => message,
        Err(error) => {
            return Ok(build_admin_email_template_send_failure(
                ctx,
                body,
                &template,
                Some(DEFAULT_EMAIL_TRANSPORT_ID),
                502,
                error.to_string(),
                error.metadata().cloned(),
            )
            .await);
        }
    };

    if let Err(error) = provider.send_message(&message).await {
        return Ok(build_admin_email_template_send_failure(
            ctx,
            body,
            &template,
            Some(DEFAULT_EMAIL_TRANSPORT_ID),
            502,
            error.to_string(),
            message.metadata.clone(),
        )
        .await);
    }

    Ok(SendEmailTemplateCoreResult {
        status: 200,
        response: SendEmailTemplateResponse {
            success: true,
            flow: ADMIN_EMAIL_TEMPLATE_SEND_FLOW.to_string(),
            template_id: template.id.clone(),
            template_key: template.template_key.clone(),
            event_type: resolved_template_event_type(&template),
            recipient_email,
            subject: Some(message.subject),
            attachment_count: message.attachments.len(),
            delivery_metadata: message.metadata,
            email_send_failure_id: None,
            error: None,
        },
    })
}

pub(crate) async fn list_email_event_types_core<DB: DatabaseAdapter>(
    query: &ListEmailEventTypesQueryParams,
    ctx: &AuthContext<DB>,
) -> AuthResult<EmailEventTypesResponse> {
    let limit = query.limit.unwrap_or(100).min(1000);
    let offset = query.offset.unwrap_or(0);
    let event_type = normalize_optional_event_type(query.event_type.as_deref(), "eventType")?;
    let category = normalize_optional_non_empty(query.category.as_deref(), "category")?;
    let params = ListEmailEventTypesParams {
        limit: Some(limit),
        offset: Some(offset),
        event_type,
        category,
        is_active: query.is_active,
        is_system: query.is_system,
    };

    let total = ctx.database.count_email_event_types(params.clone()).await?;
    let email_event_types = ctx.database.list_email_event_types(params).await?;

    Ok(EmailEventTypesResponse {
        email_event_types,
        total,
        limit,
        offset,
    })
}
