"""services/projects_plus — the PlusROI-shaped Projects write surface
(phase-3 slice 2a, docs/design/plusroi-forms.md §3).

A **project** is a `projects` row (the entity) + its 1:1 `project_billing`
profile + EITHER a recurrence (recurring / cycle / timing / default
amount) OR a `project_milestones` schedule (deposit / milestone / final).
It is created **attached to a client** and **inherits** that client's
commercial defaults (currency / terms / commission / invoice timing) from
the tenant's `client_profiles` row — editable per project.

This is the owner-facing rich create/edit; the bare
`resolve_or_create_project` (imports) still exists for the entity graph.
Both write the SAME `projects` table (uniqueness stays
`(tenant, client_party, lower(name))`).

Every mutation is **owner-gated** (`can_transfer`), refused on
**sheet-owned** books (409 `SHEET_OWNED_READ_ONLY`), **tenant-scoped**,
and **audited** — the shared `services.plusroi_gates` gate, EXACTLY as the
slice-1 Clients/Vendors forms.

Recurrence model (alembic 0008, ADR #022): a project is recurring XOR
milestone-scheduled. Recurring carries `is_recurring=true`, `cycle_months`
(1 monthly / 3 quarterly / 12 annual), `billing_timing`
(`month_end`|`mid_month`) and a default `amount`. Non-recurring carries
`project_milestones` rows and leaves `amount` NULL. A milestone that has
already generated an invoice (`generated_invoice_id`) is PRESERVED across
a schedule edit — only un-generated milestones are replaced.
"""

from __future__ import annotations

import re
import uuid
from decimal import Decimal, InvalidOperation

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

from app.models import (
    ClientProfile,
    Engagement,
    Party,
    Project,
    ProjectBilling,
    ProjectMilestone,
    Tenant,
)
from app.services.audit import write_audit
from app.services.errors import ServiceError
from app.services.plusroi_gates import (
    require_owner,
    require_tenant,
    require_writable_books,
)

#: Currencies a project may be billed in (mirror of clients.ALLOWED_CURRENCIES;
#: kept local so the module is self-contained — 07 #21 all-CAD default + USD).
ALLOWED_CURRENCIES = ("CAD", "USD", "EUR", "GBP")

#: The clean billing-timing vocabulary (alembic 0008 CHECK).
BILLING_TIMINGS = ("month_end", "mid_month")

#: Project lifecycle (parity with the projects table CHECK).
PROJECT_STATUSES = ("active", "archived")


# --------------------------------------------------------------- parsing


def _clean_text(raw) -> str | None:
    if raw is None:
        return None
    trimmed = str(raw).strip()
    return trimmed or None


def _require_name(raw) -> str:
    name = (raw or "").strip()
    if not name:
        raise ServiceError(422, "NAME_REQUIRED", "A project name is required.")
    return name


def _clean_currency(raw, default: str) -> str:
    code = (raw or "").strip().upper() or default
    if code not in ALLOWED_CURRENCIES:
        raise ServiceError(
            422,
            "BAD_CURRENCY",
            f"Currency must be one of {', '.join(ALLOWED_CURRENCIES)}.",
            f"Got {raw!r}.",
        )
    return code


def _parse_money(raw, label: str) -> Decimal | None:
    """A 2dp money field: blank/None -> None; else a finite, non-negative
    number rounded to cents."""
    if raw is None or str(raw).strip() == "":
        return None
    try:
        value = Decimal(re.sub(r"[$,\s]", "", str(raw)))
    except (InvalidOperation, ValueError) as exc:
        raise ServiceError(
            422, "BAD_AMOUNT", f"{label} must be a number, or empty."
        ) from exc
    if not value.is_finite() or value < 0:
        raise ServiceError(
            422, "BAD_AMOUNT", f"{label} must be a non-negative number, or empty."
        )
    return value.quantize(Decimal("0.01"))


def _parse_pct(raw) -> Decimal | None:
    if raw is None or str(raw).strip() == "":
        return None
    try:
        value = Decimal(re.sub(r"[%,\s]", "", str(raw)))
    except (InvalidOperation, ValueError) as exc:
        raise ServiceError(
            422, "BAD_COMMISSION", "Commission % must be a number, or empty."
        ) from exc
    if not value.is_finite():
        raise ServiceError(
            422, "BAD_COMMISSION", "Commission % must be a number, or empty."
        )
    return value.quantize(Decimal("0.0001"))


def _parse_cycle(raw) -> int | None:
    """Recurrence period in months. Blank -> None; else a positive int
    (1..120 — a sane ceiling; 1/3/12 are the usual monthly/quarterly/annual)."""
    if raw is None or str(raw).strip() == "":
        return None
    try:
        n = int(str(raw).strip())
    except (TypeError, ValueError) as exc:
        raise ServiceError(
            422, "BAD_CYCLE", "Billing cycle must be a whole number of months."
        ) from exc
    if n < 1 or n > 120:
        raise ServiceError(
            422, "BAD_CYCLE", "Billing cycle must be between 1 and 120 months."
        )
    return n


def _parse_timing(raw, *, required: bool) -> str | None:
    v = (raw or "").strip().lower()
    if not v:
        if required:
            raise ServiceError(
                422, "TIMING_REQUIRED", "Choose month-end or mid-month billing."
            )
        return None
    if v not in BILLING_TIMINGS:
        raise ServiceError(
            422,
            "BAD_TIMING",
            "Billing timing must be month_end or mid_month.",
            f"Got {raw!r}.",
        )
    return v


def _clean_status(raw) -> str:
    v = (raw or "active").strip().lower()
    if v not in PROJECT_STATUSES:
        raise ServiceError(
            422, "BAD_STATUS", "Status must be active or archived."
        )
    return v


def _validate_party(session: Session, raw) -> uuid.UUID | None:
    """Resolve an optional party-id reference (the commission earner)."""
    if raw in (None, ""):
        return None
    try:
        party_id = uuid.UUID(str(raw))
    except ValueError as exc:
        raise ServiceError(
            422, "BAD_PARTY", "commission_party_id must be a UUID."
        ) from exc
    if session.scalar(select(Party.id).where(Party.id == party_id)) is None:
        raise ServiceError(404, "PARTY_NOT_FOUND", "Selected commission party not found.")
    return party_id


def _parse_milestones(raw) -> list[tuple[str, Decimal]]:
    """A non-recurring schedule: a list of {label, amount}. Fully-blank
    repeater rows are dropped; a row with one of the two is a hard error.
    At least one milestone is required (else make it recurring)."""
    out: list[tuple[str, Decimal]] = []
    for row in raw or []:
        if not isinstance(row, dict):
            continue
        label = (row.get("label") or "").strip()
        amount_raw = row.get("amount")
        blank_amount = amount_raw is None or str(amount_raw).strip() == ""
        if not label and blank_amount:
            continue
        if not label:
            raise ServiceError(
                422, "MILESTONE_LABEL_REQUIRED", "Each milestone needs a label."
            )
        amount = _parse_money(amount_raw, "Milestone amount")
        if amount is None:
            raise ServiceError(
                422, "MILESTONE_AMOUNT_REQUIRED", "Each milestone needs an amount."
            )
        out.append((label, amount))
    if not out:
        raise ServiceError(
            422,
            "MILESTONES_REQUIRED",
            "Add at least one milestone, or make the project recurring.",
        )
    return out


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


def _tenant_party_id(session: Session, tenant_id: uuid.UUID) -> uuid.UUID:
    party_id = session.scalar(select(Tenant.party_id).where(Tenant.id == tenant_id))
    if party_id is None:
        raise ServiceError(400, "NO_TENANT", "Tenant has no party.")
    return party_id


def _tenant_currency(session: Session, tenant_id: uuid.UUID) -> str:
    settings = session.scalar(select(Tenant.settings).where(Tenant.id == tenant_id))
    if isinstance(settings, dict):
        code = settings.get("default_currency")
        if isinstance(code, str) and code.strip():
            return code.strip().upper()
    return "CAD"


def _require_client(
    session: Session, tenant_id: uuid.UUID, raw
) -> tuple[uuid.UUID, ClientProfile]:
    """The project's client MUST be a client of THIS tenant (a
    ``client_profiles`` row) — that's the tenant-scope anchor AND the
    inheritance source."""
    if raw in (None, ""):
        raise ServiceError(422, "CLIENT_REQUIRED", "Pick a client for this project.")
    try:
        cid = uuid.UUID(str(raw))
    except ValueError as exc:
        raise ServiceError(422, "BAD_CLIENT", "client_party_id must be a UUID.") from exc
    profile = session.get(ClientProfile, (tenant_id, cid))
    if profile is None:
        raise ServiceError(
            404,
            "CLIENT_NOT_FOUND",
            "That client isn't one of these books' clients.",
            "Create the client first (Clients), then attach the project.",
        )
    return cid, profile


def _ensure_direct_client_edge(
    session: Session,
    tenant_id: uuid.UUID,
    operator_party_id: uuid.UUID,
    client_party_id: uuid.UUID,
) -> None:
    """Idempotently ensure the tenant->client ``direct_client`` engagement
    (the client form already made it; defensive so a project always sits on
    a resolvable edge)."""
    existing = session.scalar(
        select(Engagement.id).where(
            Engagement.tenant_id == tenant_id,
            Engagement.type == "direct_client",
            Engagement.party_a_id == operator_party_id,
            Engagement.party_b_id == client_party_id,
            Engagement.ended_on.is_(None),
        )
    )
    if existing is None:
        session.add(
            Engagement(
                tenant_id=tenant_id,
                type="direct_client",
                party_a_id=operator_party_id,
                party_b_id=client_party_id,
                role_a="operator",
                role_b="client",
            )
        )
        session.flush()


def _name_clash(
    session: Session,
    tenant_id: uuid.UUID,
    client_party_id: uuid.UUID,
    name: str,
    *,
    exclude_id: uuid.UUID | None = None,
) -> bool:
    """Uniqueness is (tenant, client party, lower(name)) — judgment #33."""
    q = select(Project.id).where(
        Project.tenant_id == tenant_id,
        Project.client_party_id == client_party_id,
        func.lower(Project.name) == name.lower(),
    )
    if exclude_id is not None:
        q = q.where(Project.id != exclude_id)
    return session.scalar(q) is not None


def _num(value) -> str | None:
    return str(value) if value is not None else None


def _party_name(session: Session, party_id: uuid.UUID | None) -> str | None:
    if party_id is None:
        return None
    return session.scalar(select(Party.name).where(Party.id == party_id))


def _milestone_dto(m: ProjectMilestone) -> dict:
    return {
        "id": str(m.id),
        "label": m.label,
        "amount": _num(m.amount),
        "currency": m.currency,
        "sort_order": m.sort_order,
        "generated_invoice_id": (
            str(m.generated_invoice_id) if m.generated_invoice_id is not None else None
        ),
    }


def _load_milestones(session: Session, project_id: uuid.UUID) -> list[ProjectMilestone]:
    return list(
        session.scalars(
            select(ProjectMilestone)
            .where(ProjectMilestone.project_id == project_id)
            .order_by(ProjectMilestone.sort_order, ProjectMilestone.label)
        ).all()
    )


def _project_row(
    session: Session,
    project: Project,
    billing: ProjectBilling | None,
    client_name: str | None,
    milestones: list[ProjectMilestone],
) -> dict:
    commission_party_id = billing.commission_party_id if billing is not None else None
    total = sum((m.amount for m in milestones), Decimal("0")) if milestones else None
    return {
        "project_id": str(project.id),
        "name": project.name,
        "status": project.status,
        "description": project.notes,
        "client": {"id": str(project.client_party_id), "name": client_name},
        "currency": billing.currency if billing is not None else None,
        "terms": billing.terms if billing is not None else None,
        "invoice_timing": billing.invoice_timing if billing is not None else None,
        "is_recurring": bool(billing.is_recurring) if billing is not None else False,
        "cycle_months": billing.cycle_months if billing is not None else None,
        "billing_timing": billing.billing_timing if billing is not None else None,
        "amount": _num(billing.amount) if billing is not None else None,
        "default_commission_pct": (
            _num(billing.default_commission_pct) if billing is not None else None
        ),
        "commission_party": (
            {
                "id": str(commission_party_id),
                "name": _party_name(session, commission_party_id),
            }
            if commission_party_id is not None
            else None
        ),
        "milestone_count": len(milestones),
        "milestones_total": _num(total),
    }


def _project_image(
    session: Session, project: Project, billing: ProjectBilling
) -> dict:
    """JSON-serializable image of a project's mutable billing state — the
    audit before/after payload (incl. the milestone schedule)."""
    milestones = _load_milestones(session, project.id)
    return {
        "name": project.name,
        "status": project.status,
        "description": project.notes,
        "client_party_id": str(project.client_party_id),
        "currency": billing.currency,
        "terms": billing.terms,
        "invoice_timing": billing.invoice_timing,
        "is_recurring": bool(billing.is_recurring),
        "cycle_months": billing.cycle_months,
        "billing_timing": billing.billing_timing,
        "amount": _num(billing.amount),
        "default_commission_pct": _num(billing.default_commission_pct),
        "commission_party_id": (
            str(billing.commission_party_id)
            if billing.commission_party_id is not None
            else None
        ),
        "milestones": [(m.label, _num(m.amount)) for m in milestones],
    }


def _write_milestones(
    session: Session,
    project_id: uuid.UUID,
    milestones: list[tuple[str, Decimal]],
    currency: str | None,
    *,
    start_order: int,
) -> None:
    for i, (label, amount) in enumerate(milestones):
        session.add(
            ProjectMilestone(
                project_id=project_id,
                label=label,
                amount=amount,
                currency=currency,
                sort_order=start_order + i,
            )
        )
    session.flush()


def _replace_ungenerated_milestones(
    session: Session,
    project_id: uuid.UUID,
    milestones: list[tuple[str, Decimal]],
    currency: str | None,
) -> None:
    """Replace the schedule, PRESERVING any milestone that already generated
    an invoice (generated_invoice_id set) — only un-generated rows are
    swapped. New rows are appended after the kept ones."""
    kept = session.scalar(
        select(func.count(ProjectMilestone.id)).where(
            ProjectMilestone.project_id == project_id,
            ProjectMilestone.generated_invoice_id.is_not(None),
        )
    )
    session.execute(
        delete(ProjectMilestone).where(
            ProjectMilestone.project_id == project_id,
            ProjectMilestone.generated_invoice_id.is_(None),
        )
    )
    _write_milestones(
        session, project_id, milestones, currency, start_order=int(kept or 0)
    )


def _drop_ungenerated_milestones(session: Session, project_id: uuid.UUID) -> None:
    session.execute(
        delete(ProjectMilestone).where(
            ProjectMilestone.project_id == project_id,
            ProjectMilestone.generated_invoice_id.is_(None),
        )
    )


# --------------------------------------------------------------- reads


def list_projects(
    session: Session,
    actor,
    *,
    recurring: bool | None = None,
    timing: str | None = None,
) -> dict:
    """This tenant's projects with their billing/recurrence shape. Optional
    filters ``recurring`` + ``timing`` (the batch month-end selectors §4).
    Owner-only, tenant-scoped."""
    require_owner(actor, "Only an owner can view projects.")
    tenant_id = require_tenant(actor)

    client = Party
    q = (
        select(Project, ProjectBilling, client.name)
        .join(
            ProjectBilling,
            ProjectBilling.project_id == Project.id,
            isouter=True,
        )
        .join(client, client.id == Project.client_party_id, isouter=True)
        .where(Project.tenant_id == tenant_id)
    )
    if recurring is not None:
        q = q.where(ProjectBilling.is_recurring.is_(recurring))
    if timing is not None:
        q = q.where(ProjectBilling.billing_timing == timing)
    q = q.order_by(func.lower(Project.name))
    rows = session.execute(q).all()

    project_ids = [p.id for p, _, _ in rows]
    counts = (
        dict(
            session.execute(
                select(
                    ProjectMilestone.project_id, func.count(ProjectMilestone.id)
                )
                .where(ProjectMilestone.project_id.in_(project_ids))
                .group_by(ProjectMilestone.project_id)
            ).all()
        )
        if project_ids
        else {}
    )
    totals = (
        dict(
            session.execute(
                select(
                    ProjectMilestone.project_id, func.sum(ProjectMilestone.amount)
                )
                .where(ProjectMilestone.project_id.in_(project_ids))
                .group_by(ProjectMilestone.project_id)
            ).all()
        )
        if project_ids
        else {}
    )
    commission_ids = {
        b.commission_party_id
        for _, b, _ in rows
        if b is not None and b.commission_party_id is not None
    }
    commission_names = (
        dict(
            session.execute(
                select(Party.id, Party.name).where(Party.id.in_(commission_ids))
            ).all()
        )
        if commission_ids
        else {}
    )

    out = []
    for project, billing, client_name in rows:
        count = int(counts.get(project.id, 0))
        total = totals.get(project.id)
        commission_party_id = (
            billing.commission_party_id if billing is not None else None
        )
        out.append(
            {
                "project_id": str(project.id),
                "name": project.name,
                "status": project.status,
                "description": project.notes,
                "client": {
                    "id": str(project.client_party_id),
                    "name": client_name,
                },
                "currency": billing.currency if billing is not None else None,
                "terms": billing.terms if billing is not None else None,
                "invoice_timing": (
                    billing.invoice_timing if billing is not None else None
                ),
                "is_recurring": (
                    bool(billing.is_recurring) if billing is not None else False
                ),
                "cycle_months": billing.cycle_months if billing is not None else None,
                "billing_timing": (
                    billing.billing_timing if billing is not None else None
                ),
                "amount": _num(billing.amount) if billing is not None else None,
                "default_commission_pct": (
                    _num(billing.default_commission_pct)
                    if billing is not None
                    else None
                ),
                "commission_party": (
                    {
                        "id": str(commission_party_id),
                        "name": commission_names.get(commission_party_id),
                    }
                    if commission_party_id is not None
                    else None
                ),
                "milestone_count": count,
                "milestones_total": _num(total),
            }
        )
    return {"rows": out}


def get_project(session: Session, actor, project_id: uuid.UUID) -> dict:
    """One project (billing shape + milestone schedule). A project not in
    THIS tenant is a clean 404 (tenant isolation)."""
    require_owner(actor, "Only an owner can view projects.")
    tenant_id = require_tenant(actor)

    project = session.scalar(
        select(Project).where(Project.id == project_id, Project.tenant_id == tenant_id)
    )
    if project is None:
        raise ServiceError(404, "PROJECT_NOT_FOUND", "Project not found.")
    billing = session.get(ProjectBilling, project_id)
    milestones = _load_milestones(session, project_id)
    return {
        "project": _project_row(
            session,
            project,
            billing,
            _party_name(session, project.client_party_id),
            milestones,
        ),
        "milestones": [_milestone_dto(m) for m in milestones],
    }


# --------------------------------------------------------------- writes


def create_project(session: Session, actor, payload: dict) -> dict:
    """Create a project attached to a client, inheriting the client's
    commercial defaults (editable), with EITHER a recurrence OR a milestone
    schedule. Owner-only, writable-books, tenant-scoped, audited."""
    require_owner(actor, "Only an owner can create projects.")
    require_writable_books(session, actor)
    tenant_id = require_tenant(actor)
    operator_party_id = _tenant_party_id(session, tenant_id)

    name = _require_name(payload.get("name"))
    description = _clean_text(payload.get("description"))
    client_party_id, profile = _require_client(
        session, tenant_id, payload.get("client_party_id")
    )

    # Inherit the client profile's defaults; a provided (non-blank) value
    # overrides. Currency falls back through profile -> tenant currency.
    raw_currency = _clean_text(payload.get("currency")) or profile.currency
    currency = _clean_currency(raw_currency, _tenant_currency(session, tenant_id))
    terms = _clean_text(payload.get("terms")) or profile.terms
    invoice_timing = _clean_text(payload.get("invoice_timing")) or profile.invoice_timing
    if _clean_text(payload.get("default_commission_pct")) is not None:
        commission_pct = _parse_pct(payload.get("default_commission_pct"))
    else:
        commission_pct = profile.default_commission_pct
    if _clean_text(payload.get("commission_party_id")) is not None:
        commission_party_id = _validate_party(session, payload.get("commission_party_id"))
    else:
        commission_party_id = profile.commission_party_id

    is_recurring = bool(payload.get("is_recurring", False))
    if is_recurring:
        cycle_months = _parse_cycle(payload.get("cycle_months")) or 1
        billing_timing = _parse_timing(payload.get("billing_timing"), required=True)
        amount = _parse_money(payload.get("amount"), "Amount")
        if amount is None:
            raise ServiceError(
                422,
                "AMOUNT_REQUIRED",
                "A recurring project needs a default amount.",
            )
        milestones: list[tuple[str, Decimal]] = []
    else:
        cycle_months = None
        billing_timing = None
        amount = None
        milestones = _parse_milestones(payload.get("milestones"))

    if _name_clash(session, tenant_id, client_party_id, name):
        raise ServiceError(
            409,
            "PROJECT_EXISTS",
            f'"{name}" already exists for this client.',
            "Open the existing project to edit it, or use a different name.",
        )

    _ensure_direct_client_edge(session, tenant_id, operator_party_id, client_party_id)

    project = Project(
        tenant_id=tenant_id,
        operator_party_id=operator_party_id,
        client_party_id=client_party_id,
        name=name,
        status="active",
        notes=description,
    )
    session.add(project)
    session.flush()

    billing = ProjectBilling(
        project_id=project.id,
        billing_type="recurring" if is_recurring else "milestone",
        amount=amount,
        currency=currency,
        invoice_timing=invoice_timing,
        terms=terms,
        default_commission_pct=commission_pct,
        commission_party_id=commission_party_id,
        is_recurring=is_recurring,
        cycle_months=cycle_months,
        billing_timing=billing_timing,
    )
    session.add(billing)
    session.flush()

    if milestones:
        _write_milestones(session, project.id, milestones, currency, start_order=0)

    write_audit(
        session,
        actor,
        action="projects.create_plus",
        entity_type="project",
        entity_id=project.id,
        after=_project_image(session, project, billing),
    )
    session.commit()

    return {
        "ok": True,
        "row": _project_row(
            session,
            project,
            billing,
            _party_name(session, client_party_id),
            _load_milestones(session, project.id),
        ),
    }


def update_project(session: Session, actor, project_id: uuid.UUID, patch: dict) -> dict:
    """PATCH a project: only the SET keys change. Identity/default keys touch
    projects + project_billing; the recurrence keys switch the billing shape.
    Switching TO recurring drops un-generated milestones; a milestone schedule
    edit replaces only the un-generated rows (generated ones are preserved).
    Owner-only, writable-books, tenant-scoped, audited."""
    require_owner(actor, "Only an owner can edit projects.")
    require_writable_books(session, actor)
    tenant_id = require_tenant(actor)

    project = session.scalar(
        select(Project).where(Project.id == project_id, Project.tenant_id == tenant_id)
    )
    if project is None:
        raise ServiceError(404, "PROJECT_NOT_FOUND", "Project not found.")
    billing = session.get(ProjectBilling, project_id)
    if billing is None:  # defensive: a project should always have billing
        billing = ProjectBilling(project_id=project_id, is_recurring=False)
        session.add(billing)
        session.flush()

    before = _project_image(session, project, billing)

    # --- identity + inherited defaults -----------------------------------
    if "name" in patch:
        new_name = _require_name(patch.get("name"))
        if _name_clash(
            session,
            tenant_id,
            project.client_party_id,
            new_name,
            exclude_id=project_id,
        ):
            raise ServiceError(
                409,
                "PROJECT_EXISTS",
                f'"{new_name}" already exists for this client.',
            )
        project.name = new_name
    if "description" in patch:
        project.notes = _clean_text(patch.get("description"))
    if "status" in patch:
        project.status = _clean_status(patch.get("status"))
    if "currency" in patch:
        billing.currency = _clean_currency(
            patch.get("currency"), _tenant_currency(session, tenant_id)
        )
    if "terms" in patch:
        billing.terms = _clean_text(patch.get("terms"))
    if "invoice_timing" in patch:
        billing.invoice_timing = _clean_text(patch.get("invoice_timing"))
    if "default_commission_pct" in patch:
        billing.default_commission_pct = _parse_pct(patch.get("default_commission_pct"))
    if "commission_party_id" in patch:
        billing.commission_party_id = _validate_party(
            session, patch.get("commission_party_id")
        )

    # --- recurrence shape -------------------------------------------------
    toggling = "is_recurring" in patch
    if toggling:
        billing.is_recurring = bool(patch.get("is_recurring"))

    if billing.is_recurring:
        if "cycle_months" in patch:
            billing.cycle_months = _parse_cycle(patch.get("cycle_months"))
        if toggling and billing.cycle_months is None:
            billing.cycle_months = 1  # sensible default on the switch
        if "billing_timing" in patch:
            billing.billing_timing = _parse_timing(
                patch.get("billing_timing"), required=False
            )
        if "amount" in patch:
            billing.amount = _parse_money(patch.get("amount"), "Amount")
        if toggling:
            # became recurring — the milestone schedule no longer applies;
            # drop only un-generated rows (generated invoices are history).
            billing.billing_type = "recurring"
            _drop_ungenerated_milestones(session, project_id)
    else:
        if toggling:
            # became non-recurring — clear the recurrence fields.
            billing.cycle_months = None
            billing.billing_timing = None
            billing.amount = None
            billing.billing_type = "milestone"
        if "milestones" in patch:
            _replace_ungenerated_milestones(
                session,
                project_id,
                _parse_milestones(patch.get("milestones")),
                billing.currency,
            )

    session.flush()
    after = _project_image(session, project, billing)

    write_audit(
        session,
        actor,
        action="projects.update_plus",
        entity_type="project",
        entity_id=project.id,
        before=before,
        after=after,
    )
    session.commit()

    return {
        "ok": True,
        "changed": before != after,
        "row": _project_row(
            session,
            project,
            billing,
            _party_name(session, project.client_party_id),
            _load_milestones(session, project_id),
        ),
    }