"""RBS REGAL — Advanced Transaction Prefix & Invoice Number Management.

Per-company, per-type prefix series with atomic counter increment to guarantee
no duplicates under concurrent billing.

Endpoints (mounted under /api/txn-prefixes):
    GET    /catalog                  — supported transaction types + tokens
    GET    /                         — list all prefixes (filtered by company_id)
    POST   /                         — create a new prefix (admin)
    PUT    /{id}                     — edit a prefix
    DELETE /{id}                     — remove a prefix (admin only)
    POST   /{id}/lock                — lock (admin only)
    POST   /{id}/unlock              — unlock (admin only)
    POST   /{id}/set-default         — mark as default for that type+company
    POST   /{id}/reset-series        — admin: change current/next number
    POST   /next-number              — internal: atomically reserve the next number
    GET    /{id}/audit               — audit trail for one prefix
    GET    /audit                    — global audit trail (admin)
"""
from __future__ import annotations

from datetime import datetime, timezone
from typing import Any, Optional

from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field

from auth import get_current_user, require_admin
from bson import ObjectId

router = APIRouter(prefix="/api/txn-prefixes", tags=["txn-prefixes"])


# ----------------------------------------------------------------------------
# Catalogue — supported transaction types & token tags
# ----------------------------------------------------------------------------

TXN_TYPES = [
    {"key": "sale", "label": "Sale Invoice", "default_prefix": "INV"},
    {"key": "purchase", "label": "Purchase Invoice", "default_prefix": "PUR"},
    {"key": "quotation", "label": "Quotation / Estimate", "default_prefix": "QUO"},
    {"key": "sale_order", "label": "Sale Order", "default_prefix": "SO"},
    {"key": "proforma", "label": "Proforma Invoice", "default_prefix": "PI"},
    {"key": "challan", "label": "Delivery Challan", "default_prefix": "CHL"},
    {"key": "credit_note", "label": "Credit Note", "default_prefix": "CN"},
    {"key": "debit_note", "label": "Debit Note", "default_prefix": "DN"},
    {"key": "return", "label": "Sales Return", "default_prefix": "RET"},
    {"key": "purchase_return", "label": "Purchase Return", "default_prefix": "PRT"},
    {"key": "payment", "label": "Payment / Receipt", "default_prefix": "PAY"},
    {"key": "expense", "label": "Expense Voucher", "default_prefix": "EXP"},
]

TOKENS = [
    {"token": "{fy}", "desc": "Financial year (e.g. 2026-27)"},
    {"token": "{yy}", "desc": "Short year (e.g. 26)"},
    {"token": "{branch}", "desc": "Branch code"},
    {"token": "{seq}", "desc": "Running counter (auto-padded)"},
]


@router.get("/catalog")
async def catalog(user=Depends(get_current_user)):
    return {"types": TXN_TYPES, "tokens": TOKENS}


# ----------------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------------

def _ser(d: dict) -> dict:
    d = dict(d)
    d["id"] = str(d.pop("_id"))
    return d


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _current_fy() -> str:
    """Indian Financial year: Apr 1 - Mar 31."""
    now = datetime.now(timezone.utc)
    y = now.year
    if now.month < 4:
        return f"{y - 1}-{str(y)[-2:]}"
    return f"{y}-{str(y + 1)[-2:]}"


def _build_invoice_no(template: str, seq: int, padding: int, fy: str, branch: str = "") -> str:
    yy = fy.split("-")[0][-2:] if fy else ""
    out = (template
           .replace("{fy}", fy)
           .replace("{yy}", yy)
           .replace("{branch}", branch or "")
           .replace("{seq}", str(seq).zfill(padding)))
    # Default if no {seq} token in template
    if "{seq}" not in template:
        out = f"{template}{str(seq).zfill(padding)}"
    return out


async def _audit(db, action: str, prefix_id: str, user: dict, before: dict | None, after: dict | None, request: Request | None = None):
    ip = ""
    ua = ""
    if request is not None:
        ip = request.client.host if request.client else ""
        ua = request.headers.get("user-agent", "")[:200]
    await db.prefix_audit.insert_one({
        "prefix_id": prefix_id,
        "action": action,
        "user_id": user.get("id") or "",
        "user_email": user.get("email", ""),
        "user_name": user.get("name", ""),
        "ip": ip,
        "user_agent": ua,
        "before": before,
        "after": after,
        "ts": _now_iso(),
    })


# ----------------------------------------------------------------------------
# Models
# ----------------------------------------------------------------------------

class PrefixIn(BaseModel):
    company_id: str
    type: str
    name: str = ""              # display name (eg "Regal Goa Sales")
    template: str               # e.g. "RM/{fy}/{seq}" or "INV/"
    fy: Optional[str] = None
    branch_code: Optional[str] = ""
    starting_number: int = 1
    current_number: int = 0     # last issued
    padding: int = 5
    is_default: bool = False
    is_locked: bool = False
    is_archived: bool = False
    allowed_users: list[str] = Field(default_factory=list)   # email list; empty = all
    notes: Optional[str] = ""


class PrefixUpdate(BaseModel):
    name: Optional[str] = None
    template: Optional[str] = None
    fy: Optional[str] = None
    branch_code: Optional[str] = None
    starting_number: Optional[int] = None
    padding: Optional[int] = None
    allowed_users: Optional[list[str]] = None
    is_archived: Optional[bool] = None
    notes: Optional[str] = None


class SeriesResetIn(BaseModel):
    next_number: int   # the very next bill will use this exact number


class NextNumberIn(BaseModel):
    prefix_id: Optional[str] = None       # if provided, use this exact prefix
    company_id: Optional[str] = None      # otherwise resolve by company + type defaults
    type: Optional[str] = None


# ----------------------------------------------------------------------------
# CRUD
# ----------------------------------------------------------------------------

@router.get("")
async def list_prefixes(
    request: Request,
    company_id: str | None = None,
    type: str | None = None,
    include_archived: bool = False,
    user=Depends(get_current_user),
):
    """List prefix series for a company. By default excludes archived series so
    the picker/UI never shows broken options. Pass include_archived=true for
    audit / management screens."""
    db = request.app.state.db
    q: dict[str, Any] = {}
    if company_id:
        q["company_id"] = company_id
    if type:
        q["type"] = type
    if not include_archived:
        q["is_archived"] = {"$ne": True}
    docs = await db.txn_prefixes.find(q).sort([("type", 1), ("name", 1)]).to_list(1000)
    return [_ser(d) for d in docs]


@router.post("")
async def create_prefix(payload: PrefixIn, request: Request, user=Depends(require_admin)):
    db = request.app.state.db
    if payload.type not in {t["key"] for t in TXN_TYPES}:
        raise HTTPException(status_code=400, detail="Unknown transaction type")
    doc = payload.model_dump()
    doc["fy"] = payload.fy or _current_fy()
    doc["created_at"] = _now_iso()
    doc["created_by"] = user.get("email", "")
    doc["updated_at"] = doc["created_at"]
    doc["updated_by"] = doc["created_by"]

    # If this is the first prefix for this company+type, automatically mark it default
    has_existing = await db.txn_prefixes.find_one({"company_id": payload.company_id, "type": payload.type})
    if not has_existing:
        doc["is_default"] = True
    elif payload.is_default:
        # Ensure only one default per (company, type)
        await db.txn_prefixes.update_many(
            {"company_id": payload.company_id, "type": payload.type},
            {"$set": {"is_default": False}},
        )

    res = await db.txn_prefixes.insert_one(doc)
    created = await db.txn_prefixes.find_one({"_id": res.inserted_id})
    await _audit(db, "create", str(res.inserted_id), user, None, _ser(created), request)
    return _ser(created)


@router.put("/{prefix_id}")
async def update_prefix(prefix_id: str, payload: PrefixUpdate, request: Request, user=Depends(require_admin)):
    db = request.app.state.db
    try:
        oid = ObjectId(prefix_id)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid id")
    existing = await db.txn_prefixes.find_one({"_id": oid})
    if not existing:
        raise HTTPException(status_code=404, detail="Prefix not found")
    if existing.get("is_locked"):
        raise HTTPException(status_code=423, detail="Prefix is locked — unlock first")
    patch = {k: v for k, v in payload.model_dump(exclude_none=True).items()}
    patch["updated_at"] = _now_iso()
    patch["updated_by"] = user.get("email", "")
    await db.txn_prefixes.update_one({"_id": oid}, {"$set": patch})
    after = await db.txn_prefixes.find_one({"_id": oid})
    await _audit(db, "update", str(oid), user, _ser(existing), _ser(after), request)
    return _ser(after)


@router.delete("/{prefix_id}")
async def delete_prefix(prefix_id: str, request: Request, user=Depends(require_admin)):
    db = request.app.state.db
    try:
        oid = ObjectId(prefix_id)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid id")
    existing = await db.txn_prefixes.find_one({"_id": oid})
    if not existing:
        raise HTTPException(status_code=404, detail="Prefix not found")
    if existing.get("is_locked"):
        raise HTTPException(status_code=423, detail="Cannot delete a locked prefix")
    if (existing.get("current_number") or 0) > 0:
        # Refuse to delete a prefix that has actually issued invoices — archive instead
        await db.txn_prefixes.update_one({"_id": oid}, {"$set": {
            "is_archived": True,
            "updated_at": _now_iso(),
            "updated_by": user.get("email", ""),
        }})
        after = await db.txn_prefixes.find_one({"_id": oid})
        await _audit(db, "archive", str(oid), user, _ser(existing), _ser(after), request)
        return {"deleted": False, "archived": True, "reason": "Series already used — archived to preserve audit"}
    await db.txn_prefixes.delete_one({"_id": oid})
    await _audit(db, "delete", str(oid), user, _ser(existing), None, request)
    return {"deleted": True, "archived": False}


@router.post("/{prefix_id}/lock")
async def lock_prefix(prefix_id: str, request: Request, user=Depends(require_admin)):
    db = request.app.state.db
    try:
        oid = ObjectId(prefix_id)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid id")
    existing = await db.txn_prefixes.find_one({"_id": oid})
    if not existing:
        raise HTTPException(status_code=404, detail="Prefix not found")
    await db.txn_prefixes.update_one({"_id": oid}, {"$set": {"is_locked": True, "updated_at": _now_iso(), "updated_by": user.get("email", "")}})
    after = await db.txn_prefixes.find_one({"_id": oid})
    await _audit(db, "lock", str(oid), user, _ser(existing), _ser(after), request)
    return _ser(after)


@router.post("/{prefix_id}/unlock")
async def unlock_prefix(prefix_id: str, request: Request, user=Depends(require_admin)):
    db = request.app.state.db
    try:
        oid = ObjectId(prefix_id)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid id")
    existing = await db.txn_prefixes.find_one({"_id": oid})
    if not existing:
        raise HTTPException(status_code=404, detail="Prefix not found")
    await db.txn_prefixes.update_one({"_id": oid}, {"$set": {"is_locked": False, "updated_at": _now_iso(), "updated_by": user.get("email", "")}})
    after = await db.txn_prefixes.find_one({"_id": oid})
    await _audit(db, "unlock", str(oid), user, _ser(existing), _ser(after), request)
    return _ser(after)


@router.post("/{prefix_id}/set-default")
async def set_default(prefix_id: str, request: Request, user=Depends(require_admin)):
    db = request.app.state.db
    try:
        oid = ObjectId(prefix_id)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid id")
    existing = await db.txn_prefixes.find_one({"_id": oid})
    if not existing:
        raise HTTPException(status_code=404, detail="Prefix not found")
    if existing.get("is_locked"):
        raise HTTPException(status_code=423, detail="Prefix is locked")
    await db.txn_prefixes.update_many(
        {"company_id": existing["company_id"], "type": existing["type"]},
        {"$set": {"is_default": False}},
    )
    await db.txn_prefixes.update_one({"_id": oid}, {"$set": {"is_default": True, "updated_at": _now_iso(), "updated_by": user.get("email", "")}})
    after = await db.txn_prefixes.find_one({"_id": oid})
    await _audit(db, "set_default", str(oid), user, _ser(existing), _ser(after), request)
    return _ser(after)


@router.post("/{prefix_id}/reset-series")
async def reset_series(prefix_id: str, payload: SeriesResetIn, request: Request, user=Depends(require_admin)):
    """Force the NEXT bill to use this exact number.

    Stores current_number = next_number - 1 so the next atomic $inc lands at
    next_number. Super-admin only. Heavily audit-logged.
    Also resets ALL per-user counters under this prefix so the change applies
    uniformly across every user (matches user expectation of "reset to 1").
    """
    db = request.app.state.db
    try:
        oid = ObjectId(prefix_id)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid id")
    existing = await db.txn_prefixes.find_one({"_id": oid})
    if not existing:
        raise HTTPException(status_code=404, detail="Prefix not found")
    if existing.get("is_locked"):
        raise HTTPException(status_code=423, detail="Prefix is locked — unlock first")
    if payload.next_number < 1:
        raise HTTPException(status_code=400, detail="next_number must be ≥ 1")
    new_current = payload.next_number - 1
    await db.txn_prefixes.update_one({"_id": oid}, {"$set": {
        "current_number": new_current,
        "updated_at": _now_iso(),
        "updated_by": user.get("email", ""),
    }})
    # Per-user counters: snap them to the same baseline so they all "start" from next_number
    await db.prefix_user_counters.update_many(
        {"prefix_id": oid},
        {"$set": {"current_number": new_current, "updated_at": _now_iso()}},
    )
    after = await db.txn_prefixes.find_one({"_id": oid})
    await _audit(db, "reset_series", str(oid), user, _ser(existing), _ser(after), request)
    return _ser(after)


@router.delete("/{prefix_id}/user-counters")
async def wipe_user_counters(prefix_id: str, request: Request, user=Depends(require_admin)):
    """Admin-only — delete every per-user counter doc for a prefix. Used by
    test fixtures to start each scenario from a clean slate; also useful for
    production "Reset all users to 1" recovery."""
    db = request.app.state.db
    try:
        oid = ObjectId(prefix_id)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid id")
    existing = await db.txn_prefixes.find_one({"_id": oid})
    if not existing:
        raise HTTPException(status_code=404, detail="Prefix not found")
    n = (await db.prefix_user_counters.delete_many({"prefix_id": oid})).deleted_count
    await _audit(db, "wipe_user_counters", str(oid), user, None, {"deleted": n}, request)
    return {"deleted": n}


# ----------------------------------------------------------------------------
# Atomic next-number reservation
# ----------------------------------------------------------------------------

async def resolve_next_invoice_no(db, company_id: str, txn_type: str) -> tuple[str, str]:
    """Atomically reserve next number for (company, type). Returns (invoice_no, prefix_id).

    If a default prefix exists, use it; otherwise create one on the fly using
    the catalog defaults so legacy data still works.
    """
    pref = await db.txn_prefixes.find_one_and_update(
        {"company_id": company_id, "type": txn_type, "is_default": True, "is_archived": {"$ne": True}},
        {"$inc": {"current_number": 1}},
        return_document=True,
    )
    if not pref:
        # Auto-bootstrap: create default prefix mirroring legacy behavior
        defaults = next((t for t in TXN_TYPES if t["key"] == txn_type), None)
        default_prefix = defaults["default_prefix"] if defaults else "DOC"
        doc = {
            "company_id": company_id,
            "type": txn_type,
            "name": f"Default {txn_type}",
            "template": f"{default_prefix}/{{yy}}/{{seq}}",
            "fy": _current_fy(),
            "branch_code": "",
            "starting_number": 1,
            "current_number": 1,
            "padding": 5,
            "is_default": True,
            "is_locked": False,
            "is_archived": False,
            "allowed_users": [],
            "notes": "Auto-created on first use",
            "created_at": _now_iso(),
            "updated_at": _now_iso(),
            "created_by": "system",
            "updated_by": "system",
        }
        res = await db.txn_prefixes.insert_one(doc)
        pref = await db.txn_prefixes.find_one({"_id": res.inserted_id})

    seq = int(pref.get("current_number", 1))
    inv_no = _build_invoice_no(
        template=pref.get("template", "INV/{yy}/{seq}"),
        seq=seq,
        padding=int(pref.get("padding", 5)),
        fy=pref.get("fy", _current_fy()),
        branch=pref.get("branch_code", ""),
    )
    return inv_no, str(pref["_id"])


async def resolve_next_invoice_no_by_id(db, prefix_id: str) -> tuple[str, str, str]:
    """v12 — Atomically reserve next number for a SPECIFIC prefix series id.

    Returns: (invoice_no, prefix_id, financial_year)
    Used when the user manually picks a different series from the inline picker
    on the New Invoice screen.
    """
    try:
        oid = ObjectId(prefix_id)
    except Exception as ex:
        raise ValueError(f"Invalid prefix_id: {prefix_id}") from ex
    pref = await db.txn_prefixes.find_one_and_update(
        {"_id": oid, "is_archived": {"$ne": True}, "is_locked": {"$ne": True}},
        {"$inc": {"current_number": 1}},
        return_document=True,
    )
    if not pref:
        raise ValueError(f"Prefix {prefix_id} not found or locked")
    seq = int(pref.get("current_number", 1))
    inv_no = _build_invoice_no(
        template=pref.get("template", "INV/{yy}/{seq}"),
        seq=seq,
        padding=int(pref.get("padding", 5)),
        fy=pref.get("fy", _current_fy()),
        branch=pref.get("branch_code", ""),
    )
    return inv_no, str(pref["_id"]), pref.get("fy", _current_fy())


# ----------------------------------------------------------------------------
# v12.7 — Per-user counters (Multi-user + Auto/Manual unified sequence)
# ----------------------------------------------------------------------------
# A separate counter is kept per (prefix_id, user_email, fy) so:
#   * User A's running sequence never collides with User B's
#   * Same prefix series can serve multiple users in parallel
#   * Auto + Manual entry share the SAME counter — manual bumps it forward,
#     auto continues from the max ever issued
#   * Counter naturally resets when financial year changes (different fy key)

async def _parse_trailing_seq(inv_no: str) -> int:
    """Pull the trailing integer out of an invoice number.
    "RM/2026-27/10"  → 10
    "INV-0001"       → 1
    "RM/2026-27/"    → 0  (no trailing integer)
    """
    import re as _re
    if not inv_no:
        return 0
    m = _re.search(r"(\d+)\s*$", inv_no)
    if not m:
        return 0
    try:
        return int(m.group(1))
    except Exception:
        return 0


async def _seed_user_counter_if_missing(db, prefix_id_obj, user_email: str, fy: str, company_id: str, txn_type: str) -> int:
    """On first use for a (prefix, user, fy) triple, seed the counter from the
    HIGHEST trailing number this user has ever issued for the same (company,
    type, prefix_id) — so legacy data continues smoothly. Returns the seed
    value (the doc is created with current_number = seed)."""
    # If the per-user counter doc already exists, do nothing
    existing = await db.prefix_user_counters.find_one({"prefix_id": prefix_id_obj, "user_email": user_email, "fy": fy})
    if existing is not None:
        return int(existing.get("current_number", 0))

    # Scan this user's existing invoices for the same (company, type, prefix_id)
    # to find the highest issued sequence — keeps backward compat for existing data
    max_seq = 0
    try:
        cursor = db.invoices.find({
            "company_id": company_id,
            "type": txn_type,
            "created_by": user_email,
            "prefix_id": str(prefix_id_obj),
        }, {"invoice_no": 1})
        async for inv in cursor:
            n = await _parse_trailing_seq(inv.get("invoice_no", ""))
            if n > max_seq:
                max_seq = n
    except Exception:
        pass

    # Create the per-user counter document. Use upsert with $setOnInsert to
    # be race-safe (two concurrent first-uses would both see existing=None).
    await db.prefix_user_counters.update_one(
        {"prefix_id": prefix_id_obj, "user_email": user_email, "fy": fy},
        {"$setOnInsert": {
            "prefix_id": prefix_id_obj,
            "user_email": user_email,
            "fy": fy,
            "company_id": company_id,
            "txn_type": txn_type,
            "current_number": max_seq,
            "created_at": _now_iso(),
        }},
        upsert=True,
    )
    return max_seq


async def resolve_next_invoice_no_per_user(db, *, company_id: str, txn_type: str, user_email: str, prefix_id: Optional[str] = None) -> tuple[str, str, str]:
    """Atomically reserve the next number for (prefix, user, fy).
    Returns (invoice_no, prefix_id_str, financial_year).

    If prefix_id is None, the default prefix for (company, type) is used and
    auto-bootstrapped if it doesn't exist (mirrors the legacy fallback).
    """
    # Resolve the prefix doc (use specific or default)
    if prefix_id:
        try:
            pref_oid = ObjectId(prefix_id)
        except Exception as ex:
            raise ValueError(f"Invalid prefix_id: {prefix_id}") from ex
        pref = await db.txn_prefixes.find_one({"_id": pref_oid, "is_archived": {"$ne": True}, "is_locked": {"$ne": True}})
        if not pref:
            raise ValueError(f"Prefix {prefix_id} not found or locked")
    else:
        pref = await db.txn_prefixes.find_one(
            {"company_id": company_id, "type": txn_type, "is_default": True, "is_archived": {"$ne": True}},
        )
        if not pref:
            # Auto-bootstrap default (same logic as legacy resolve)
            defaults = next((t for t in TXN_TYPES if t["key"] == txn_type), None)
            default_prefix = defaults["default_prefix"] if defaults else "DOC"
            doc = {
                "company_id": company_id, "type": txn_type,
                "name": f"Default {txn_type}",
                "template": f"{default_prefix}/{{yy}}/{{seq}}",
                "fy": _current_fy(), "branch_code": "",
                "starting_number": 1, "current_number": 0, "padding": 5,
                "is_default": True, "is_locked": False, "is_archived": False,
                "allowed_users": [], "notes": "Auto-created on first use",
                "created_at": _now_iso(), "updated_at": _now_iso(),
                "created_by": "system", "updated_by": "system",
            }
            res = await db.txn_prefixes.insert_one(doc)
            pref = await db.txn_prefixes.find_one({"_id": res.inserted_id})

    fy = pref.get("fy") or _current_fy()
    pref_oid = pref["_id"]

    # Seed first-use counter from highest issued for backward compat
    await _seed_user_counter_if_missing(db, pref_oid, user_email, fy, company_id, txn_type)

    # Atomic increment — concurrent safe
    counter = await db.prefix_user_counters.find_one_and_update(
        {"prefix_id": pref_oid, "user_email": user_email, "fy": fy},
        {"$inc": {"current_number": 1}, "$set": {"updated_at": _now_iso()}},
        return_document=True,
    )
    seq = int(counter.get("current_number", 1))
    inv_no = _build_invoice_no(
        template=pref.get("template", "INV/{yy}/{seq}"),
        seq=seq,
        padding=int(pref.get("padding", 5)),
        fy=fy,
        branch=pref.get("branch_code", ""),
    )
    return inv_no, str(pref_oid), fy


async def bump_user_counter_for_manual(db, *, company_id: str, txn_type: str, user_email: str, manual_invoice_no: str, prefix_id: Optional[str] = None) -> tuple[str, str]:
    """Manual bill saved — sync the per-user counter to max(current, manual_seq).
    Guarantees the next AUTO bill continues from where the manual left off.
    Returns (prefix_id_str, financial_year).
    """
    # Resolve the prefix to attach the counter to
    if prefix_id:
        try:
            pref_oid = ObjectId(prefix_id)
        except Exception as ex:
            raise ValueError(f"Invalid prefix_id: {prefix_id}") from ex
        pref = await db.txn_prefixes.find_one({"_id": pref_oid})
    else:
        pref = await db.txn_prefixes.find_one(
            {"company_id": company_id, "type": txn_type, "is_default": True, "is_archived": {"$ne": True}},
        )
    if not pref:
        # No prefix series exists yet — nothing to bump
        return "", _current_fy()
    fy = pref.get("fy") or _current_fy()
    pref_oid = pref["_id"]
    manual_seq = await _parse_trailing_seq(manual_invoice_no)
    # Seed if needed (so the counter doc exists for max-compare)
    await _seed_user_counter_if_missing(db, pref_oid, user_email, fy, company_id, txn_type)
    # Atomic max-update: only bump forward, never backward
    # MongoDB doesn't have $max directly applied to an update we want to chain
    # with concurrent safety, so use a careful conditional update with
    # $max — guaranteed atomic and non-decreasing.
    await db.prefix_user_counters.update_one(
        {"prefix_id": pref_oid, "user_email": user_email, "fy": fy},
        {"$max": {"current_number": manual_seq}, "$set": {"updated_at": _now_iso()}},
    )
    return str(pref_oid), fy


async def preview_next_invoice_no(db, company_id: str, txn_type: str) -> tuple[str, str | None, str]:
    """Return what the NEXT invoice number WOULD BE — without consuming the
    sequence. Used by NewInvoice screen to show the upcoming number before
    the user clicks Save.

    Returns: (invoice_no_preview, prefix_id or None, financial_year)
    """
    pref = await db.txn_prefixes.find_one(
        {"company_id": company_id, "type": txn_type, "is_default": True, "is_archived": {"$ne": True}},
    )
    if pref:
        seq = int(pref.get("current_number", 1))
        inv_no = _build_invoice_no(
            template=pref.get("template", "INV/{yy}/{seq}"),
            seq=seq,
            padding=int(pref.get("padding", 5)),
            fy=pref.get("fy", _current_fy()),
            branch=pref.get("branch_code", ""),
        )
        return inv_no, str(pref["_id"]), pref.get("fy", _current_fy())
    # No prefix yet — show what the auto-bootstrapped one would look like
    defaults = next((t for t in TXN_TYPES if t["key"] == txn_type), None)
    default_prefix = defaults["default_prefix"] if defaults else "DOC"
    fy = _current_fy()
    inv_no = _build_invoice_no(f"{default_prefix}/{{yy}}/{{seq}}", 1, 5, fy, "")
    return inv_no, None, fy


@router.post("/next-number")
async def next_number_api(payload: NextNumberIn, request: Request, user=Depends(get_current_user)):
    db = request.app.state.db
    if payload.prefix_id:
        try:
            pref = await db.txn_prefixes.find_one_and_update(
                {"_id": ObjectId(payload.prefix_id), "is_archived": {"$ne": True}},
                {"$inc": {"current_number": 1}},
                return_document=True,
            )
        except Exception:
            raise HTTPException(status_code=400, detail="Invalid prefix id")
        if not pref:
            raise HTTPException(status_code=404, detail="Prefix not found")
        seq = int(pref.get("current_number", 1))
        inv_no = _build_invoice_no(
            pref.get("template", "INV/{yy}/{seq}"),
            seq, int(pref.get("padding", 5)),
            pref.get("fy", _current_fy()),
            pref.get("branch_code", ""),
        )
        return {"invoice_no": inv_no, "prefix_id": str(pref["_id"])}

    if not payload.company_id or not payload.type:
        raise HTTPException(status_code=400, detail="company_id + type or prefix_id required")
    inv_no, pid = await resolve_next_invoice_no(db, payload.company_id, payload.type)
    return {"invoice_no": inv_no, "prefix_id": pid}


@router.get("/preview-number")
async def preview_number_api(
    request: Request,
    company_id: str,
    type: str,
    user=Depends(get_current_user),
):
    """Returns the NEXT invoice number that WOULD be assigned, without consuming
    the sequence. Used by NewInvoice screen for upfront display."""
    db = request.app.state.db
    inv_no, pid, fy = await preview_next_invoice_no(db, company_id, type)
    return {
        "invoice_no": inv_no,
        "prefix_id": pid,
        "financial_year": fy,
        "is_preview": True,
    }


# ----------------------------------------------------------------------------
# Audit endpoints
# ----------------------------------------------------------------------------

@router.get("/audit")
async def global_audit(request: Request, user=Depends(require_admin), limit: int = 200):
    db = request.app.state.db
    cursor = db.prefix_audit.find({}, {"_id": 0}).sort("ts", -1).limit(limit)
    return [d async for d in cursor]


@router.get("/{prefix_id}/audit")
async def prefix_audit(prefix_id: str, request: Request, user=Depends(get_current_user), limit: int = 200):
    db = request.app.state.db
    cursor = db.prefix_audit.find({"prefix_id": prefix_id}, {"_id": 0}).sort("ts", -1).limit(limit)
    return [d async for d in cursor]