dodil/erp-brick-contracts

src/brick_contracts/infrastructure/public_door.py

"""Anti-corruption layer: the platform's public-link door (erp_platform.public).

The door holds the signing token (hashed; the secret is returned once and never stored), resolves it to
(contract, contract_id|signer_no) with its grant, checks expiry and revocation, audits every use, and calls this
brick's public use cases with a PUBLIC identity scoped to that one signer. contracts keeps only the link_id.

  SIGN  one per EXTERNAL signer per version sent; single use (claimed with link.use() inside the signing
        transaction when the link tables are on the same engine — tier 1, one bucket — else just before it)
  READ  the signed copy, one per EXTERNAL signer, issued when the executed copy exists (IWG: let signers download it)

SAME TRANSACTION when it can be: the link rows live in erp_public_link on the composition's bucket. When this brick's
connection is on that same engine, `conn` is passed so the send and its links — or the signature and the link's
claim — commit together. In the in-process scenario harness each brick has its own SQLite file and the links another,
so the calls run in their own short transactions; the rules hold either way (a claim is a plain INSERT).
"""
from datetime import timedelta
from typing import Optional

from erp_platform import public

SUBJECT = "contract"
BRICK = "contracts"


def links() -> public.Links:
    return public.Links(BRICK)


def shared(conn) -> Optional[object]:
    """`conn` if the link tables are on the same engine as this connection, else None."""
    if conn is None:
        return None
    target = public.CONNECT
    if target is None:
        try:
            from erp_platform import db                             # the app's own bucket: the same engine
        except Exception:                                           # noqa: BLE001 — no BUCKET: not the same engine
            return None
        return conn if getattr(db, "engine", None) is conn.engine else None
    return conn if getattr(target, "__self__", None) is conn.engine else None


def subject(contract_id: str, signer_no: int) -> str:
    return f"{contract_id}|{int(signer_no)}"


def issue(contract_id: str, signer_no: int, email: str, grant: str, *, days: int, actor: str, conn=None):
    return links().issue(SUBJECT, subject(contract_id, signer_no), email, grant, expires_in=timedelta(days=days),
                         actor=actor, conn=shared(conn))


def revoke(link_id: str, *, actor: str, reason: str, conn=None) -> bool:
    if not link_id:
        return False
    return links().revoke(link_id, actor=actor, reason=reason, conn=shared(conn))


def reissue(link_id: str, *, actor: str, conn=None):
    return links().reissue(link_id, actor=actor, conn=shared(conn))


def status(contract_id: str, signer_no: int) -> list:
    return links().status(SUBJECT, subject(contract_id, signer_no))


__all__ = ["links", "shared", "issue", "revoke", "reissue", "status", "subject", "SUBJECT", "BRICK"]