//! `UsersService` implementation backed by Postgres (v1.1.8).
//!
//! Wires the `app_user_repo` (Argon2id-hashed user rows), the
//! `app_user_session_repo` (SHA-256-hashed sliding-window sessions),
//! the authz layer (so script-side calls require the right capability
//! when the principal is non-public), the `ServiceEventEmitter` (so
//! every mutation publishes an event for future triggers), and the
//! shared email service (for the verification / reset / invitation
//! flows in later commits).
//!
//! Cross-app isolation lives on every method that takes
//! `&SdkCallCx`: we read `cx.app_id` and never trust a script-passed
//! identifier. The `verify_session_for_realtime` method takes
//! `app_id` explicitly because it's invoked from the realtime
//! authority (no script execution context); the value comes from the
//! topic row, also not script-controlled.
//!
//! Commit 4 ships CRUD + login/verify/logout + verify_session_for_realtime.
//! Email-tied flows (verification, password reset, invitations),
//! roles, and admin-mediated helpers are stubbed with
//! `UsersError::Backend("not yet implemented")` and ship in later
//! commits in the v1.1.8 series.

use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use chrono::Utc;
use picloud_shared::{
    AppId, AppUser, AppUserId, CreateUserInput, EmailError, EmailService, EmailTemplateOpts,
    GeneratedAccept, GeneratedSession, Invitation, InvitationId, InviteOpts, OutboundEmail,
    Principal, SdkCallCx, ServiceEvent, ServiceEventEmitter, UpdateUserInput, UsersError,
    UsersListOpts, UsersListPage, UsersService,
};

use crate::app_user_invitation_repo::{AppUserInvitationRepo, AppUserInvitationRepoError};
use crate::app_user_password_reset_repo::{
    AppUserPasswordResetRepo, AppUserPasswordResetRepoError,
};
use crate::app_user_repo::{
    AppUserRepository, AppUserRepositoryError, AppUserRow, ListOpts as RepoListOpts,
};
use crate::app_user_role_repo::{AppUserRoleRepo, AppUserRoleRepoError};
use crate::app_user_session_repo::{AppUserSessionRepository, AppUserSessionRepositoryError};
use crate::app_user_verification_repo::{AppUserVerificationRepo, AppUserVerificationRepoError};
use crate::auth::{
    generate_session_token, hash_password, hash_token, verify_password, TIMING_FLAT_DUMMY_HASH,
};
use crate::authz::{self, AuthzRepo, Capability};

/// Runtime configuration. Defaults match the v1.1.8 brief — overridable
/// via env vars wired in the picloud binary.
#[derive(Debug, Clone, Copy)]
pub struct UsersServiceConfig {
    /// Sliding-window TTL on a session (default 24h).
    pub session_ttl: Duration,
    /// Hard cap from session creation (default 30 days). Beyond this
    /// the session is dead regardless of recent activity.
    pub session_absolute_ttl: Duration,
    /// Verification-token TTL (default 48h).
    pub verification_ttl: Duration,
    /// Password-reset-token TTL (default 1h).
    pub password_reset_ttl: Duration,
    /// Invitation-token TTL (default 7 days).
    pub invitation_ttl: Duration,
}

impl Default for UsersServiceConfig {
    fn default() -> Self {
        Self {
            session_ttl: Duration::from_secs(24 * 3600),
            session_absolute_ttl: Duration::from_secs(30 * 24 * 3600),
            verification_ttl: Duration::from_secs(48 * 3600),
            password_reset_ttl: Duration::from_secs(3600),
            invitation_ttl: Duration::from_secs(7 * 24 * 3600),
        }
    }
}

impl UsersServiceConfig {
    /// Resolve all knobs from the process environment. Falls back to
    /// the documented defaults on any parse failure (logging a warn).
    #[must_use]
    pub fn from_env() -> Self {
        let default = Self::default();
        Self {
            session_ttl: env_duration_hours(
                "PICLOUD_APP_USER_SESSION_TTL_HOURS",
                default.session_ttl,
            ),
            session_absolute_ttl: env_duration_hours(
                "PICLOUD_APP_USER_SESSION_ABSOLUTE_HOURS",
                default.session_absolute_ttl,
            ),
            verification_ttl: env_duration_hours(
                "PICLOUD_APP_USER_VERIFICATION_TTL_HOURS",
                default.verification_ttl,
            ),
            password_reset_ttl: env_duration_hours(
                "PICLOUD_APP_USER_PASSWORD_RESET_TTL_HOURS",
                default.password_reset_ttl,
            ),
            invitation_ttl: env_duration_days(
                "PICLOUD_APP_USER_INVITATION_TTL_DAYS",
                default.invitation_ttl,
            ),
        }
    }
}

fn env_duration_hours(var: &str, default: Duration) -> Duration {
    std::env::var(var)
        .ok()
        .and_then(|s| s.parse::<u64>().ok())
        .map_or(default, |h| Duration::from_secs(h * 3600))
}

fn env_duration_days(var: &str, default: Duration) -> Duration {
    std::env::var(var)
        .ok()
        .and_then(|s| s.parse::<u64>().ok())
        .map_or(default, |d| Duration::from_secs(d * 24 * 3600))
}

/// Postgres-backed `UsersService`.
/// In-memory rate limiter for outbound-email surfaces (verification +
/// password reset). Two scopes:
///
/// 1. **Per-(app, recipient)**: max `recipient_burst` calls per
///    `recipient_window` (default 5 per hour). Defends against an
///    attacker hammering one address.
/// 2. **Per-app daily cap**: max `app_daily_cap` calls per app per
///    rolling 24h window (default 200). Defends the SMTP-relay quota.
///
/// Both windows reset lazily — buckets that have aged out get rebuilt
/// on the next call. Stored in a single Mutex; reads + writes are
/// short, contention is low for the call volume this limits.
#[derive(Debug)]
struct EmailRateLimiter {
    recipient_burst: u32,
    recipient_window: Duration,
    app_daily_cap: u32,
    state: std::sync::Mutex<EmailRateState>,
}

#[derive(Debug, Default)]
struct EmailRateState {
    per_recipient: std::collections::HashMap<(AppId, String), Bucket>,
    per_app: std::collections::HashMap<AppId, Bucket>,
}

#[derive(Debug, Clone, Copy)]
struct Bucket {
    window_started_at: std::time::Instant,
    count: u32,
}

impl EmailRateLimiter {
    fn new() -> Self {
        Self {
            recipient_burst: 5,
            recipient_window: Duration::from_secs(3600),
            app_daily_cap: 200,
            state: std::sync::Mutex::new(EmailRateState::default()),
        }
    }

    /// Attempt to consume one slot for `(app, recipient)`. Returns
    /// `Err(())` if either cap is exceeded.
    fn try_consume(&self, app_id: AppId, recipient: &str) -> Result<(), ()> {
        let now = std::time::Instant::now();
        let app_cap_window = Duration::from_secs(24 * 3600);
        let mut state = self.state.lock().expect("email rate state poisoned");

        // Check + roll the per-app bucket.
        {
            let app_bucket = state.per_app.entry(app_id).or_insert(Bucket {
                window_started_at: now,
                count: 0,
            });
            if now.duration_since(app_bucket.window_started_at) >= app_cap_window {
                *app_bucket = Bucket {
                    window_started_at: now,
                    count: 0,
                };
            }
            if app_bucket.count >= self.app_daily_cap {
                return Err(());
            }
        }

        // Check + roll the per-recipient bucket.
        let key = (app_id, recipient.to_string());
        {
            let bucket = state.per_recipient.entry(key).or_insert(Bucket {
                window_started_at: now,
                count: 0,
            });
            if now.duration_since(bucket.window_started_at) >= self.recipient_window {
                *bucket = Bucket {
                    window_started_at: now,
                    count: 0,
                };
            }
            if bucket.count >= self.recipient_burst {
                return Err(());
            }
            bucket.count += 1;
        }
        // Both checks passed; bump the app counter.
        if let Some(app_bucket) = state.per_app.get_mut(&app_id) {
            app_bucket.count += 1;
        }
        Ok(())
    }
}

pub struct UsersServiceImpl {
    users: Arc<dyn AppUserRepository>,
    sessions: Arc<dyn AppUserSessionRepository>,
    verifications: Arc<dyn AppUserVerificationRepo>,
    password_resets: Arc<dyn AppUserPasswordResetRepo>,
    invitations: Arc<dyn AppUserInvitationRepo>,
    roles: Arc<dyn AppUserRoleRepo>,
    email: Arc<dyn EmailService>,
    authz: Arc<dyn AuthzRepo>,
    events: Arc<dyn ServiceEventEmitter>,
    config: UsersServiceConfig,
    email_rate_limiter: EmailRateLimiter,
}

impl UsersServiceImpl {
    #[must_use]
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        users: Arc<dyn AppUserRepository>,
        sessions: Arc<dyn AppUserSessionRepository>,
        verifications: Arc<dyn AppUserVerificationRepo>,
        password_resets: Arc<dyn AppUserPasswordResetRepo>,
        invitations: Arc<dyn AppUserInvitationRepo>,
        roles: Arc<dyn AppUserRoleRepo>,
        email: Arc<dyn EmailService>,
        authz: Arc<dyn AuthzRepo>,
        events: Arc<dyn ServiceEventEmitter>,
        config: UsersServiceConfig,
    ) -> Self {
        Self {
            users,
            sessions,
            verifications,
            password_resets,
            invitations,
            roles,
            email,
            authz,
            events,
            config,
            email_rate_limiter: EmailRateLimiter::new(),
        }
    }

    async fn require(
        &self,
        principal: Option<&Principal>,
        cap: Capability,
    ) -> Result<(), UsersError> {
        let Some(p) = principal else {
            // Public HTTP execution — script-as-gate; not denied here.
            return Ok(());
        };
        authz::require_mapped(
            &*self.authz,
            p,
            cap,
            || UsersError::Forbidden,
            UsersError::Backend,
        )
        .await
    }

    /// Compose the public `AppUser` shape from an `AppUserRow`, with the caller's
    /// already-resolved per-app `roles` (see [`Self::fetch_roles`]).
    fn to_app_user(row: AppUserRow, roles: Vec<String>) -> AppUser {
        AppUser {
            id: row.id,
            app_id: row.app_id,
            email: row.email,
            display_name: row.display_name,
            email_verified_at: row.email_verified_at,
            last_login_at: row.last_login_at,
            created_at: row.created_at,
            updated_at: row.updated_at,
            roles,
        }
    }

    async fn fetch_roles(
        &self,
        app_id: AppId,
        user_id: AppUserId,
    ) -> Result<Vec<String>, UsersError> {
        Ok(self.roles.list_for_user(app_id, user_id).await?)
    }

    async fn emit(&self, cx: &SdkCallCx, op: &'static str, user_id: AppUserId) {
        let _ = self
            .events
            .emit(
                cx,
                ServiceEvent {
                    source: "users",
                    op,
                    collection: None,
                    key: Some(user_id.to_string()),
                    payload: None,
                    old_payload: None,
                },
            )
            .await;
    }
}

fn validate_email(email: &str) -> Result<String, UsersError> {
    let trimmed = email.trim();
    if trimmed.is_empty() {
        return Err(UsersError::Invalid("email must not be empty".into()));
    }
    if trimmed.len() > 254 {
        return Err(UsersError::Invalid("email exceeds 254 characters".into()));
    }
    if !trimmed.contains('@') {
        return Err(UsersError::Invalid("email must contain '@'".into()));
    }
    Ok(trimmed.to_string())
}

fn validate_password(password: &str) -> Result<(), UsersError> {
    if password.chars().count() < 8 {
        return Err(UsersError::Invalid(
            "password must be at least 8 characters".into(),
        ));
    }
    if password.len() > 1024 {
        // Guard against absurd inputs that would make Argon2id costly.
        return Err(UsersError::Invalid("password exceeds 1024 bytes".into()));
    }
    Ok(())
}

fn validate_display_name(name: Option<&str>) -> Result<Option<String>, UsersError> {
    let Some(raw) = name else { return Ok(None) };
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Ok(None);
    }
    if trimmed.chars().count() > 256 {
        return Err(UsersError::Invalid(
            "display_name exceeds 256 characters".into(),
        ));
    }
    Ok(Some(trimmed.to_string()))
}

impl From<AppUserRepositoryError> for UsersError {
    fn from(e: AppUserRepositoryError) -> Self {
        match e {
            AppUserRepositoryError::NotFound(_) => UsersError::NotFound,
            AppUserRepositoryError::DuplicateEmail(_) => UsersError::DuplicateEmail,
            AppUserRepositoryError::Db(err) => UsersError::Backend(err.to_string()),
        }
    }
}

impl From<AppUserSessionRepositoryError> for UsersError {
    fn from(e: AppUserSessionRepositoryError) -> Self {
        match e {
            AppUserSessionRepositoryError::Db(err) => UsersError::Backend(err.to_string()),
        }
    }
}

impl From<AppUserVerificationRepoError> for UsersError {
    fn from(e: AppUserVerificationRepoError) -> Self {
        match e {
            AppUserVerificationRepoError::Db(err) => UsersError::Backend(err.to_string()),
        }
    }
}

impl From<AppUserPasswordResetRepoError> for UsersError {
    fn from(e: AppUserPasswordResetRepoError) -> Self {
        match e {
            AppUserPasswordResetRepoError::Db(err) => UsersError::Backend(err.to_string()),
        }
    }
}

impl From<AppUserInvitationRepoError> for UsersError {
    fn from(e: AppUserInvitationRepoError) -> Self {
        match e {
            AppUserInvitationRepoError::Db(err) => UsersError::Backend(err.to_string()),
        }
    }
}

impl From<AppUserRoleRepoError> for UsersError {
    fn from(e: AppUserRoleRepoError) -> Self {
        match e {
            AppUserRoleRepoError::Db(err) => UsersError::Backend(err.to_string()),
        }
    }
}

fn validate_role(role: &str) -> Result<String, UsersError> {
    let trimmed = role.trim();
    if trimmed.is_empty() {
        return Err(UsersError::Invalid("role must not be empty".into()));
    }
    if trimmed.chars().count() > 128 {
        return Err(UsersError::Invalid("role exceeds 128 characters".into()));
    }
    Ok(trimmed.to_string())
}

fn map_email_error(e: EmailError) -> UsersError {
    match e {
        EmailError::NotConfigured => UsersError::EmailNotConfigured,
        EmailError::Forbidden => UsersError::Forbidden,
        EmailError::MissingField(_)
        | EmailError::InvalidAddress(_)
        | EmailError::TooLarge { .. }
        | EmailError::TooManyRecipients { .. }
        | EmailError::RateLimited(_) => UsersError::EmailTransport(e.to_string()),
        EmailError::Transport(s) => UsersError::EmailTransport(s),
    }
}

/// Build the token-bearing URL: `link_base?token=<raw>` (or `&token=...`
/// if the base already has a query string). The raw token never lives
/// in storage; it appears only here, in the email body, and in the
/// user's inbox.
fn build_link(link_base: &str, raw_token: &str) -> String {
    let sep = if link_base.contains('?') { '&' } else { '?' };
    format!("{link_base}{sep}token={raw_token}")
}

/// Synthesize an `SdkCallCx` with `principal: None` for the internal
/// email send. The users::* surface has already done its own authz
/// check (AppUsersWrite / AppUsersAdmin); the underlying email send is
/// a system-level action, not the user's direct invocation. Mirrors
/// how downstream services would internally hop without re-running
/// the caller's authz.
fn internal_cx(cx: &SdkCallCx) -> SdkCallCx {
    SdkCallCx {
        app_id: cx.app_id,
        principal: None,
        script_id: cx.script_id,
        execution_id: cx.execution_id,
        request_id: cx.request_id,
        trigger_depth: cx.trigger_depth,
        root_execution_id: cx.root_execution_id,
        is_dead_letter_handler: cx.is_dead_letter_handler,
        event: cx.event.clone(),
    }
}

#[async_trait]
impl UsersService for UsersServiceImpl {
    async fn create(&self, cx: &SdkCallCx, input: CreateUserInput) -> Result<AppUser, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        let email = validate_email(&input.email)?;
        validate_password(&input.password)?;
        let display_name = validate_display_name(input.display_name.as_deref())?;
        let password_hash = hash_password(&input.password)
            .map_err(|e| UsersError::Backend(format!("password hashing failed: {e}")))?;
        let row = self
            .users
            .create(cx.app_id, &email, &password_hash, display_name.as_deref())
            .await?;
        let user_id = row.id;
        let user = Self::to_app_user(row, Vec::new());
        self.emit(cx, "create", user_id).await;
        Ok(user)
    }

    async fn get(&self, cx: &SdkCallCx, id: AppUserId) -> Result<Option<AppUser>, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
            .await?;
        let Some(row) = self.users.get(cx.app_id, id).await? else {
            return Ok(None);
        };
        let roles = self.fetch_roles(cx.app_id, row.id).await?;
        Ok(Some(Self::to_app_user(row, roles)))
    }

    async fn find_by_email(
        &self,
        cx: &SdkCallCx,
        email: &str,
    ) -> Result<Option<AppUser>, UsersError> {
        // F-S-003: an anonymous public-HTTP script could iterate emails
        // via find_by_email to enumerate registered users (the `require`
        // helper short-circuits on principal == None). Tighten: require
        // an authenticated principal — scripts that need a "does this
        // email exist?" probe must wrap it in their own auth gate.
        if cx.principal.is_none() {
            return Err(UsersError::Forbidden);
        }
        self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
            .await?;
        let normalized = validate_email(email)?;
        let Some(row) = self.users.find_by_email(cx.app_id, &normalized).await? else {
            return Ok(None);
        };
        let roles = self.fetch_roles(cx.app_id, row.id).await?;
        Ok(Some(Self::to_app_user(row, roles)))
    }

    async fn email_available(&self, cx: &SdkCallCx, email: &str) -> Result<bool, UsersError> {
        // Gated like `create` (the registration write path) so the probe
        // is available exactly where self-serve registration is — and,
        // crucially, WITHOUT find_by_email's anonymous-principal rejection.
        //
        // SECURITY (F-S-003, enumeration): a single probe reveals only
        // "exists / doesn't" — the same bit a register form leaks via its
        // "email already taken" path. But unlike `create`, this probe has
        // no Argon2 cost and no side effect (a `create` miss writes a junk
        // row), so it is a *cheaper, quieter* enumeration oracle when a
        // script exposes it on an anonymous public route. It is NOT rate
        // limited here, and there is no built-in per-route throttle/CAPTCHA
        // primitive. Scripts that put this behind a public endpoint and
        // care about enumeration must add their own `kv`-based counter.
        // See docs/stdlib-reference.md.
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        let normalized = validate_email(email)?;
        Ok(self
            .users
            .find_by_email(cx.app_id, &normalized)
            .await?
            .is_none())
    }

    async fn update(
        &self,
        cx: &SdkCallCx,
        id: AppUserId,
        patch: UpdateUserInput,
    ) -> Result<AppUser, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        let mut row_opt = self.users.get(cx.app_id, id).await?;
        if let Some(display) = patch.display_name {
            let normalized = match display {
                Some(s) => validate_display_name(Some(s.as_str()))?,
                None => None,
            };
            row_opt = Some(
                self.users
                    .update_display_name(cx.app_id, id, normalized.as_deref())
                    .await?,
            );
        }
        let row = row_opt.ok_or(UsersError::NotFound)?;
        let roles = self.fetch_roles(cx.app_id, row.id).await?;
        let user = Self::to_app_user(row, roles);
        self.emit(cx, "update", user.id).await;
        Ok(user)
    }

    async fn delete(&self, cx: &SdkCallCx, id: AppUserId) -> Result<bool, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        let deleted = self.users.delete(cx.app_id, id).await?;
        if deleted {
            self.emit(cx, "delete", id).await;
        }
        Ok(deleted)
    }

    async fn list(&self, cx: &SdkCallCx, opts: UsersListOpts) -> Result<UsersListPage, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
            .await?;
        let limit = opts.limit.unwrap_or(50).clamp(1, 500);
        let page = self
            .users
            .list(
                cx.app_id,
                RepoListOpts {
                    cursor: opts
                        .cursor
                        .as_deref()
                        .and_then(crate::app_user_repo::ListCursor::decode),
                    limit,
                },
            )
            .await?;

        // F-P-001: one batched roles lookup instead of N per-user
        // queries. With limit=500 this is 1 query vs 501.
        let user_ids: Vec<AppUserId> = page.items.iter().map(|r| r.id).collect();
        let mut roles_by_user = self
            .roles
            .list_for_users(cx.app_id, &user_ids)
            .await
            .map_err(|e| UsersError::Backend(e.to_string()))?;
        let items: Vec<AppUser> = page
            .items
            .into_iter()
            .map(|row| {
                let roles = roles_by_user.remove(&row.id).unwrap_or_default();
                Self::to_app_user(row, roles)
            })
            .collect();
        Ok(UsersListPage {
            items,
            next_cursor: page
                .next_cursor
                .as_ref()
                .map(crate::app_user_repo::ListCursor::encode),
        })
    }

    async fn login(
        &self,
        cx: &SdkCallCx,
        email: &str,
        password: &str,
    ) -> Result<Option<GeneratedSession>, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        // Both branches normalize the email so an attacker can't probe
        // existence via "" / "x" / very-long-string differences.
        let password_owned = password.to_string();
        let Ok(normalized) = validate_email(email) else {
            // Still run a dummy verify so the timing of an invalid
            // email shape matches the timing of a legitimate miss.
            // F-P-002: off the async worker — Argon2 is CPU-bound.
            let _ = tokio::task::spawn_blocking(move || {
                verify_password(TIMING_FLAT_DUMMY_HASH, &password_owned)
            })
            .await;
            return Ok(None);
        };
        let creds = self
            .users
            .get_credentials_by_email(cx.app_id, &normalized)
            .await?;
        let (hash, user_id, app_id) = match creds {
            Some(c) => (c.password_hash, Some(c.id), Some(c.app_id)),
            None => (TIMING_FLAT_DUMMY_HASH.to_string(), None, None),
        };
        // F-P-002: Argon2id verify off the async worker.
        let password_ok =
            tokio::task::spawn_blocking(move || verify_password(&hash, &password_owned))
                .await
                .map_err(|e| UsersError::Backend(format!("verify spawn_blocking join: {e}")))?;
        let (Some(user_id), Some(app_id)) = (user_id, app_id) else {
            return Ok(None);
        };
        if !password_ok {
            return Ok(None);
        }
        // Cross-app isolation: a credentials row for a different app
        // shouldn't be reachable by find/get_credentials with a fixed
        // app_id, but defense-in-depth here.
        if app_id != cx.app_id {
            return Ok(None);
        }

        let now = Utc::now();
        let absolute = now
            + chrono::Duration::from_std(self.config.session_absolute_ttl)
                .map_err(|e| UsersError::Backend(format!("absolute ttl: {e}")))?;
        let sliding = now
            + chrono::Duration::from_std(self.config.session_ttl)
                .map_err(|e| UsersError::Backend(format!("sliding ttl: {e}")))?;
        let expires_at = sliding.min(absolute);

        let token = generate_session_token();
        self.sessions
            .create(cx.app_id, user_id, &token.hash, expires_at, absolute)
            .await?;
        self.users.touch_last_login(cx.app_id, user_id).await?;

        let row = self
            .users
            .get(cx.app_id, user_id)
            .await?
            .ok_or(UsersError::NotFound)?;
        let roles = self.fetch_roles(cx.app_id, row.id).await?;
        let user = Self::to_app_user(row, roles);
        Ok(Some(GeneratedSession {
            token: token.raw,
            user,
            expires_at,
        }))
    }

    async fn verify(&self, cx: &SdkCallCx, token: &str) -> Result<Option<AppUser>, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
            .await?;
        self.resolve_session(cx.app_id, token).await
    }

    async fn logout(&self, cx: &SdkCallCx, token: &str) -> Result<(), UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        let token_hash = hash_token(token);
        // Only revoke when the row belongs to this app — otherwise the
        // script could prove existence of cross-app sessions by trying
        // arbitrary tokens. Lookup is the cross-app filter here.
        if let Some(lookup) = self.sessions.lookup(&token_hash).await? {
            if lookup.app_id != cx.app_id {
                return Ok(());
            }
            self.sessions.revoke(&token_hash).await?;
        }
        Ok(())
    }

    async fn verify_session_for_realtime(
        &self,
        app_id: AppId,
        token: &str,
    ) -> Result<Option<AppUser>, UsersError> {
        self.resolve_session(app_id, token).await
    }

    async fn send_verification_email(
        &self,
        cx: &SdkCallCx,
        user_id: AppUserId,
        opts: EmailTemplateOpts,
    ) -> Result<(), UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        let user = self
            .users
            .get(cx.app_id, user_id)
            .await?
            .ok_or(UsersError::NotFound)?;
        // F-S-002: token-bucket rate limit keyed on (app_id, recipient).
        // Anonymous public-route callers (cx.principal == None) skip the
        // authz gate above, so without this an attacker could burn the
        // SMTP-relay quota by spamming any public endpoint that calls
        // users::send_verification_email.
        if self
            .email_rate_limiter
            .try_consume(cx.app_id, &user.email)
            .is_err()
        {
            tracing::warn!(
                app_id = %cx.app_id,
                user_id = %user.id,
                "send_verification_email rate-limited"
            );
            return Err(UsersError::EmailRateLimited);
        }

        // Mint a single-use token. The hash hits the verifications
        // table; the raw token only ever appears in the email body.
        let token = generate_session_token();
        let expires_at = Utc::now()
            + chrono::Duration::from_std(self.config.verification_ttl)
                .map_err(|e| UsersError::Backend(format!("verification ttl: {e}")))?;
        self.verifications
            .create(cx.app_id, user.id, &token.hash, expires_at)
            .await?;

        let link = build_link(&opts.link_base, &token.raw);
        let body = opts.body_template.replace("{link}", &link);
        let outbound = OutboundEmail {
            to: vec![user.email.clone()],
            from: opts.from,
            subject: opts.subject,
            text: Some(body),
            ..Default::default()
        };
        let send_cx = internal_cx(cx);
        self.email
            .send(&send_cx, outbound)
            .await
            .map_err(map_email_error)?;
        Ok(())
    }

    async fn verify_email(
        &self,
        cx: &SdkCallCx,
        token: &str,
    ) -> Result<Option<AppUser>, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        let token_hash = hash_token(token);
        let Some(user_id) = self.verifications.consume(cx.app_id, &token_hash).await? else {
            return Ok(None);
        };
        let row = self.users.mark_email_verified(cx.app_id, user_id).await?;
        let roles = self.fetch_roles(cx.app_id, row.id).await?;
        let user = Self::to_app_user(row, roles);
        self.emit(cx, "email_verified", user.id).await;
        Ok(Some(user))
    }
    async fn request_password_reset(
        &self,
        cx: &SdkCallCx,
        email: &str,
        opts: EmailTemplateOpts,
    ) -> Result<(), UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        // F-S-004: no existence leak. The wall-clock delta between
        // "user not found" (handler returns immediately) and "user
        // found" (Argon2 token generate + DB INSERT + SMTP RTT) was
        // tens-of-ms — externally observable to any attacker who could
        // measure HTTP latency. Eliminate that observable by running
        // the same expensive work shape on the no-match branch:
        // generate a dummy token, do NOT INSERT, and skip the send.
        // (Skipping the INSERT keeps us honest about not leaking via
        // disk-space side channels either.)
        let Ok(normalized) = validate_email(email) else {
            // Bad email shape — still spend the dummy budget.
            let _ = generate_session_token();
            return Ok(());
        };
        let Some(user) = self.users.find_by_email(cx.app_id, &normalized).await? else {
            // No user — same dummy work as the matching branch.
            let _ = generate_session_token();
            return Ok(());
        };

        // F-S-002: rate limit per (app, recipient). Return Ok silently
        // on exhaustion so an attacker can't probe whether an email
        // exists by comparing the rate-limit response to the no-match
        // response.
        if self
            .email_rate_limiter
            .try_consume(cx.app_id, &user.email)
            .is_err()
        {
            tracing::warn!(
                app_id = %cx.app_id,
                user_id = %user.id,
                "request_password_reset rate-limited; suppressing to avoid existence leak"
            );
            return Ok(());
        }

        let token = generate_session_token();
        let expires_at = Utc::now()
            + chrono::Duration::from_std(self.config.password_reset_ttl)
                .map_err(|e| UsersError::Backend(format!("reset ttl: {e}")))?;
        self.password_resets
            .create(cx.app_id, user.id, &token.hash, expires_at)
            .await?;

        let link = build_link(&opts.link_base, &token.raw);
        let body = opts.body_template.replace("{link}", &link);
        let outbound = OutboundEmail {
            to: vec![user.email.clone()],
            from: opts.from,
            subject: opts.subject,
            text: Some(body),
            ..Default::default()
        };
        let send_cx = internal_cx(cx);
        // Email-not-configured isn't a security signal — it's the
        // operator's known state. Surface it to script-land so the
        // script can fall back to a sync reset flow. But: never
        // surface email send failures (TooLarge/InvalidAddress/
        // Transport) as anything other than a generic "email failed"
        // — those still could be used to probe whether the user
        // existed; clamp the visible variants.
        match self.email.send(&send_cx, outbound).await {
            Ok(()) => Ok(()),
            Err(EmailError::NotConfigured) => Err(UsersError::EmailNotConfigured),
            Err(e) => {
                tracing::warn!(
                    error = %e,
                    user_id = %user.id,
                    app_id = %cx.app_id,
                    "request_password_reset: email send failed; suppressing to avoid existence leak"
                );
                Ok(())
            }
        }
    }

    async fn complete_password_reset(
        &self,
        cx: &SdkCallCx,
        token: &str,
        new_password: &str,
    ) -> Result<Option<AppUser>, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
            .await?;
        validate_password(new_password)?;
        let token_hash = hash_token(token);
        let Some(user_id) = self.password_resets.consume(cx.app_id, &token_hash).await? else {
            return Ok(None);
        };
        let password_hash = hash_password(new_password)
            .map_err(|e| UsersError::Backend(format!("password hashing failed: {e}")))?;
        let row = self
            .users
            .update_password_hash(cx.app_id, user_id, &password_hash)
            .await?;
        // Revoke every active session for the user — anyone with a
        // stale token shouldn't be able to ride out the reset.
        let _ = self.sessions.revoke_for_user(cx.app_id, user_id).await?;
        let roles = self.fetch_roles(cx.app_id, row.id).await?;
        let user = Self::to_app_user(row, roles);
        self.emit(cx, "password_changed", user.id).await;
        Ok(Some(user))
    }
    async fn invite(
        &self,
        cx: &SdkCallCx,
        email: &str,
        opts: InviteOpts,
    ) -> Result<(), UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersAdmin(cx.app_id))
            .await?;
        let normalized = validate_email(email)?;
        // Roles are stored verbatim; commit 8 (per-app roles) is what
        // actually applies them on accept. v1.1.8 ships with empty
        // role lists or whatever the script supplies — the storage
        // round-trip is identical either way.
        let display_name = validate_display_name(opts.display_name.as_deref())?;
        let token = generate_session_token();
        let expires_at = Utc::now()
            + chrono::Duration::from_std(self.config.invitation_ttl)
                .map_err(|e| UsersError::Backend(format!("invitation ttl: {e}")))?;
        let row = self
            .invitations
            .create(
                cx.app_id,
                &normalized,
                display_name.as_deref(),
                &opts.roles,
                &token.hash,
                expires_at,
            )
            .await?;

        // Send the invitation email if a template was supplied.
        // Inviters can skip the email by omitting the template — useful
        // when the invitation flow is driven by an external delivery
        // (e.g., printed onboarding letter).
        if let Some(template) = opts.template {
            let link = build_link(&template.link_base, &token.raw);
            let body = template.body_template.replace("{link}", &link);
            let outbound = OutboundEmail {
                to: vec![normalized.clone()],
                from: template.from,
                subject: template.subject,
                text: Some(body),
                ..Default::default()
            };
            let send_cx = internal_cx(cx);
            if let Err(e) = self.email.send(&send_cx, outbound).await {
                if matches!(e, EmailError::NotConfigured) {
                    return Err(UsersError::EmailNotConfigured);
                }
                tracing::warn!(
                    error = %e,
                    invite_id = %row.id,
                    app_id = %cx.app_id,
                    "invite email send failed; invitation row remains valid"
                );
            }
        }
        Ok(())
    }

    async fn accept_invite(
        &self,
        cx: &SdkCallCx,
        token: &str,
        password: &str,
        display_name: Option<String>,
    ) -> Result<Option<GeneratedAccept>, UsersError> {
        // No require: the invitation token IS the authorization.
        validate_password(password)?;
        let token_hash = hash_token(token);
        let Some(consumed) = self
            .invitations
            .consume_by_token(cx.app_id, &token_hash)
            .await?
        else {
            return Ok(None);
        };

        // Display name on the new user: caller override beats the
        // invitation's pre-staged value beats None.
        let final_display = display_name
            .as_deref()
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .map(str::to_string)
            .or(consumed.display_name);

        let password_hash = hash_password(password)
            .map_err(|e| UsersError::Backend(format!("password hashing failed: {e}")))?;
        let row = match self
            .users
            .create(
                cx.app_id,
                &consumed.email,
                &password_hash,
                final_display.as_deref(),
            )
            .await
        {
            Ok(row) => row,
            Err(AppUserRepositoryError::DuplicateEmail(_)) => {
                // User already exists (sign-up beat acceptance). The
                // invitation is consumed — they'll need to log in
                // normally with their existing password.
                return Ok(None);
            }
            Err(e) => return Err(e.into()),
        };
        // Apply pre-staged roles (commit 8). Each role goes through
        // the same trim + length validation the script's add_role
        // call would; invalid ones are skipped with a warn rather
        // than aborting the whole accept (the invitation was a
        // promise the admin made; we honor as much of it as we can).
        let user_id = row.id;
        let mut applied: Vec<String> = Vec::with_capacity(consumed.roles.len());
        for raw_role in &consumed.roles {
            match validate_role(raw_role) {
                Ok(role) => {
                    self.roles.add(cx.app_id, user_id, &role).await?;
                    applied.push(role);
                }
                Err(e) => {
                    tracing::warn!(
                        invalid_role = %raw_role,
                        user_id = %user_id,
                        app_id = %cx.app_id,
                        error = %e,
                        "accept_invite: skipping malformed pre-staged role"
                    );
                }
            }
        }
        let user = Self::to_app_user(row, applied);
        let session = self.mint_session(cx.app_id, user_id, &user).await?;
        self.emit(cx, "create", user_id).await;
        Ok(Some(GeneratedAccept { user, session }))
    }
    async fn add_role(
        &self,
        cx: &SdkCallCx,
        user_id: AppUserId,
        role: &str,
    ) -> Result<(), UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersAdmin(cx.app_id))
            .await?;
        // Ensure the user actually exists in this app — otherwise an
        // FK-failing insert leaks "no such user" via the error shape.
        let exists = self.users.get(cx.app_id, user_id).await?.is_some();
        if !exists {
            return Err(UsersError::NotFound);
        }
        let role = validate_role(role)?;
        self.roles.add(cx.app_id, user_id, &role).await?;
        self.emit(cx, "role_added", user_id).await;
        Ok(())
    }
    async fn remove_role(
        &self,
        cx: &SdkCallCx,
        user_id: AppUserId,
        role: &str,
    ) -> Result<bool, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersAdmin(cx.app_id))
            .await?;
        let role = validate_role(role)?;
        let removed = self.roles.remove(cx.app_id, user_id, &role).await?;
        if removed {
            self.emit(cx, "role_removed", user_id).await;
        }
        Ok(removed)
    }
    async fn has_role(
        &self,
        cx: &SdkCallCx,
        user_id: AppUserId,
        role: &str,
    ) -> Result<bool, UsersError> {
        self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
            .await?;
        let role = validate_role(role)?;
        Ok(self.roles.has(cx.app_id, user_id, &role).await?)
    }
    async fn admin_reset_password_token(
        &self,
        principal: &Principal,
        app_id: AppId,
        user_id: AppUserId,
    ) -> Result<String, UsersError> {
        self.require(Some(principal), Capability::AppUsersAdmin(app_id))
            .await?;
        // Verify the user exists in this app — surfaces 404 cleanly
        // instead of failing on the FK insert.
        let exists = self.users.get(app_id, user_id).await?.is_some();
        if !exists {
            return Err(UsersError::NotFound);
        }
        let token = generate_session_token();
        let expires_at = Utc::now()
            + chrono::Duration::from_std(self.config.password_reset_ttl)
                .map_err(|e| UsersError::Backend(format!("reset ttl: {e}")))?;
        self.password_resets
            .create(app_id, user_id, &token.hash, expires_at)
            .await?;
        // No email send — the admin pastes the raw token into their
        // own out-of-band reset link. The script-side
        // users::request_password_reset is the email-bearing path.
        Ok(token.raw)
    }
    async fn admin_revoke_all_sessions(
        &self,
        principal: &Principal,
        app_id: AppId,
        user_id: AppUserId,
    ) -> Result<u64, UsersError> {
        self.require(Some(principal), Capability::AppUsersAdmin(app_id))
            .await?;
        Ok(self.sessions.revoke_for_user(app_id, user_id).await?)
    }
    async fn admin_create_invitation(
        &self,
        principal: &Principal,
        app_id: AppId,
        email: &str,
        opts: InviteOpts,
    ) -> Result<Invitation, UsersError> {
        self.require(Some(principal), Capability::AppUsersAdmin(app_id))
            .await?;
        let normalized = validate_email(email)?;
        let display_name = validate_display_name(opts.display_name.as_deref())?;
        let token = generate_session_token();
        let expires_at = Utc::now()
            + chrono::Duration::from_std(self.config.invitation_ttl)
                .map_err(|e| UsersError::Backend(format!("invitation ttl: {e}")))?;
        let row = self
            .invitations
            .create(
                app_id,
                &normalized,
                display_name.as_deref(),
                &opts.roles,
                &token.hash,
                expires_at,
            )
            .await?;

        if let Some(template) = opts.template {
            let link = build_link(&template.link_base, &token.raw);
            let body = template.body_template.replace("{link}", &link);
            let outbound = OutboundEmail {
                to: vec![normalized.clone()],
                from: template.from,
                subject: template.subject,
                text: Some(body),
                ..Default::default()
            };
            // Admin-issued: no SdkCallCx exists; synthesize a minimal
            // one with the admin's principal so the email service's
            // own authz pass is honored. (For users::invite the cx is
            // the script's; the internal_cx() helper there clears the
            // principal because the users::* gate already fired.
            // Here the admin's principal is the actual caller of the
            // email — keep the gate.)
            let send_cx = SdkCallCx {
                app_id,
                principal: Some(principal.clone()),
                script_id: picloud_shared::ScriptId::new(),
                execution_id: picloud_shared::ExecutionId::new(),
                request_id: picloud_shared::RequestId::new(),
                trigger_depth: 0,
                root_execution_id: picloud_shared::ExecutionId::new(),
                is_dead_letter_handler: false,
                event: None,
            };
            if let Err(e) = self.email.send(&send_cx, outbound).await {
                if matches!(e, EmailError::NotConfigured) {
                    return Err(UsersError::EmailNotConfigured);
                }
                tracing::warn!(
                    error = %e,
                    invite_id = %row.id,
                    app_id = %app_id,
                    "admin_create_invitation email failed; row remains valid"
                );
            }
        }

        Ok(Invitation {
            id: row.id,
            app_id: row.app_id,
            email: row.email,
            display_name: row.display_name,
            roles: row.roles,
            created_at: row.created_at,
            expires_at: row.expires_at,
        })
    }

    async fn list_invitations(
        &self,
        principal: &Principal,
        app_id: AppId,
    ) -> Result<Vec<Invitation>, UsersError> {
        self.require(Some(principal), Capability::AppUsersAdmin(app_id))
            .await?;
        let rows = self.invitations.list_pending(app_id).await?;
        Ok(rows
            .into_iter()
            .map(|r| Invitation {
                id: r.id,
                app_id: r.app_id,
                email: r.email,
                display_name: r.display_name,
                roles: r.roles,
                created_at: r.created_at,
                expires_at: r.expires_at,
            })
            .collect())
    }
    async fn revoke_invitation(
        &self,
        principal: &Principal,
        app_id: AppId,
        invite_id: InvitationId,
    ) -> Result<bool, UsersError> {
        self.require(Some(principal), Capability::AppUsersAdmin(app_id))
            .await?;
        Ok(self.invitations.revoke(app_id, invite_id).await?)
    }
}

impl UsersServiceImpl {
    /// Mint a session for an already-authenticated user. Used by
    /// `login` (after password verify) and `accept_invite` (after
    /// invitation consume). Returns the raw token to embed in the
    /// response; only the SHA-256 hash hits storage.
    async fn mint_session(
        &self,
        app_id: AppId,
        user_id: AppUserId,
        user: &AppUser,
    ) -> Result<GeneratedSession, UsersError> {
        let now = Utc::now();
        let absolute = now
            + chrono::Duration::from_std(self.config.session_absolute_ttl)
                .map_err(|e| UsersError::Backend(format!("absolute ttl: {e}")))?;
        let sliding = now
            + chrono::Duration::from_std(self.config.session_ttl)
                .map_err(|e| UsersError::Backend(format!("sliding ttl: {e}")))?;
        let expires_at = sliding.min(absolute);

        let token = generate_session_token();
        self.sessions
            .create(app_id, user_id, &token.hash, expires_at, absolute)
            .await?;
        self.users.touch_last_login(app_id, user_id).await?;
        Ok(GeneratedSession {
            token: token.raw,
            user: user.clone(),
            expires_at,
        })
    }

    /// Shared lookup logic for `verify` and `verify_session_for_realtime`.
    /// Honors the absolute cap and cross-app isolation, then bumps the
    /// sliding window (capped at the absolute cap) on success.
    async fn resolve_session(
        &self,
        app_id: AppId,
        token: &str,
    ) -> Result<Option<AppUser>, UsersError> {
        let token_hash = hash_token(token);
        let Some(lookup) = self.sessions.lookup(&token_hash).await? else {
            return Ok(None);
        };
        // Cross-app isolation: a token issued for app A cannot
        // authorize a request into app B.
        if lookup.app_id != app_id {
            return Ok(None);
        }

        let now = Utc::now();
        let sliding = now
            + chrono::Duration::from_std(self.config.session_ttl)
                .map_err(|e| UsersError::Backend(format!("sliding ttl: {e}")))?;
        let new_expires = sliding.min(lookup.absolute_expires_at);
        self.sessions.touch(&token_hash, new_expires).await?;

        let row = self
            .users
            .get(lookup.app_id, lookup.user_id)
            .await?
            .ok_or(UsersError::NotFound)?;
        let roles = self.fetch_roles(lookup.app_id, row.id).await?;
        Ok(Some(Self::to_app_user(row, roles)))
    }
}