dodil/erp-brick-payments

DODIL ERP brick: payments

src/brick_payments/domain/refund.py

"""Refund: money paid back to a customer. Money going OUT needs an approval, always, by someone other than the
requester; the AI never moves money."""
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from brick_payments.domain import events
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO, money
from brick_payments.domain.readiness import Gate, fmt, met, refuse, unmet
from brick_payments.domain.values import RefundStatus


def credit_covers(amount: Decimal, credit: Decimal, currency: str) -> Gate:
    if amount > credit:
        return unmet("credit_covers", f"Refund {fmt(amount)}; the customer's unallocated credit in {currency} is {fmt(credit)}",
                     f"Refund at most {fmt(credit)}")
    return met("credit_covers")


def second_person(requested_by: str, approver: str, requested_sub: str = "", approver_ids: tuple = ()) -> Gate:
    """Never the same person under EITHER identity: the email, or the sub, of the requester is the approver's."""
    theirs = {x.strip().lower() for x in (requested_by, requested_sub) if x and x.strip()}
    mine = {x.strip().lower() for x in (approver, *approver_ids) if x and x.strip()}
    if theirs & mine:
        return unmet("second_person", "You requested this refund", "Another approver approves it")
    return met("second_person")


def paid_dated_ok(paid_on: date, approved_on: Optional[date], today: date) -> Gate:
    if paid_on > today or (approved_on is not None and paid_on < approved_on):
        return unmet("paid_dated_ok", f"Paid on {paid_on}, after today or before the approval ({approved_on})",
                     "Use the date on the bank's payment")
    return met("paid_dated_ok")


@dataclass
class Refund:
    refund_id: str
    bp_id: int
    currency: str
    amount: Decimal
    status: str = RefundStatus.REQUESTED.value
    requested_by: str = ""
    requested_at: Optional[datetime] = None
    approved_by: str = ""
    approved_at: Optional[datetime] = None
    approved_on: Optional[date] = None
    reason: str = ""
    bank_ref: str = ""
    paid_on: Optional[date] = None
    origin: str = ""
    requested_sub: str = ""                   # the requester's sub beside their email: either identity is the same person
    events: list = field(default_factory=list)

    @classmethod
    def request(cls, *, refund_id: str, bp_id: int, currency: str, amount, credit: Decimal, by: str,
                at: datetime, reason: str = "", sub: str = "") -> "Refund":
        amount = money(amount)
        if amount <= ZERO:
            raise RuleBroken("A refund is more than zero")
        currency = currency.upper()
        refuse(credit_covers(amount, credit, currency))
        r = cls(refund_id, bp_id, currency, amount, RefundStatus.REQUESTED.value, by, at, reason=reason,
                requested_sub=(sub or "").strip().lower())
        r.events.append(events.RefundRequested(refund_id, amount))
        return r

    def _from(self, *allowed: RefundStatus) -> None:
        if self.status not in [s.value for s in allowed]:
            raise RuleBroken(f"Refund {self.refund_id} is {self.status}; this step needs it to be "
                             f"{' or '.join(s.value for s in allowed)}")

    def approve(self, *, by: str, credit: Decimal, at: datetime, on: date, identities: tuple = ()) -> None:
        self._from(RefundStatus.REQUESTED)
        refuse(second_person(self.requested_by, by, self.requested_sub, identities))
        refuse(credit_covers(self.amount, credit, self.currency))
        self.status, self.approved_by, self.approved_at, self.approved_on = RefundStatus.APPROVED.value, by, at, on
        self.events.append(events.RefundApproved(self.refund_id, by))

    def reject(self, *, by: str, reason: str) -> None:
        self._from(RefundStatus.REQUESTED)
        if not reason.strip():
            raise RuleBroken("Rejecting a refund needs a reason. Say why")
        self.status, self.approved_by, self.reason = RefundStatus.REJECTED.value, by, reason.strip()
        self.events.append(events.RefundRejected(self.refund_id, by))

    def pay(self, *, paid_on: date, bank_ref: str, today: date) -> None:
        self._from(RefundStatus.APPROVED)
        if not bank_ref.strip():
            raise RuleBroken("Recording a refund paid needs the bank reference. Take it from the bank's payment advice")
        refuse(paid_dated_ok(paid_on, self.approved_on, today))
        self.status, self.paid_on, self.bank_ref = RefundStatus.PAID.value, paid_on, bank_ref.strip()
        self.events.append(events.RefundPaid(self.refund_id, paid_on.isoformat()))