"""services/cc/service.py — THE CC pass-through expenses service
(cc-expenses.md B4–B54). Batches, statement-paste ingestion with
auto-rule stamping, per-line assignment, highlight keywords, re-apply,
and the PlusROI paste-block export. Thin HTTP layer lives in
app/routers/cc.py; all rules live here (04-architecture).

R1 parity + the plan's rebuild deltas:
  - **Owner-only across the board** (cc-expenses.md Permissions): every
    read and write requires ``can_transfer``. The legacy RLS-any-member
    layer does not port — authz is the app-layer choke point (delta #4).
  - **Explicit tenant scoping** on every mutation (delta #5): the legacy
    RLS-only paths (update/delete line, delete batch, append, re-apply)
    become explicit tenant checks here.
  - **Project-tenant re-validated on rule create AND update** (delta:
    the legacy update skipped it — known inconsistency, fixed here).
  - **Separate client/project end-to-end in the export** (delta: kills
    the legacy ``' : '`` split wart) — the paste block joins
    project->client directly instead of round-tripping a label.
  - **Column-6 literal is a tenant setting** (delta / M6 parity):
    ``format_cc_paste_block`` reads ``tenants.settings["paste_block"]``.
  - **invoice_line_id is never written** (task R1: bill-through stays a
    stateless export; the lock wiring is future work).
  - Carried warts (faithful): created_at-DESC rule precedence
    (newest-rule-wins, no priority column), no paste dedupe, credit
    sign-agnostic amounts, loose date validation — see the parser
    module + docs/features/cc-expenses.md "Open considerations".

Every mutation writes an audit row (reuses services/audit.write_audit)
for forensics; CC actions are NOT in UNDOABLE_ACTIONS (parity: CC has
no undo — cc-expenses.md B396).
"""

from __future__ import annotations

import uuid
from datetime import date
from decimal import Decimal

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from app.models import (
    CcAutoRule,
    CcExpenseBatch,
    CcExpenseLine,
    CcHighlightKeyword,
    Expense,
    Party,
    Project,
    Tenant,
)
from app.services.audit import write_audit
from app.services.cc.parse import (
    AssignedCcLineForPaste,
    AutoRuleMatch,
    find_auto_rule_match,
    format_cc_paste_block,
    parse_cc_paste,
)
from app.services.entities.resolve import load_entity_graph
from app.services.errors import ServiceError
from app.services.money.expenses import ledger_row
from app.services.money.paste_block import paste_block_config
from app.services.plusroi_gates import require_writable_books

#: Upper bound on a statement paste (M8 hardening). A real CC statement is
#: a few hundred lines; 1 MB is generous. Mirrors the CSV import's
#: MAX_UPLOAD_BYTES so both bulk-text ingress paths are bounded.
MAX_PASTE_BYTES = 1_000_000

# --------------------------------------------------------------- helpers


def _require_owner(actor, message: str) -> None:
    """The whole feature is owner-only (cc-expenses.md Permissions):
    ``can_transfer`` == owner OR (super-admin AND not viewing-as)."""
    if actor.tenant_id is None or not actor.allowed("can_transfer"):
        raise ServiceError(403, "NOT_ALLOWED", message)


def _amount(raw: Decimal | None) -> str | None:
    return None if raw is None else f"{Decimal(str(raw)):.2f}"


def _get_batch(session: Session, actor, batch_id: uuid.UUID) -> CcExpenseBatch:
    batch = session.scalars(
        select(CcExpenseBatch).where(
            CcExpenseBatch.id == batch_id,
            CcExpenseBatch.tenant_id == actor.tenant_id,
        )
    ).first()
    if batch is None:
        raise ServiceError(404, "CC_BATCH_NOT_FOUND", "Batch not found.")
    return batch


def _get_line(session: Session, actor, line_id: uuid.UUID) -> CcExpenseLine:
    """Line scoped to the actor's tenant THROUGH its batch (the lines
    table has no tenant_id — org scoping goes via the batch)."""
    line = session.scalars(
        select(CcExpenseLine)
        .join(CcExpenseBatch, CcExpenseBatch.id == CcExpenseLine.batch_id)
        .where(
            CcExpenseLine.id == line_id,
            CcExpenseBatch.tenant_id == actor.tenant_id,
        )
    ).first()
    if line is None:
        raise ServiceError(404, "CC_LINE_NOT_FOUND", "Expense line not found.")
    return line


def _validate_project(session: Session, actor, raw) -> uuid.UUID:
    """Explicit project-tenant check (delta #5): the project must exist
    AND belong to the actor's tenant. Used by rule create/update and line
    assignment — all three re-validate now (the legacy update did not)."""
    try:
        project_id = uuid.UUID(str(raw))
    except (ValueError, TypeError) as exc:
        raise ServiceError(422, "BAD_PROJECT", "project_id must be a UUID.") from exc
    ok = session.scalar(
        select(Project.id).where(
            Project.id == project_id, Project.tenant_id == actor.tenant_id
        )
    )
    if ok is None:
        raise ServiceError(
            404, "PROJECT_NOT_IN_ORG", "Selected project not found in this org."
        )
    return project_id


def _tenant_currency(session: Session, actor) -> str:
    settings = (
        session.scalar(select(Tenant.settings).where(Tenant.id == actor.tenant_id))
        or {}
    )
    return settings.get("default_currency", "CAD")


def _operator_label(session: Session, actor) -> str:
    """Paste-block column 6 — a tenant setting (M6 parity, delta), default
    'Bowden Works' so the export is byte-identical to the legacy hard-code."""
    settings = session.scalar(
        select(Tenant.settings).where(Tenant.id == actor.tenant_id)
    )
    return paste_block_config(settings).operator_label


def _project_labels(session: Session, actor, project_ids: set[uuid.UUID]) -> dict:
    """project id -> "{client} : {project}" (or None when either name is
    missing). Direct project->client join — no ``' : '`` round-trip."""
    if not project_ids:
        return {}
    rows = session.execute(
        select(Project.id, Project.name, Party.name)
        .join(Party, Party.id == Project.client_party_id)
        .where(Project.id.in_(project_ids))
    ).all()
    out: dict[uuid.UUID, str | None] = {}
    for pid, project_name, client_name in rows:
        out[pid] = (
            f"{client_name} : {project_name}"
            if client_name and project_name
            else None
        )
    return out


def _project_options(session: Session, actor) -> list[dict]:
    """Picker options: every tenant project labeled ``{client} :
    {project}`` (the single EntityPicker vocabulary — ADR #042). Built
    from the entity graph, sorted by label."""
    graph = load_entity_graph(session, actor.tenant_id)
    options = [
        {
            "id": str(project_id),
            "label": f"{graph.client_names.get(cl_id, '')} : {name}",
            "client": graph.client_names.get(cl_id, ""),
            "project": name,
            "operator": graph.operators.get(op_id, ""),
        }
        for project_id, (name, op_id, cl_id) in graph.project_chain.items()
    ]
    options.sort(key=lambda o: o["label"].lower())
    return options


def _line_categories(session: Session, actor) -> list[str]:
    """Distinct non-null categories across ALL the tenant's CC lines
    (B27 datalist), joined through the batch for tenant scope."""
    rows = session.execute(
        select(CcExpenseLine.category)
        .join(CcExpenseBatch, CcExpenseBatch.id == CcExpenseLine.batch_id)
        .where(
            CcExpenseBatch.tenant_id == actor.tenant_id,
            CcExpenseLine.category.is_not(None),
        )
        .distinct()
    ).all()
    return sorted({r[0] for r in rows if r[0]})


def _keywords(session: Session, actor) -> list[str]:
    """Highlight keyword strings, ordered keyword ASC (B31 order)."""
    return list(
        session.scalars(
            select(CcHighlightKeyword.keyword)
            .where(CcHighlightKeyword.tenant_id == actor.tenant_id)
            .order_by(CcHighlightKeyword.keyword)
        ).all()
    )


def _load_rules(session: Session, actor) -> list[AutoRuleMatch]:
    """All tenant rules as matcher inputs, ordered created_at DESC — so
    ``find_auto_rule_match`` first-match == NEWEST rule wins (B22/B23)."""
    rows = session.scalars(
        select(CcAutoRule)
        .where(CcAutoRule.tenant_id == actor.tenant_id)
        .order_by(CcAutoRule.created_at.desc(), CcAutoRule.id)
    ).all()
    return [
        AutoRuleMatch(
            id=str(r.id),
            keyword=r.keyword,
            project_id=str(r.project_id),
            assignment_description=r.assignment_description,
            category=r.category,
        )
        for r in rows
    ]


def _line_dict(line: CcExpenseLine, label: str | None) -> dict:
    return {
        "id": str(line.id),
        "line_index": line.line_index,
        "expense_date": line.expense_date.isoformat() if line.expense_date else None,
        "description": line.description,
        "amount": _amount(line.amount),
        "balance": _amount(line.balance),
        "project_id": str(line.project_id) if line.project_id else None,
        "project_label": label,
        "assignment_description": line.assignment_description,
        "category": line.category,
        "auto_assigned": bool(line.auto_assigned),
    }


# ----------------------------------------------------------- batch reads


def list_batches(session: Session, actor) -> list[dict]:
    """B4–B7: batches created_at DESC with per-batch aggregates in ONE
    grouped query. ``total`` sums ALL lines (unassigned included).

    **Per-user batch privacy (phase-3 slice 3, ADR #024):** the list
    scopes to the batches the EFFECTIVE user created (Rob sees only his,
    never rian's) plus ownerless rows (``created_by IS NULL`` — migrated
    batches whose legacy creator didn't map, or a since-deleted user;
    hiding those from everyone would strand them). Safe for BW (one
    owner, who created every batch). Only the LIST scopes — every other
    read/write stays tenant-scoped as before, so a shared batch URL
    still opens."""
    _require_owner(actor, "Owner only.")
    stmt = (
        select(CcExpenseBatch)
        .where(CcExpenseBatch.tenant_id == actor.tenant_id)
        .where(
            or_(
                CcExpenseBatch.created_by == actor.user_id,
                CcExpenseBatch.created_by.is_(None),
            )
            if actor.user_id is not None
            else CcExpenseBatch.created_by.is_(None)
        )
        .order_by(CcExpenseBatch.created_at.desc(), CcExpenseBatch.id)
    )
    batches = list(session.scalars(stmt).all())
    if not batches:
        return []
    agg = {
        row.batch_id: row
        for row in session.execute(
            select(
                CcExpenseLine.batch_id.label("batch_id"),
                func.count().label("lines"),
                func.count(CcExpenseLine.project_id).label("assigned"),
                func.coalesce(func.sum(CcExpenseLine.amount), 0).label("total"),
            )
            .where(CcExpenseLine.batch_id.in_([b.id for b in batches]))
            .group_by(CcExpenseLine.batch_id)
        ).all()
    }
    out = []
    for b in batches:
        a = agg.get(b.id)
        lines = int(a.lines) if a else 0
        assigned = int(a.assigned) if a else 0
        total = Decimal(str(a.total)) if a and a.total is not None else Decimal(0)
        out.append(
            {
                "id": str(b.id),
                "name": b.name,
                "currency": b.currency,
                "created_at": b.created_at.isoformat() if b.created_at else None,
                "lines": lines,
                "assigned": assigned,
                "assigned_pct": round(assigned / lines * 100) if lines else None,
                "total": f"{total:.2f}",
            }
        )
    return out


def batch_detail(session: Session, actor, batch_id: uuid.UUID) -> dict:
    """B40–B43: the batch, its lines (line_index ASC) with project labels,
    header totals, plus the picker/highlight data the page renders with."""
    _require_owner(actor, "Owner only.")
    batch = _get_batch(session, actor, batch_id)
    lines = list(
        session.scalars(
            select(CcExpenseLine)
            .where(CcExpenseLine.batch_id == batch.id)
            .order_by(CcExpenseLine.line_index)
        ).all()
    )
    labels = _project_labels(
        session, actor, {ln.project_id for ln in lines if ln.project_id}
    )
    total_amount = sum((ln.amount or Decimal(0) for ln in lines), Decimal(0))
    assigned = sum(1 for ln in lines if ln.project_id is not None)
    return {
        "batch": {
            "id": str(batch.id),
            "name": batch.name,
            "currency": batch.currency,
            "created_at": batch.created_at.isoformat() if batch.created_at else None,
        },
        "lines": [_line_dict(ln, labels.get(ln.project_id)) for ln in lines],
        "totals": {
            "lines": len(lines),
            "assigned": assigned,
            "total": f"{total_amount:.2f}",
        },
        "highlight_keywords": _keywords(session, actor),
        "category_options": _line_categories(session, actor),
        "project_options": _project_options(session, actor),
    }


def paste_block(session: Session, actor, batch_id: uuid.UUID) -> dict:
    """B44–B47: the PlusROI export. Only lines with project + label +
    amount + date export; the rest are silently skipped. Byte-identical
    to the legacy formatCcPasteBlock (tests/cc/test_parser_gate.py)."""
    _require_owner(actor, "Owner only.")
    batch = _get_batch(session, actor, batch_id)
    rows = session.execute(
        select(
            CcExpenseLine.expense_date,
            CcExpenseLine.assignment_description,
            CcExpenseLine.amount,
            CcExpenseLine.category,
            Party.name.label("client"),
            Project.name.label("project"),
        )
        .join(Project, Project.id == CcExpenseLine.project_id)
        .join(Party, Party.id == Project.client_party_id)
        .where(
            CcExpenseLine.batch_id == batch.id,
            CcExpenseLine.project_id.is_not(None),
            CcExpenseLine.expense_date.is_not(None),
            CcExpenseLine.amount.is_not(None),
        )
        .order_by(CcExpenseLine.line_index)
    ).all()
    assigned = [
        AssignedCcLineForPaste(
            client=r.client,
            project=r.project,
            assignment_description=r.assignment_description or "",
            amount=Decimal(str(r.amount)),
            expense_date=r.expense_date.isoformat(),
            category=r.category or "",
        )
        for r in rows
        if r.client and r.project
    ]
    text = format_cc_paste_block(assigned, operator_label=_operator_label(session, actor))
    return {"text": text, "line_count": len(assigned)}


# ---------------------------------------------------------- batch writes


def create_batch(session: Session, actor, name: str | None) -> dict:
    """B10: name required. currency defaults to the tenant currency (the
    batch is one card / one currency).

    ``created_by`` = the EFFECTIVE user (ADR #024): with per-user batch
    privacy the stamp is the OWNERSHIP key the list filters on, so it
    must follow the impersonated identity or a batch created under
    view-as-Rob would vanish from Rob's own list. Outside view-as the
    effective and real users are the same person, and under view-as the
    real human stays recoverable on the audit row (``impersonated_by``)."""
    _require_owner(actor, "Only an owner can create expense batches.")
    clean = (name or "").strip()
    if not clean:
        raise ServiceError(422, "NAME_REQUIRED", "Batch name is required.")
    batch = CcExpenseBatch(
        id=uuid.uuid4(),
        tenant_id=actor.tenant_id,
        name=clean,
        currency=_tenant_currency(session, actor),
        created_by=actor.user_id,
    )
    session.add(batch)
    session.flush()
    write_audit(
        session,
        actor,
        action="cc.batch_create",
        entity_type="cc_batch",
        entity_id=batch.id,
        after={"name": clean},
    )
    session.commit()
    return {
        "id": str(batch.id),
        "name": batch.name,
        "currency": batch.currency,
        "created_at": batch.created_at.isoformat() if batch.created_at else None,
        "lines": 0,
        "assigned": 0,
        "assigned_pct": None,
        "total": "0.00",
    }


def delete_batch(session: Session, actor, batch_id: uuid.UUID) -> dict:
    """B11: delete the batch; lines cascade (FK ON DELETE CASCADE). No
    undo (parity) — the browser confirm is the only guard."""
    _require_owner(actor, "Only an owner can delete expense batches.")
    batch = _get_batch(session, actor, batch_id)
    name = batch.name
    session.delete(batch)
    write_audit(
        session,
        actor,
        action="cc.batch_delete",
        entity_type="cc_batch",
        entity_id=batch_id,
        before={"name": name},
    )
    session.commit()
    return {"ok": True, "deleted": str(batch_id)}


def append_lines(session: Session, actor, batch_id: uuid.UUID, paste: str | None) -> dict:
    """B19–B25: parse the paste, auto-stamp matching rules (newest wins),
    bulk-insert continuing line_index from the batch max. No dedupe
    (parity). Returns the count + auto-assigned count + flash text."""
    _require_owner(actor, "Only an owner can add expense lines.")
    batch = _get_batch(session, actor, batch_id)
    # B20 two distinct guards: a truly empty field -> "Paste some lines
    # first."; non-empty content that parses to nothing (whitespace/blank
    # lines only) -> "No usable lines found in the paste." (T-CCE-A-011/012).
    if paste is None or paste == "":
        raise ServiceError(422, "EMPTY_PASTE", "Paste some lines first.")
    if len(paste.encode("utf-8")) > MAX_PASTE_BYTES:
        raise ServiceError(
            413, "PASTE_TOO_LARGE", "Statement paste too large (1MB max)."
        )
    parsed = parse_cc_paste(paste)
    if not parsed:
        raise ServiceError(422, "NO_USABLE_LINES", "No usable lines found in the paste.")

    max_index = session.scalar(
        select(func.max(CcExpenseLine.line_index)).where(
            CcExpenseLine.batch_id == batch.id
        )
    )
    start = (max_index if max_index is not None else -1) + 1

    rules = _load_rules(session, actor)
    auto = 0
    for i, p in enumerate(parsed):
        match = find_auto_rule_match(p.description, rules)
        if match is not None:
            auto += 1
        session.add(
            CcExpenseLine(
                id=uuid.uuid4(),
                batch_id=batch.id,
                line_index=start + i,
                raw_line=p.raw_line,
                expense_date=(
                    date.fromisoformat(p.expense_date) if p.expense_date else None
                ),
                description=p.description,
                amount=p.amount,
                balance=p.balance,
                project_id=uuid.UUID(match.project_id) if match else None,
                assignment_description=match.assignment_description if match else None,
                category=match.category if match else None,
                auto_assigned=match is not None,
            )
        )
    write_audit(
        session,
        actor,
        action="cc.lines_append",
        entity_type="cc_batch",
        entity_id=batch.id,
        after={"added": len(parsed), "auto_assigned": auto},
    )
    session.commit()
    n = len(parsed)
    message = (
        f"Added {n} lines ({auto} auto-assigned)." if auto > 0 else f"Added {n} lines."
    )
    return {"added": n, "auto_assigned": auto, "message": message}


def _port_marker(line_id: uuid.UUID) -> str:
    """The provenance marker a ported line leaves on its expense row
    (``expenses.external_invoice_id``). A CC charge has no vendor invoice
    # of its own, so the source-document ref IS the cc line — and the
    marker doubles as the double-port guard (ADR #024): a line whose
    marker already exists on a tenant expense is skipped. Deleting the
    ported expense frees the marker, deliberately re-enabling the port."""
    return f"cc:{line_id}"


def port_to_expenses(session: Session, actor, batch_id: uuid.UUID, payload: dict) -> dict:
    """POST /api/cc/batches/{id}/port-to-expenses — move assigned CC
    lines into the CURRENT tenant's PlusROI expenses journal (phase-3
    slice 3, docs/design/plusroi-forms.md §5; ADR #024).

    Eligibility per line: a project assignment AND an amount. Everything
    else is a counted skip — ``skipped_unassigned`` /
    ``skipped_no_amount`` / ``skipped_already_ported`` (the marker
    guard) — so re-running is safe and Rob can port a batch as he
    assigns it, in waves. ``line_ids`` narrows the port to a selection;
    omitted = the whole batch. Each created row: ``source='manual'``,
    project/amount/category/date from the line, currency from the batch,
    ``paid_by`` = the caller's own party unless a party is chosen (the
    fronting partner the settlement later reimburses), notes = the
    bill-through description. Owner + writable-books gated, audited."""
    _require_owner(actor, "Only an owner can port expenses.")
    require_writable_books(session, actor)
    batch = _get_batch(session, actor, batch_id)

    lines = list(
        session.scalars(
            select(CcExpenseLine)
            .where(CcExpenseLine.batch_id == batch.id)
            .order_by(CcExpenseLine.line_index)
        ).all()
    )
    raw_ids = payload.get("line_ids")
    if raw_ids is not None:
        try:
            wanted = {uuid.UUID(str(v)) for v in raw_ids}
        except (ValueError, TypeError) as exc:
            raise ServiceError(
                422, "BAD_IDS", "line_ids must be a list of UUIDs."
            ) from exc
        by_id = {ln.id for ln in lines}
        if not wanted <= by_id:
            raise ServiceError(
                404, "CC_LINE_NOT_FOUND", "Some selected lines aren't in this batch."
            )
        lines = [ln for ln in lines if ln.id in wanted]

    raw_paid_by = payload.get("paid_by_party_id")
    if raw_paid_by in (None, ""):
        paid_by = getattr(actor, "person_party_id", None)
    else:
        try:
            paid_by = uuid.UUID(str(raw_paid_by))
        except (ValueError, TypeError) as exc:
            raise ServiceError(
                422, "BAD_PARTY", "paid_by_party_id must be a UUID."
            ) from exc
        if session.scalar(select(Party.id).where(Party.id == paid_by)) is None:
            raise ServiceError(
                404, "PARTY_NOT_FOUND", "Selected paid-by party not found."
            )

    candidates = [ln for ln in lines if ln.project_id is not None and ln.amount is not None]
    skipped_unassigned = sum(1 for ln in lines if ln.project_id is None)
    skipped_no_amount = sum(
        1 for ln in lines if ln.project_id is not None and ln.amount is None
    )

    already = set()
    if candidates:
        markers = [_port_marker(ln.id) for ln in candidates]
        already = {
            marker
            for (marker,) in session.execute(
                select(Expense.external_invoice_id).where(
                    Expense.tenant_id == actor.tenant_id,
                    Expense.external_invoice_id.in_(markers),
                )
            ).all()
        }
    to_port = [ln for ln in candidates if _port_marker(ln.id) not in already]
    skipped_already_ported = len(candidates) - len(to_port)

    if not to_port:
        raise ServiceError(
            409,
            "NOTHING_TO_PORT",
            "No lines left to port — every selected line is unassigned,"
            " amount-less, or already ported.",
            f"Skipped: {skipped_unassigned} unassigned, {skipped_no_amount}"
            f" without an amount, {skipped_already_ported} already ported.",
        )

    created: list[Expense] = []
    for line in to_port:
        expense = Expense(
            id=uuid.uuid4(),
            tenant_id=actor.tenant_id,
            project_id=line.project_id,
            vendor_party_id=None,  # the merchant string lives in notes
            amount=Decimal(str(line.amount)),
            currency=batch.currency,
            expense_date=line.expense_date,
            external_invoice_id=_port_marker(line.id),
            paid_by_party_id=paid_by,
            category=line.category,
            notes=line.assignment_description or line.description,
            source="manual",
            created_by=actor.real_user_id,
        )
        session.add(expense)
        created.append(expense)
    session.flush()

    write_audit(
        session,
        actor,
        action="cc.port_to_expenses",
        entity_type="cc_batch",
        entity_id=batch.id,
        after={
            "ported": {
                str(ln.id): str(e.id)
                for ln, e in zip(to_port, created, strict=True)
            },
            "paid_by_party_id": str(paid_by) if paid_by is not None else None,
            "skipped_unassigned": skipped_unassigned,
            "skipped_no_amount": skipped_no_amount,
            "skipped_already_ported": skipped_already_ported,
        },
    )
    session.commit()
    return {
        "ok": True,
        "ported": len(created),
        "skipped_unassigned": skipped_unassigned,
        "skipped_no_amount": skipped_no_amount,
        "skipped_already_ported": skipped_already_ported,
        "rows": [ledger_row(session, e) for e in created],
    }


# ----------------------------------------------------------- line writes


def update_line(session: Session, actor, line_id: uuid.UUID, patch: dict) -> dict:
    """B28: set project/description/category and flip auto_assigned FALSE
    (saving IS confirming: orange -> green). Clearing the project makes
    the line unassigned again (B29). project-tenant re-validated."""
    _require_owner(actor, "Only an owner can edit expense lines.")
    line = _get_line(session, actor, line_id)

    raw_project = patch.get("project_id")
    if raw_project in (None, ""):
        line.project_id = None
    else:
        line.project_id = _validate_project(session, actor, raw_project)
    line.assignment_description = (patch.get("assignment_description") or "").strip() or None
    line.category = (patch.get("category") or "").strip() or None
    line.auto_assigned = False  # every save confirms (B28/B35)

    write_audit(
        session,
        actor,
        action="cc.line_update",
        entity_type="cc_line",
        entity_id=line.id,
        after={
            "project_id": str(line.project_id) if line.project_id else None,
            "assignment_description": line.assignment_description,
            "category": line.category,
        },
    )
    session.commit()
    labels = _project_labels(
        session, actor, {line.project_id} if line.project_id else set()
    )
    return {"ok": True, "row": _line_dict(line, labels.get(line.project_id))}


def delete_line(session: Session, actor, line_id: uuid.UUID) -> dict:
    """B30: remove one mis-parsed / non-billable row. No undo (parity)."""
    _require_owner(actor, "Only an owner can edit expense lines.")
    line = _get_line(session, actor, line_id)
    batch_id = line.batch_id
    session.delete(line)
    write_audit(
        session,
        actor,
        action="cc.line_delete",
        entity_type="cc_line",
        entity_id=line_id,
        before={"batch_id": str(batch_id), "raw_line": line.raw_line},
    )
    session.commit()
    return {"ok": True, "deleted": str(line_id)}


def reapply_rules(session: Session, actor, batch_id: uuid.UUID) -> dict:
    """B36–B39: match the current rules against ONLY the batch's still-
    unassigned lines (assigned rows are never touched), stamping them
    orange (auto_assigned TRUE). Early-exit flashes for no-rules /
    no-unassigned. Newest-rule-wins."""
    _require_owner(actor, "Only an owner can re-apply rules.")
    batch = _get_batch(session, actor, batch_id)
    rules = _load_rules(session, actor)
    if not rules:
        return {"reapplied": 0, "message": "No auto-assign rules to apply."}
    unassigned = list(
        session.scalars(
            select(CcExpenseLine).where(
                CcExpenseLine.batch_id == batch.id,
                CcExpenseLine.project_id.is_(None),
            )
        ).all()
    )
    if not unassigned:
        return {"reapplied": 0, "message": "No unassigned lines to match."}

    changed = 0
    for line in unassigned:
        match = find_auto_rule_match(line.description, rules)
        if match is None:
            continue
        line.project_id = uuid.UUID(match.project_id)
        line.assignment_description = match.assignment_description
        line.category = match.category
        line.auto_assigned = True
        changed += 1

    write_audit(
        session,
        actor,
        action="cc.reapply_rules",
        entity_type="cc_batch",
        entity_id=batch.id,
        after={"reapplied": changed},
    )
    session.commit()
    if changed == 0:
        message = "No new matches found."
    else:
        message = f"Re-applied rules: {changed} line{'s' if changed != 1 else ''} auto-assigned."
    return {"reapplied": changed, "message": message}


# ------------------------------------------------------- rules & keywords


def rules_payload(session: Session, actor) -> dict:
    """B48–B53 data: highlight keywords (keyword ASC) + auto-rules
    (created_at DESC == precedence) with project labels, the picker
    options, and the rule-category datalist source."""
    _require_owner(actor, "Owner only.")
    keyword_rows = list(
        session.scalars(
            select(CcHighlightKeyword)
            .where(CcHighlightKeyword.tenant_id == actor.tenant_id)
            .order_by(CcHighlightKeyword.keyword)
        ).all()
    )
    rule_rows = list(
        session.scalars(
            select(CcAutoRule)
            .where(CcAutoRule.tenant_id == actor.tenant_id)
            .order_by(CcAutoRule.created_at.desc(), CcAutoRule.id)
        ).all()
    )
    labels = _project_labels(session, actor, {r.project_id for r in rule_rows})
    return {
        "keywords": [
            {
                "id": str(k.id),
                "keyword": k.keyword,
                "created_at": k.created_at.isoformat() if k.created_at else None,
            }
            for k in keyword_rows
        ],
        "rules": [
            {
                "id": str(r.id),
                "keyword": r.keyword,
                "project_id": str(r.project_id),
                "project_label": labels.get(r.project_id),
                "assignment_description": r.assignment_description,
                "category": r.category,
                "created_at": r.created_at.isoformat() if r.created_at else None,
            }
            for r in rule_rows
        ],
        "project_options": _project_options(session, actor),
        "rule_categories": sorted(
            {r.category for r in rule_rows if r.category}
        ),
    }


def _keyword_exists(session: Session, actor, keyword: str) -> bool:
    return (
        session.scalar(
            select(CcHighlightKeyword.id).where(
                CcHighlightKeyword.tenant_id == actor.tenant_id,
                func.lower(CcHighlightKeyword.keyword) == keyword.lower(),
            )
        )
        is not None
    )


def create_highlight_keyword(session: Session, actor, keyword: str | None) -> dict:
    """B50: case-insensitive-unique per tenant. Pre-checked so the
    friendly message (not a raw 23505) reaches the user."""
    _require_owner(actor, "Only an owner can add highlight keywords.")
    clean = (keyword or "").strip()
    if not clean:
        raise ServiceError(422, "KEYWORD_REQUIRED", "Keyword is required.")
    if _keyword_exists(session, actor, clean):
        raise ServiceError(
            409, "DUPLICATE_KEYWORD", f'"{clean}" is already a highlight keyword.'
        )
    kw = CcHighlightKeyword(
        id=uuid.uuid4(), tenant_id=actor.tenant_id, keyword=clean
    )
    session.add(kw)
    session.flush()
    write_audit(
        session,
        actor,
        action="cc.keyword_create",
        entity_type="cc_keyword",
        entity_id=kw.id,
        after={"keyword": clean},
    )
    session.commit()
    return {"ok": True, "id": str(kw.id), "message": f'Added highlight keyword "{clean}".'}


def delete_highlight_keyword(session: Session, actor, keyword_id: uuid.UUID) -> dict:
    """B50: delete by id + tenant. Revalidate only, no flash."""
    _require_owner(actor, "Only an owner can delete highlight keywords.")
    kw = session.scalars(
        select(CcHighlightKeyword).where(
            CcHighlightKeyword.id == keyword_id,
            CcHighlightKeyword.tenant_id == actor.tenant_id,
        )
    ).first()
    if kw is None:
        raise ServiceError(404, "KEYWORD_NOT_FOUND", "Highlight keyword not found.")
    session.delete(kw)
    write_audit(
        session,
        actor,
        action="cc.keyword_delete",
        entity_type="cc_keyword",
        entity_id=keyword_id,
        before={"keyword": kw.keyword},
    )
    session.commit()
    return {"ok": True, "deleted": str(keyword_id)}


def _rule_exists(
    session: Session, actor, keyword: str, *, exclude: uuid.UUID | None = None
) -> bool:
    stmt = select(CcAutoRule.id).where(
        CcAutoRule.tenant_id == actor.tenant_id,
        func.lower(CcAutoRule.keyword) == keyword.lower(),
    )
    if exclude is not None:
        stmt = stmt.where(CcAutoRule.id != exclude)
    return session.scalar(stmt) is not None


def create_auto_rule(session: Session, actor, payload: dict) -> dict:
    """B52: keyword + project required; project-tenant validated;
    case-insensitive-unique keyword per tenant."""
    _require_owner(actor, "Only an owner can add auto-assign rules.")
    keyword = (payload.get("keyword") or "").strip()
    if not keyword:
        raise ServiceError(422, "KEYWORD_REQUIRED", "Keyword is required.")
    if payload.get("project_id") in (None, ""):
        raise ServiceError(422, "PROJECT_REQUIRED", "Pick a project.")
    project_id = _validate_project(session, actor, payload.get("project_id"))
    if _rule_exists(session, actor, keyword):
        raise ServiceError(
            409, "DUPLICATE_RULE", f'A rule for "{keyword}" already exists.'
        )
    rule = CcAutoRule(
        id=uuid.uuid4(),
        tenant_id=actor.tenant_id,
        keyword=keyword,
        project_id=project_id,
        assignment_description=(payload.get("assignment_description") or "").strip() or None,
        category=(payload.get("category") or "").strip() or None,
    )
    session.add(rule)
    session.flush()
    write_audit(
        session,
        actor,
        action="cc.rule_create",
        entity_type="cc_rule",
        entity_id=rule.id,
        after={"keyword": keyword, "project_id": str(project_id)},
    )
    session.commit()
    return {"ok": True, "id": str(rule.id), "message": f'Added auto-assign rule for "{keyword}".'}


def update_auto_rule(session: Session, actor, rule_id: uuid.UUID, payload: dict) -> dict:
    """B53 + delta: same field rules as create AND re-validate the
    project-tenant membership (the legacy update skipped it)."""
    _require_owner(actor, "Only an owner can edit auto-assign rules.")
    rule = session.scalars(
        select(CcAutoRule).where(
            CcAutoRule.id == rule_id, CcAutoRule.tenant_id == actor.tenant_id
        )
    ).first()
    if rule is None:
        raise ServiceError(404, "RULE_NOT_FOUND", "Auto-assign rule not found.")
    keyword = (payload.get("keyword") or "").strip()
    if not keyword:
        raise ServiceError(422, "KEYWORD_REQUIRED", "Keyword is required.")
    if payload.get("project_id") in (None, ""):
        raise ServiceError(422, "PROJECT_REQUIRED", "Pick a project.")
    project_id = _validate_project(session, actor, payload.get("project_id"))
    if _rule_exists(session, actor, keyword, exclude=rule.id):
        raise ServiceError(
            409, "DUPLICATE_RULE", f'A rule for "{keyword}" already exists.'
        )
    rule.keyword = keyword
    rule.project_id = project_id
    rule.assignment_description = (payload.get("assignment_description") or "").strip() or None
    rule.category = (payload.get("category") or "").strip() or None
    write_audit(
        session,
        actor,
        action="cc.rule_update",
        entity_type="cc_rule",
        entity_id=rule.id,
        after={"keyword": keyword, "project_id": str(project_id)},
    )
    session.commit()
    return {"ok": True, "id": str(rule.id), "message": f'Updated rule for "{keyword}".'}


def delete_auto_rule(session: Session, actor, rule_id: uuid.UUID) -> dict:
    """B53: delete by id + tenant. Revalidate only, no flash."""
    _require_owner(actor, "Only an owner can delete auto-assign rules.")
    rule = session.scalars(
        select(CcAutoRule).where(
            CcAutoRule.id == rule_id, CcAutoRule.tenant_id == actor.tenant_id
        )
    ).first()
    if rule is None:
        raise ServiceError(404, "RULE_NOT_FOUND", "Auto-assign rule not found.")
    session.delete(rule)
    write_audit(
        session,
        actor,
        action="cc.rule_delete",
        entity_type="cc_rule",
        entity_id=rule_id,
        before={"keyword": rule.keyword},
    )
    session.commit()
    return {"ok": True, "deleted": str(rule_id)}