"""Phase-3 slice 3 (ADR #024): CC per-user batch privacy + the
port-to-expenses action.

* ``list_batches`` scopes to the EFFECTIVE user's own batches (Rob never
  sees rian's) + ownerless (NULL created_by) rows; every OTHER cc read/
  write stays tenant-scoped exactly as before.
* ``port_to_expenses`` moves assigned lines into the CURRENT tenant's
  expenses journal — ``source='manual'`` rows carrying the line's
  project/amount/category/date + the batch currency, ``paid_by`` = the
  caller's own party unless chosen. The ``cc:<line-id>`` provenance
  marker on ``expenses.external_invoice_id`` is the double-port guard
  (no schema change); deleting the ported expense frees the marker.

SQLite fixtures (house style): one tenant with TWO owner users, a second
tenant for isolation, a sheet-owned tenant for the write-block."""

import uuid
from datetime import date
from decimal import Decimal
from types import SimpleNamespace

import pytest
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app.models import (
    AuditLog,
    Base,
    CcExpenseBatch,
    CcExpenseLine,
    Expense,
    Party,
    Project,
    Tenant,
    User,
)
from app.services import books as books_svc
from app.services.cc import service as cc
from app.services.errors import ServiceError
from app.services.money import expenses as expenses_svc

APP = {"books_owner": "app", "default_currency": "CAD"}


@pytest.fixture()
def s():
    engine = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        yield session
    engine.dispose()


def _party(s, kind, name, slug):
    party = Party(kind=kind, name=name, slug=slug)
    s.add(party)
    s.flush()
    return party


def _actor(tenant_id, user, *, person_party_id=None, viewing_as_of=None):
    """An owner actor. ``viewing_as_of`` simulates a super admin (the
    REAL user) impersonating ``user`` (the EFFECTIVE user)."""
    return SimpleNamespace(
        tenant_id=tenant_id,
        user_id=user.id if user else None,
        real_user_id=(viewing_as_of.id if viewing_as_of else (user.id if user else None)),
        person_party_id=person_party_id,
        is_viewing_as=viewing_as_of is not None,
        allowed=lambda cap: True,
        capabilities={"can_see_income": True},
    )


def _worker(tenant_id):
    return SimpleNamespace(
        tenant_id=tenant_id,
        user_id=None,
        real_user_id=None,
        person_party_id=None,
        is_viewing_as=False,
        allowed=lambda cap: False,
        capabilities={"can_see_income": False},
    )


@pytest.fixture()
def fx(s):
    plusroi_party = _party(s, "org", "PlusROI", "plusroi")
    other_party = _party(s, "org", "Other", "other")
    sheet_party = _party(s, "org", "SheetCo", "sheetco")
    plusroi = Tenant(party_id=plusroi_party.id, settings=APP)
    other = Tenant(party_id=other_party.id, settings=APP)
    sheet = Tenant(party_id=sheet_party.id, settings={"books_owner": "sheet"})
    s.add_all([plusroi, other, sheet])
    s.flush()

    p_rian = _party(s, "person", "Rian Bowden", "rian-bowden")
    p_rob = _party(s, "person", "Rob", "rob")
    rian = User(idauth_username="rian", email="rian@rian.ca", person_party_id=p_rian.id)
    rob = User(idauth_username="rob", email="rob@plusroi.com", person_party_id=p_rob.id)
    s.add_all([rian, rob])
    s.flush()

    client = _party(s, "org", "Copernic", "copernic")
    project = Project(
        tenant_id=plusroi.id,
        operator_party_id=plusroi_party.id,
        client_party_id=client.id,
        name="Monthly PPC",
        status="active",
    )
    s.add(project)
    s.flush()

    return SimpleNamespace(
        plusroi=plusroi,
        plusroi_party=plusroi_party,
        other=other,
        sheet=sheet,
        rian=_actor(plusroi.id, rian, person_party_id=p_rian.id),
        rob=_actor(plusroi.id, rob, person_party_id=p_rob.id),
        rian_user=rian,
        rob_user=rob,
        p_rian=p_rian,
        p_rob=p_rob,
        project=project,
    )


def _batch(s, actor, name):
    return uuid.UUID(cc.create_batch(s, actor, name)["id"])


def _line(
    s,
    batch_id,
    index,
    *,
    project_id=None,
    amount="10.00",
    expense_date=date(2026, 7, 3),
    category="Client Advertising Spend",
    description="FACEBK *X",
    assignment_description="Facebook",
):
    line = CcExpenseLine(
        batch_id=batch_id,
        line_index=index,
        raw_line=f"raw {index}",
        expense_date=expense_date,
        description=description,
        amount=Decimal(amount) if amount is not None else None,
        project_id=project_id,
        assignment_description=assignment_description,
        category=category,
    )
    s.add(line)
    s.flush()
    return line


# ------------------------------------------------------ per-user privacy


def test_batch_list_is_per_user(s, fx):
    a = _batch(s, fx.rian, "Rian TD June")
    b = _batch(s, fx.rob, "Rob TD June")

    rian_rows = {r["id"] for r in cc.list_batches(s, fx.rian)}
    rob_rows = {r["id"] for r in cc.list_batches(s, fx.rob)}
    assert rian_rows == {str(a)}
    assert rob_rows == {str(b)}  # Rob NEVER sees rian's batch


def test_ownerless_batches_stay_visible_to_every_owner(s, fx):
    """A migrated/orphaned batch (created_by NULL) shows for everyone —
    hiding it from all owners would strand it."""
    legacy = CcExpenseBatch(
        tenant_id=fx.plusroi.id, name="Legacy import", currency="CAD", created_by=None
    )
    s.add(legacy)
    s.commit()
    assert str(legacy.id) in {r["id"] for r in cc.list_batches(s, fx.rian)}
    assert str(legacy.id) in {r["id"] for r in cc.list_batches(s, fx.rob)}


def test_only_the_list_scopes_other_reads_stay_tenant_wide(s, fx):
    """Per spec, ONLY the list is per-user: a direct batch open (shared
    URL) still works across users in the same tenant."""
    a = _batch(s, fx.rian, "Rian TD June")
    detail = cc.batch_detail(s, fx.rob, a)
    assert detail["batch"]["name"] == "Rian TD June"


def test_created_by_follows_the_effective_user_under_view_as(s, fx):
    """rian view-as Rob creates a batch → it is ROB's (else it would
    vanish from Rob's own list mid-demo). The audit row still records
    the real human via impersonated_by."""
    rian_as_rob = _actor(
        fx.plusroi.id, fx.rob_user, person_party_id=fx.p_rob.id,
        viewing_as_of=fx.rian_user,
    )
    batch_id = _batch(s, rian_as_rob, "Rob's TD (via view-as)")
    batch = s.get(CcExpenseBatch, batch_id)
    assert batch.created_by == fx.rob_user.id
    assert str(batch_id) in {r["id"] for r in cc.list_batches(s, fx.rob)}
    assert str(batch_id) not in {r["id"] for r in cc.list_batches(s, fx.rian)}
    audit = s.scalars(
        select(AuditLog).where(AuditLog.action == "cc.batch_create")
    ).one()
    assert audit.actor_user_id == fx.rob_user.id
    assert audit.impersonated_by == fx.rian_user.id


def test_cross_tenant_batch_still_404(s, fx):
    a = _batch(s, fx.rian, "Rian TD June")
    foreign = _actor(fx.other.id, fx.rob_user)
    with pytest.raises(ServiceError) as err:
        cc.batch_detail(s, foreign, a)
    assert err.value.code == "CC_BATCH_NOT_FOUND" and err.value.status == 404


# -------------------------------------------------------- port-to-expenses


def test_port_creates_manual_expense_rows(s, fx):
    batch_id = _batch(s, fx.rob, "Rob TD June")
    good = _line(s, batch_id, 0, project_id=fx.project.id, amount="14.97")
    _line(s, batch_id, 1, project_id=None, amount="5.00")  # unassigned
    _line(s, batch_id, 2, project_id=fx.project.id, amount=None)  # no amount
    s.commit()

    res = cc.port_to_expenses(s, fx.rob, batch_id, {})
    assert res["ported"] == 1
    assert res["skipped_unassigned"] == 1
    assert res["skipped_no_amount"] == 1
    assert res["skipped_already_ported"] == 0

    (row,) = res["rows"]
    assert row["project"] == "Monthly PPC"
    assert row["amount"] == "14.97"
    assert row["currency"] == "CAD"  # the batch currency
    assert row["category"] == "Client Advertising Spend"
    assert row["expense_date"] == "2026-07-03"
    assert row["paid_by"] == "Rob"  # the caller's own party by default
    assert row["paid_by_is_tenant"] is False  # fronted → reimbursed later
    assert row["notes"] == "Facebook"
    assert row["external_invoice_id"] == f"cc:{good.id}"
    assert row["settlement_month"] is None  # untagged → month-end apply

    expense = s.get(Expense, uuid.UUID(row["id"]))
    assert expense.tenant_id == fx.plusroi.id
    assert expense.source == "manual"
    assert expense.created_by == fx.rob_user.id

    # …and it shows in the SAME ledger the Books UI reads.
    listed = books_svc.list_expenses(s, fx.rob)
    assert listed["total_count"] == 1
    assert listed["rows"][0]["id"] == row["id"]


def test_port_is_idempotent_no_double_port(s, fx):
    batch_id = _batch(s, fx.rob, "Rob TD June")
    _line(s, batch_id, 0, project_id=fx.project.id, amount="14.97")
    s.commit()

    assert cc.port_to_expenses(s, fx.rob, batch_id, {})["ported"] == 1
    with pytest.raises(ServiceError) as err:
        cc.port_to_expenses(s, fx.rob, batch_id, {})
    assert err.value.code == "NOTHING_TO_PORT" and err.value.status == 409
    assert s.scalar(select(func.count()).select_from(Expense)) == 1

    # a line assigned AFTER the first port comes through in wave two.
    _line(s, batch_id, 1, project_id=fx.project.id, amount="246.11",
          description="GOOGLE *ADS", assignment_description="Google")
    s.commit()
    res = cc.port_to_expenses(s, fx.rob, batch_id, {})
    assert res["ported"] == 1 and res["skipped_already_ported"] == 1
    assert s.scalar(select(func.count()).select_from(Expense)) == 2


def test_port_selected_lines_only(s, fx):
    batch_id = _batch(s, fx.rob, "Rob TD June")
    keep = _line(s, batch_id, 0, project_id=fx.project.id, amount="1.00")
    skip = _line(s, batch_id, 1, project_id=fx.project.id, amount="2.00")
    s.commit()
    res = cc.port_to_expenses(s, fx.rob, batch_id, {"line_ids": [str(keep.id)]})
    assert res["ported"] == 1
    marker = s.scalar(select(Expense.external_invoice_id))
    assert marker == f"cc:{keep.id}"

    with pytest.raises(ServiceError) as err:
        cc.port_to_expenses(s, fx.rob, batch_id, {"line_ids": [str(uuid.uuid4())]})
    assert err.value.code == "CC_LINE_NOT_FOUND" and err.value.status == 404
    assert skip is not None


def test_port_paid_by_override(s, fx):
    batch_id = _batch(s, fx.rob, "Rob TD June")
    _line(s, batch_id, 0, project_id=fx.project.id)
    s.commit()
    res = cc.port_to_expenses(
        s, fx.rob, batch_id, {"paid_by_party_id": str(fx.plusroi_party.id)}
    )
    assert res["rows"][0]["paid_by"] == "PlusROI"
    assert res["rows"][0]["paid_by_is_tenant"] is True

    with pytest.raises(ServiceError) as err:
        cc.port_to_expenses(
            s, fx.rob, batch_id, {"paid_by_party_id": str(uuid.uuid4())}
        )
    assert err.value.code == "PARTY_NOT_FOUND"


def test_deleting_the_ported_expense_frees_the_marker(s, fx):
    batch_id = _batch(s, fx.rob, "Rob TD June")
    _line(s, batch_id, 0, project_id=fx.project.id)
    s.commit()
    row = cc.port_to_expenses(s, fx.rob, batch_id, {})["rows"][0]
    expenses_svc.delete_expense(s, fx.rob, uuid.UUID(row["id"]))
    res = cc.port_to_expenses(s, fx.rob, batch_id, {})
    assert res["ported"] == 1  # deliberately re-portable after delete


def test_port_gates_and_tenant_isolation(s, fx):
    batch_id = _batch(s, fx.rob, "Rob TD June")
    _line(s, batch_id, 0, project_id=fx.project.id)
    s.commit()

    with pytest.raises(ServiceError) as err:
        cc.port_to_expenses(s, _worker(fx.plusroi.id), batch_id, {})
    assert err.value.status == 403

    with pytest.raises(ServiceError) as err:
        cc.port_to_expenses(s, _actor(fx.sheet.id, fx.rian_user), batch_id, {})
    assert err.value.code == "SHEET_OWNED_READ_ONLY" and err.value.status == 409

    with pytest.raises(ServiceError) as err:
        cc.port_to_expenses(s, _actor(fx.other.id, fx.rob_user), batch_id, {})
    assert err.value.code == "CC_BATCH_NOT_FOUND" and err.value.status == 404


def test_port_is_audited(s, fx):
    batch_id = _batch(s, fx.rob, "Rob TD June")
    line = _line(s, batch_id, 0, project_id=fx.project.id)
    s.commit()
    row = cc.port_to_expenses(s, fx.rob, batch_id, {})["rows"][0]
    audit = s.scalars(
        select(AuditLog).where(AuditLog.action == "cc.port_to_expenses")
    ).one()
    assert audit.after["ported"] == {str(line.id): row["id"]}
    assert audit.after["skipped_already_ported"] == 0