"""The local directory: what the vendored kit calls `bw_auth`, plus the token mint and the
account lifecycle the CLI and the auth routes share.

Sources of truth: this module, `app/models/accounts.py`, `app/services/passwords.py`,
`app/services/mail.py`, `docs/ACCOUNTS.md`. Design: `.logs/planning/accounts-2026-09.md`
§4.4 and §4.5.

`LocalDirectory` is duck-typed to the kit's central client: `userinfo` raises for an absent
or disabled row (the kit reads the exception as "new person"); `invite_user` and
`create_user_with_password` insert the principal inside a savepoint and turn an
`IntegrityError` on username or email into `AccountsError('EXISTS')` without continuing on
an existing row of any status (the savepoint-and-re-read rule is for shared domain rows,
not principals); nothing here ever puts a link or a password in an HTTP response except
the kit's own one-time `password` field.

Tokens: `secrets.token_urlsafe(32)`, stored hashed, single use (marked used in the same
transaction as the password write), 48 hours, minted only for non-disabled accounts, and
carried only in the link's fragment (`<PUBLIC_BASE_URL>/welcome#<token>`) and then a JSON
body: a browser never sends a fragment, so the access log never sees it.
"""

from __future__ import annotations

import logging
import secrets
from dataclasses import dataclass
from datetime import datetime, timedelta

from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError

from app import db as appdb
from app.config import settings
from app.models import Account, AccountCredential, AccountMember, AccountToken
from app.services import audit_log, mail, passwords, sessions
from app.services.accounts import valid_username
from app.services.sessions import aware, hash_token
from app.vendor.bw_accounts import AccountsError

log = logging.getLogger(__name__)

TOKEN_SECONDS = 48 * 3600
TOKEN_KEEP_DAYS = 7
LOCK_FAILURES = 8
LOCK_WINDOW = timedelta(minutes=15)
LOCK_FOR = timedelta(minutes=15)


class AccountLookupError(LookupError):
    """No usable account under that name (absent, or disabled)."""


class DisabledError(AccountsError):
    def __init__(self, username: str) -> None:
        super().__init__(f"'{username}' is disabled; enable it first.", "DISABLED")


# --- lookups --------------------------------------------------------------------------------

def find(db, identifier: str) -> Account | None:
    """By username or email, both folded."""
    key = (identifier or "").strip().lower()
    if not key:
        return None
    if "@" in key:
        return db.scalar(select(Account).where(func.lower(Account.email) == key))
    return db.scalar(select(Account).where(Account.username == key))


def by_username(db, username: str) -> Account | None:
    return db.scalar(select(Account).where(Account.username == (username or "").strip().lower()))


def _credential(db, account: Account) -> AccountCredential:
    cred = db.get(AccountCredential, account.id)
    if cred is None:
        cred = AccountCredential(account_id=account.id)
        db.add(cred)
        db.flush()
    return cred


# --- creating principals --------------------------------------------------------------------

def _split_display(first: str, last: str, username: str) -> str:
    return (f"{(first or '').strip()} {(last or '').strip()}").strip() or username


def create_account(db, username: str, email: str | None, display_name: str, *,
                   status: str = "invited", invited_by_id: int | None = None) -> Account:
    """Insert the principal, or raise `AccountsError('EXISTS')` and write nothing. Inside a
    savepoint so the caller's transaction survives the refusal."""
    username = (username or "").strip().lower()
    if not valid_username(username):
        raise AccountsError("A username is 2 to 32 characters: lowercase letters, digits, '.', '_' or '-'.", "BAD_INPUT")
    email = (email or "").strip().lower() or None
    account = Account(username=username, email=email, display_name=display_name[:80],
                      status=status, invited_by_id=invited_by_id)
    try:
        with db.begin_nested():
            db.add(account)
            db.flush()
    except IntegrityError:
        raise AccountsError(f"'{username}' or that email already belongs to an account.", "EXISTS") from None
    return account


# --- tokens ---------------------------------------------------------------------------------

def mint_token(db, account: Account, mode: str, *, created_by_id: int | None = None,
               now: datetime | None = None) -> str:
    if account.status == "disabled":
        raise DisabledError(account.username or "")
    now = now or sessions.utcnow()
    token = secrets.token_urlsafe(32)
    db.add(AccountToken(token_hash=hash_token(token), account_id=account.id, mode=mode,
                        created_at=now, expires_at=now + timedelta(seconds=TOKEN_SECONDS),
                        created_by_id=created_by_id))
    db.flush()
    audit_log.record("token.mint", entity_type="account", entity_key=account.username,
                     detail={"mode": mode}, account_id=created_by_id)
    return token


def link_for(mode: str, token: str) -> str:
    """The one-time link. The token is the fragment, never a path or query segment."""
    page = "welcome" if mode == "invite" else "reset"
    base = (settings.public_base_url or "").rstrip("/")
    return f"{base}/{page}#{token}"


@dataclass
class TokenState:
    state: str  # valid | expired_resendable | used | unknown
    row: AccountToken | None = None
    account: Account | None = None


def token_state(db, token: str, *, now: datetime | None = None) -> TokenState:
    now = now or sessions.utcnow()
    if not token or len(token) > 200:
        return TokenState("unknown")
    row = db.get(AccountToken, hash_token(token))
    if row is None:
        return TokenState("unknown")
    if row.used_at is not None:
        return TokenState("used", row)
    if aware(row.expires_at) <= now:
        return TokenState("expired_resendable", row)
    account = db.get(Account, row.account_id)
    if account is None or account.status == "disabled":
        return TokenState("unknown")
    return TokenState("valid", row, account)


def redeem(db, token: str, password: str, *, mode: str, now: datetime | None = None) -> Account:
    """Set the password from a valid token of the given mode and mark the token used in the
    same transaction; every other session of the account is revoked. Raises `LookupError`
    (bad token) or `passwords.PasswordRuleError`."""
    now = now or sessions.utcnow()
    state = token_state(db, token, now=now)
    if state.state != "valid" or state.row.mode != mode:
        raise LookupError(state.state)
    account = state.account
    value = passwords.check_rules(password, username=account.username or "", email=account.email or "")
    cred = _credential(db, account)
    cred.password_hash = passwords.hash_password(value)
    cred.password_set_at = now
    cred.must_change_password = False
    cred.failed_logins = 0
    cred.failed_window_started_at = None
    cred.locked_until = None
    state.row.used_at = now
    if account.status == "invited":
        account.status = "active"
    db.commit()
    sessions.revoke_account(account.id, reason="password", now=now)
    audit_log.record("token.redeem", entity_type="account", entity_key=account.username,
                     detail={"mode": mode}, account_id=account.id)
    audit_log.record("password.set", entity_type="account", entity_key=account.username,
                     detail={"via": "welcome" if mode == "invite" else "reset"}, account_id=account.id)
    return account


def prune_tokens(db, *, now: datetime | None = None) -> int:
    """Used rows, and expired-but-unused rows older than TOKEN_KEEP_DAYS (kept that long so
    the welcome page can say a link has expired rather than that it never existed)."""
    from sqlalchemy import delete

    now = now or sessions.utcnow()
    result = db.execute(delete(AccountToken).where(or_(
        AccountToken.used_at.isnot(None),
        AccountToken.expires_at <= now - timedelta(days=TOKEN_KEEP_DAYS),
    )))
    db.commit()
    return result.rowcount or 0


def tokens_minted_since(db, account_id: int, mode: str, since: datetime) -> int:
    return db.scalar(select(func.count()).select_from(AccountToken).where(
        AccountToken.account_id == account_id, AccountToken.mode == mode, AccountToken.created_at >= since
    )) or 0


# --- passwords and lockout ------------------------------------------------------------------

def set_password(db, account: Account, password: str, *, via: str, must_change: bool = False,
                 by_id: int | None = None, keep_session_hash: str | None = None,
                 now: datetime | None = None) -> None:
    """Replace the credential (rules applied), clear the lock, sign the account out everywhere
    but the session that asked."""
    now = now or sessions.utcnow()
    value = passwords.check_rules(password, username=account.username or "", email=account.email or "")
    cred = _credential(db, account)
    cred.password_hash = passwords.hash_password(value)
    cred.password_set_at = now
    cred.must_change_password = must_change
    cred.failed_logins = 0
    cred.failed_window_started_at = None
    cred.locked_until = None
    if account.status == "invited":
        account.status = "active"
    db.commit()
    sessions.revoke_account(account.id, reason="password", except_hash=keep_session_hash, by_id=by_id, now=now)
    audit_log.record("password.set", entity_type="account", entity_key=account.username,
                     detail={"via": via, "must_change": must_change}, account_id=by_id or account.id)


def note_failure(db, cred: AccountCredential, *, now: datetime) -> bool:
    """Count a wrong password; True when this one locked the account (8 within 15 minutes)."""
    started = aware(cred.failed_window_started_at)
    if started is None or now - started > LOCK_WINDOW:
        cred.failed_window_started_at = now
        cred.failed_logins = 1
    else:
        cred.failed_logins += 1
    locked = False
    if cred.failed_logins >= LOCK_FAILURES:
        cred.locked_until = now + LOCK_FOR
        cred.failed_logins = 0
        cred.failed_window_started_at = None
        locked = True
    db.commit()
    return locked


def note_success(db, account: Account, cred: AccountCredential, *, now: datetime) -> None:
    cred.failed_logins = 0
    cred.failed_window_started_at = None
    cred.locked_until = None
    account.last_login_at = now
    db.commit()


def is_locked(cred: AccountCredential, *, now: datetime) -> bool:
    until = aware(cred.locked_until)
    return until is not None and until > now


def unlock(db, account: Account) -> None:
    cred = _credential(db, account)
    cred.failed_logins = 0
    cred.failed_window_started_at = None
    cred.locked_until = None
    db.commit()
    audit_log.record("account.unlock", entity_type="account", entity_key=account.username)


# --- lifecycle ------------------------------------------------------------------------------

def disable_account(username: str, *, via: str, reason: str = "disabled", by_id: int | None = None,
                    now: datetime | None = None) -> bool:
    """Status, disabled_at, every session revoked; `account_members` untouched, so to the kit a
    disabled member is still a member and its invite path refuses with EXISTS. Own connection:
    the kit's audit hook calls this from inside its own store call."""
    now = now or sessions.utcnow()
    with appdb.SessionLocal() as db:
        account = by_username(db, username)
        if account is None:
            return False
        if account.status != "disabled":
            account.status = "disabled"
            account.disabled_at = now
            db.commit()
        account_id, uname = account.id, account.username
    sessions.revoke_account(account_id, reason=reason, by_id=by_id, now=now)
    audit_log.record("account.disable", entity_type="account", entity_key=uname,
                     detail={"via": via}, account_id=by_id)
    return True


def enable_account(username: str, *, by_id: int | None = None) -> bool:
    """Back to active; never re-adds a member row (the owner does that in the People panel)."""
    with appdb.SessionLocal() as db:
        account = by_username(db, username)
        if account is None:
            return False
        cred = db.get(AccountCredential, account.id)
        account.status = "active" if cred is not None and cred.password_hash else "invited"
        account.disabled_at = None
        db.commit()
        uname = account.username
    audit_log.record("account.enable", entity_type="account", entity_key=uname, account_id=by_id)
    return True


def add_member_row(db, username: str, level: str) -> None:
    """The CLI's direct membership write (the owner acting outside a request)."""
    existing = db.get(AccountMember, username)
    if existing is None:
        db.add(AccountMember(username=username, level=level, active=True))
    else:
        existing.level, existing.active = level, True
    db.flush()


# --- the kit's central client, locally --------------------------------------------------------

class LocalDirectory:
    """Duck-typed to the kit's `bw_auth` module. Every method opens its own short session."""

    def userinfo(self, username: str) -> dict:
        """Raises for an absent row (the kit reads that as "new person"). A disabled row
        resolves, with its status: the kit then takes the existing-account path and the
        store's `add_member` refuses it with DISABLED, which names the real reason; raising
        here instead would send the kit down the new-account path and answer EMAIL_REQUIRED
        or EXISTS for an account that merely needs `accounts enable`."""
        with appdb.SessionLocal() as db:
            account = by_username(db, username)
            if account is None:
                raise AccountLookupError(username)
            first, _, last = account.display_name.partition(" ")
            return {"username": account.username, "email": account.email or "", "first": first,
                    "last": last, "status": account.status}

    def invite_user(self, username: str, email: str, first: str = "", last: str = "") -> dict:
        """A new principal plus an emailed welcome link. Refuses before writing when no mail
        provider exists; refuses with EXISTS on any existing row."""
        sender = mail.mailer()
        if not mail.configured():
            sender.send("", "", "")  # raises MailNotConfigured
        email = (email or "").strip().lower()
        if not email:
            raise AccountsError("A new account needs an email.", "BAD_INPUT")
        actor = sessions.current.get()
        with appdb.SessionLocal() as db:
            account = create_account(db, username, email, _split_display(first, last, username),
                                     status="invited", invited_by_id=actor.account_id if actor else None)
            token = mint_token(db, account, "invite", created_by_id=actor.account_id if actor else None)
            db.commit()
            uname, to = account.username, account.email
        try:
            mail.send_capped(mail.bucket("invite", "host"), to, "Your Duty Free Professor account",
                             f"Set your password to sign in: {link_for('invite', token)}\n"
                             f"The link works for 48 hours. Your username is {uname}.")
        except mail.MailCapReached as exc:
            raise AccountsError(str(exc), "MAIL_CAP") from None
        audit_log.record("account.create", entity_type="account", entity_key=uname,
                         detail={"via": "invite"})
        return {"success": True, "created": True, "user": {"username": uname}, "emailed_to": to}

    def create_user_with_password(self, username: str, email: str, first: str = "", last: str = "") -> dict:
        """The manual handoff: a new principal with a generated password returned once and
        never stored; `must_change_password` set. New accounts only: an existing username of
        any status gets no password and no token."""
        email = (email or "").strip().lower()
        if not email:
            raise AccountsError("A new account needs an email.", "BAD_INPUT")
        actor = sessions.current.get()
        password = passwords.generate_password()
        with appdb.SessionLocal() as db:
            account = create_account(db, username, email, _split_display(first, last, username),
                                     status="active", invited_by_id=actor.account_id if actor else None)
            cred = _credential(db, account)
            cred.password_hash = passwords.hash_password(password)
            cred.password_set_at = sessions.utcnow()
            cred.must_change_password = True
            db.commit()
            uname = account.username
        audit_log.record("account.create", entity_type="account", entity_key=uname,
                         detail={"via": "password"})
        audit_log.record("password.set", entity_type="account", entity_key=uname,
                         detail={"via": "generated", "must_change": True})
        return {"success": True, "created": True, "delivery": "password", "user": uname,
                "class": "external", "password": password}

    def reset_password_generated(self, username: str) -> dict:
        """A fresh generated password for an existing, non-disabled account; every session
        revoked; returned once."""
        actor = sessions.current.get()
        password = passwords.generate_password()
        with appdb.SessionLocal() as db:
            account = by_username(db, username)
            if account is None or account.status == "disabled":
                raise AccountLookupError(username)
            set_password(db, account, password, via="generated", must_change=True,
                         by_id=actor.account_id if actor else None)
            uname = account.username
        return {"success": True, "user": uname, "password": password}

    def send_reset(self, username: str) -> dict:
        actor = sessions.current.get()
        sender = mail.mailer()
        if not mail.configured():
            sender.send("", "", "")
        with appdb.SessionLocal() as db:
            account = by_username(db, username)
            if account is None or account.status == "disabled" or not account.email:
                raise AccountLookupError(username)
            token = mint_token(db, account, "reset", created_by_id=actor.account_id if actor else None)
            db.commit()
            to, uname, account_id = account.email, account.username, account.id
        try:
            mail.send_capped(mail.bucket("auth", account_id), to, "Reset your Duty Free Professor password",
                             f"Choose a new password here: {link_for('reset', token)}\n"
                             f"The link works for 48 hours. If you did not ask for this, ignore it.")
        except mail.MailCapReached as exc:
            raise AccountsError(str(exc), "MAIL_CAP") from None
        return {"success": True, "emailed_to": to}

    def notify_added(self, username: str) -> dict:
        with appdb.SessionLocal() as db:
            account = by_username(db, username)
            if account is None or account.status == "disabled" or not account.email:
                raise AccountLookupError(username)
            to, account_id = account.email, account.id
        try:
            mail.send_capped(mail.bucket("auth", account_id), to, "You have been given access to Duty Free Professor",
                             f"Sign in at {(settings.public_base_url or '').rstrip('/')}/login with your usual password.")
        except mail.MailCapReached as exc:
            raise AccountsError(str(exc), "MAIL_CAP") from None
        return {"success": True, "emailed_to": to}

    def search_users(self, q: str, limit: int = 8) -> list[dict]:
        """Prefix match on username and display name over active accounts."""
        key = (q or "").strip().lower()
        with appdb.SessionLocal() as db:
            stmt = select(Account).where(Account.status == "active", Account.username.isnot(None))
            if key:
                stmt = stmt.where(or_(Account.username.like(key + "%"),
                                      func.lower(Account.display_name).like(key + "%")))
            rows = db.scalars(stmt.order_by(Account.username).limit(limit)).all()
            out = []
            for a in rows:
                first, _, last = a.display_name.partition(" ")
                out.append({"username": a.username, "email": a.email or "", "first": first,
                            "last": last, "class": "external"})
        return out

    def user_access(self, username: str) -> list:
        return []

    def report_access(self, username: str, level, instances=None) -> None:
        return None

    def report_instances(self, instances) -> None:
        return None

    def report_impersonation(self, real: str, target: str, mode: str = "readonly", active: bool = True) -> None:
        """The audit row for an explicit View As start or stop."""
        with appdb.SessionLocal() as db:
            real_id = db.scalar(select(Account.id).where(Account.username == real))
            target_id = db.scalar(select(Account.id).where(Account.username == target))
        audit_log.record("view_as.start" if active else "view_as.stop", entity_type="account",
                         entity_key=target, detail={"real": real, "mode": mode},
                         account_id=real_id, acting_as_id=target_id)