""" merchant_onboarding CRUD — DB operations only, no business logic (matches this codebase's `crud.py` convention: pure DB access, no GP calls, no event dispatch — that all belongs in services.py). """ from __future__ import annotations from datetime import date, datetime, timezone from typing import Any, Dict, List, Optional from sqlalchemy import select from sqlalchemy.orm import Session from src.apps.merchant_onboarding.helpers.field_mapping import MAX_BANK_ACCOUNTS, OWNERSHIP_TYPES_SSN_EXEMPT from src.apps.merchant_onboarding.models.application import MerchantOnboardingApplication from src.apps.merchant_onboarding.models.owner import MerchantOnboardingOwner from src.apps.merchant_onboarding.models.account import MerchantOnboardingAccount from src.apps.merchant_onboarding.models.document import MerchantOnboardingDocument from src.apps.merchant_onboarding.models.api_call_log import OnboardingApiCallLog from src.apps.payment_providers.helpers.credentials import encrypt_credential def get_active_application(db: Session, merchant_id: int) -> Optional[MerchantOnboardingApplication]: """ Return the merchant's single active (non-soft-deleted) onboarding application, or None if it has never started one. "One active application per merchant" is enforced here by construction: every other function in this module reads/writes through this lookup rather than allowing callers to pass an arbitrary application_id, and create_application_row() below refuses to create a second one. """ stmt = select(MerchantOnboardingApplication).where( MerchantOnboardingApplication.merchant_id == merchant_id, MerchantOnboardingApplication.deleted_at.is_(None), ) return db.execute(stmt).scalar_one_or_none() def create_application_row(db: Session, merchant_id: int) -> MerchantOnboardingApplication: """ Create the (single) onboarding application row for a merchant. Raises ValueError if an active application already exists — callers (services.start_or_resume, services.restart_after_rejection) must soft-delete or otherwise resolve the existing row first; this function never silently reuses or overwrites one. """ existing = get_active_application(db, merchant_id) if existing is not None: raise ValueError( f"Merchant {merchant_id} already has an active onboarding application " f"(id={existing.id}) — one-active-application-per-merchant is enforced here." ) application = MerchantOnboardingApplication( merchant_id=merchant_id, status="draft", current_step="application_setup", section_state={}, ) db.add(application) db.flush() return application def soft_delete_application(db: Session, application: MerchantOnboardingApplication) -> None: """ Soft-delete an application row (services.restart_after_rejection's only caller). After this, get_active_application() returns None for the merchant, so a subsequent create_application_row() call succeeds instead of raising the "already has an active application" ValueError. """ from datetime import datetime, timezone application.deleted_at = datetime.now(timezone.utc) db.flush() def update_application( db: Session, application: MerchantOnboardingApplication, **fields ) -> MerchantOnboardingApplication: """ Generic field-setter for MerchantOnboardingApplication. Does not commit — callers control the transaction boundary (matches this codebase's get_db-autocommit-on-HTTP-request / manual-commit-in-Celery convention). Does not validate status-machine transitions itself — call helpers.status_machine.assert_transition() before passing a new `status` if the caller needs that guarantee (services.py does this). """ for key, value in fields.items(): if not hasattr(application, key): raise ValueError(f"MerchantOnboardingApplication has no field {key!r}") setattr(application, key, value) db.flush() return application def count_owners(db: Session, application_id: int) -> int: stmt = select(MerchantOnboardingOwner).where( MerchantOnboardingOwner.application_id == application_id, MerchantOnboardingOwner.deleted_at.is_(None), ) return len(db.execute(stmt).scalars().all()) def count_accounts(db: Session, application_id: int) -> int: stmt = select(MerchantOnboardingAccount).where( MerchantOnboardingAccount.application_id == application_id, MerchantOnboardingAccount.deleted_at.is_(None), ) return len(db.execute(stmt).scalars().all()) def count_documents(db: Session, application_id: int) -> int: stmt = select(MerchantOnboardingDocument).where( MerchantOnboardingDocument.application_id == application_id, MerchantOnboardingDocument.deleted_at.is_(None), ) return len(db.execute(stmt).scalars().all()) # --------------------------------------------------------------------------- # Accounts (PRD-HWONB-006) — list-upsert CRUD # --------------------------------------------------------------------------- def list_accounts(db: Session, application_id: int) -> List[MerchantOnboardingAccount]: """Active (non-soft-deleted) accounts for an application, in row-id order.""" stmt = ( select(MerchantOnboardingAccount) .where( MerchantOnboardingAccount.application_id == application_id, MerchantOnboardingAccount.deleted_at.is_(None), ) .order_by(MerchantOnboardingAccount.id.asc()) ) return list(db.execute(stmt).scalars().all()) def get_account( db: Session, application_id: int, account_id: int ) -> Optional[MerchantOnboardingAccount]: """A single active account row, scoped to `application_id` (no IDOR across applications).""" stmt = select(MerchantOnboardingAccount).where( MerchantOnboardingAccount.id == account_id, MerchantOnboardingAccount.application_id == application_id, MerchantOnboardingAccount.deleted_at.is_(None), ) return db.execute(stmt).scalar_one_or_none() def upsert_accounts( db: Session, application_id: int, accounts: List[Dict[str, Any]] ) -> List[MerchantOnboardingAccount]: """ List-upsert semantics (PRD-HWONB-006 §4): `accounts` is the FULL desired account list for this application. - An item carrying an existing row `id` updates that row in place. - An item with no `id` (or an unrecognized one) creates a new row. - Any existing active row whose `id` is NOT present in `accounts` is soft-deleted (`deleted_at`) — never hard-deleted, matching this codebase's universal soft-delete convention. Enforces the `<= MAX_BANK_ACCOUNTS` cap (CERT-confirmed real GP rule, C-23) as a `ValidationException` (422) here, BEFORE touching the DB — this is the single authoritative enforcement point (models/account.py deliberately has no UniqueConstraint that would turn this into an opaque DB IntegrityError instead of a clean field-level error). Routing/account numbers are Fernet-encrypted here (src.apps.payment_providers.helpers.credentials) before ever touching the DB — plaintext values never persist. Does not commit — caller (services.py) controls the transaction boundary, consistent with every other function in this module. """ from src.apps.payment_providers.helpers.credentials import encrypt_credential from src.core.exceptions import ValidationException if len(accounts) > MAX_BANK_ACCOUNTS: raise ValidationException( message=f"No more than {MAX_BANK_ACCOUNTS} bank accounts are allowed per application.", error={ "field": "accounts", "message": f"max_accounts_exceeded (limit={MAX_BANK_ACCOUNTS}, received={len(accounts)})", }, ) existing_by_id = {row.id: row for row in list_accounts(db, application_id)} incoming_ids = {item["id"] for item in accounts if item.get("id")} # Full-replace semantics: anything active but no longer present is removed. for existing_id, row in existing_by_id.items(): if existing_id not in incoming_ids: row.deleted_at = datetime.now(timezone.utc) result_rows: List[MerchantOnboardingAccount] = [] any_default_flagged = False for item in accounts: is_default = bool(item.get("defaultAccount")) if is_default: any_default_flagged = True usage_types = item.get("usageTypes") or [] account_id = item.get("id") row = existing_by_id.get(account_id) if account_id else None if row is not None: row.deleted_at = None # re-submission overrides a pending removal if item.get("routingNumber"): row.routing_number_enc = encrypt_credential(item["routingNumber"]) if item.get("accountNumber"): row.account_number_enc = encrypt_credential(item["accountNumber"]) # else: leave the existing encrypted value alone — "on file, # blank to keep it" (mirrors the same fix applied to owners' # ssn_enc in upsert_owners/_apply_owner_fields). row.account_type = item["accountType"] row.usage_types = usage_types row.is_default = is_default else: if not item.get("routingNumber") or not item.get("accountNumber"): raise ValidationException( message="routingNumber and accountNumber are required for a new bank account.", error={ "field": "accounts", "message": "routing_and_account_number_required_for_new_account", }, ) row = MerchantOnboardingAccount( application_id=application_id, routing_number_enc=encrypt_credential(item["routingNumber"]), account_number_enc=encrypt_credential(item["accountNumber"]), account_type=item["accountType"], usage_types=usage_types, is_default=is_default, ) db.add(row) result_rows.append(row) db.flush() # §1 — "first account defaults to true if omitted": if the submitted set # didn't flag any account as default, promote the first one. (Exactly-one # is already enforced at the schema layer for the >1-flagged case.) if not any_default_flagged and result_rows: for row in result_rows: row.is_default = False result_rows[0].is_default = True db.flush() return result_rows # ACH usage types per PRD-HWONB-006 — kept as a local literal set here rather # than importing from field_mapping since these two values are only ever # consumed by this one derived-field read for Products/Equipment. _ACH_USAGE_TYPES = {"ACHF", "ACHS"} def has_ach_usage_type(db: Session, application_id: int) -> bool: """ PRD-HWONB-009 §3 AC-5 — the Integrated product payload's `ach` boolean must reflect whether ANY of the application's (non-soft-deleted) bank accounts has an ACH usage type (`ACHF`/`ACHS`), never a hardcoded default. Read-only query against `merchant_onboarding_accounts`. """ stmt = select(MerchantOnboardingAccount.usage_types).where( MerchantOnboardingAccount.application_id == application_id, MerchantOnboardingAccount.deleted_at.is_(None), ) for (usage_types,) in db.execute(stmt).all(): if usage_types and _ACH_USAGE_TYPES.intersection(usage_types): return True return False # --------------------------------------------------------------------------- # Documents (PRD-HWONB-012) — merchant_onboarding_documents CRUD # --------------------------------------------------------------------------- def create_document_row( db: Session, application_id: int, *, doc_type: str, file_name: str, content_type: str, size_bytes: int, s3_key: str, ) -> MerchantOnboardingDocument: """Create a document row. No GP call, no validation — services.py's job.""" document = MerchantOnboardingDocument( application_id=application_id, doc_type=doc_type, file_name=file_name, content_type=content_type, size_bytes=size_bytes, s3_key=s3_key, ) db.add(document) db.flush() return document def get_document(db: Session, application_id: int, document_id: int) -> Optional[MerchantOnboardingDocument]: """ Scoped lookup — always filters by application_id so a caller can never fetch (or later delete) another application's document by guessing an id (closes the IDOR class of bug by construction, matching this module's router-level convention of never accepting a caller-supplied application_id directly). """ stmt = select(MerchantOnboardingDocument).where( MerchantOnboardingDocument.id == document_id, MerchantOnboardingDocument.application_id == application_id, MerchantOnboardingDocument.deleted_at.is_(None), ) return db.execute(stmt).scalar_one_or_none() def list_documents(db: Session, application_id: int) -> list[MerchantOnboardingDocument]: stmt = ( select(MerchantOnboardingDocument) .where( MerchantOnboardingDocument.application_id == application_id, MerchantOnboardingDocument.deleted_at.is_(None), ) .order_by(MerchantOnboardingDocument.created_at.asc()) ) return list(db.execute(stmt).scalars().all()) def update_document(db: Session, document: MerchantOnboardingDocument, **fields) -> MerchantOnboardingDocument: """Generic field-setter for MerchantOnboardingDocument — mirrors update_application().""" for key, value in fields.items(): if not hasattr(document, key): raise ValueError(f"MerchantOnboardingDocument has no field {key!r}") setattr(document, key, value) db.flush() return document def soft_delete_document(db: Session, document: MerchantOnboardingDocument) -> None: from datetime import datetime, timezone document.deleted_at = datetime.now(timezone.utc) db.flush() # --------------------------------------------------------------------------- # Owners (PRD-HWONB-007) — DB ops + the authoritative cap-enforcement layer. # # PRD-HWONB-001 §7.3: "Backend enforces role-count caps server-side as the # authoritative check — the frontend's live-disabling of checkboxes is a UX # nicety, not the source of truth." This module is that authoritative check. # # Caps enforced here (see OwnerCapViolation, raised with a clear message — # never a raw DB constraint violation, since none of these caps are backed # by a DB constraint by design, per owner.py's module docstring): # - >= 1, <= 5 owners total per application # - <= 2 with app_signer = true # - <= 2 with personal_guarantor = true # - <= 4 with beneficial_owner = true # - EXACTLY 1 with individual_with_control = true, unconditionally, for # every application regardless of ownershipType (corrected from v1's # "max 1" — PRD-HWONB-007 §1/§2). NOTE (judgment call, see this # Developer's report-back): §2 item 3 and AC-1 phrase the "at least one # individualWithControl" rule as specifically "Mandatory for # CO/CPUB/GE/FI/CNP ownership types". §1's corrected field-table text and # PRD-HWONB-001 §6.1's CRUD-enforcement note both state "exactly 1 across # all owners" with no ownership-type qualifier attached to that specific # sentence. This module treats the exactly-1 rule as UNCONDITIONAL (every # ownership type, not just those 5 codes) — the ownership-type clause is # read as emphasis/underwriting context, not a conditional gate. Sole # Proprietorship's forced-all-4-flags-true rule (§2 item 1) naturally # satisfies this too, since SOLE only ever has one owner. Revisit if CERT # ever shows GP accepting 0 individualWithControl owners for an # ownership type outside that list of 5. # - owner_percent must sum to exactly 100 across all owners # - every owner must be beneficial_owner=True and/or # individual_with_control=True — this is actually enforced per-row in # schemas/owners.py (OwnerItem model_validator), since it's checkable # one owner at a time; re-checked defensively here too since crud.py is # the authoritative layer per the PRD, and services.py could in # principle call upsert_owners() with plain dicts that never passed # through the Pydantic schema. # # ownershipType-dependent forced values / requirements (PRD-007 §2) — passed # in by services.py from `application.business_data["ownershipType"]` (the # business section's own JSON snapshot column; independent of whichever # schema the PRD-HWONB-003 developer ends up building): # - SOLE: exactly 1 owner; all 4 role flags forced True; owner_percent # forced 100 # - CNP (Non-Profit): the individual_with_control owner's owner_percent # forced to exactly 0 # - ssn is required unless ownershipType is in OWNERSHIP_TYPES_SSN_EXEMPT # (helpers/field_mapping.py, C-13) — not applicable when # non_us_citizen=True (ssn is a US-citizen-branch-only field) # --------------------------------------------------------------------------- MAX_OWNERS = 5 MAX_APP_SIGNERS = 2 MAX_PERSONAL_GUARANTORS = 2 MAX_BENEFICIAL_OWNERS = 4 REQUIRED_INDIVIDUAL_WITH_CONTROL = 1 class OwnerCapViolation(ValueError): """ Raised when an owner-list upsert would violate a role-count/percent cap or a ownershipType-dependent rule. Callers (services.py) catch this and turn `str(exc)` into a 422 ValidationException for the wizard — never a raw DB constraint violation, since these rules aren't DB constraints. """ def list_owners(db: Session, application_id: int) -> List[MerchantOnboardingOwner]: """All non-soft-deleted owner rows for an application, in creation order.""" stmt = ( select(MerchantOnboardingOwner) .where( MerchantOnboardingOwner.application_id == application_id, MerchantOnboardingOwner.deleted_at.is_(None), ) .order_by(MerchantOnboardingOwner.id) ) return list(db.execute(stmt).scalars().all()) def get_owner(db: Session, application_id: int, owner_id: int) -> Optional[MerchantOnboardingOwner]: """A single owner row, scoped to its application (closes IDOR by construction).""" stmt = select(MerchantOnboardingOwner).where( MerchantOnboardingOwner.id == owner_id, MerchantOnboardingOwner.application_id == application_id, MerchantOnboardingOwner.deleted_at.is_(None), ) return db.execute(stmt).scalar_one_or_none() # Fields on the incoming owner dict that map onto an *_enc column, never a # plaintext column, on the ORM model — mirrors the Fernet pattern used for # bank accounts (models/account.py: routing_number_enc/account_number_enc). _OWNER_ENCRYPTED_FIELD_MAP = {"ssn": "ssn_enc"} def _apply_owner_fields(row: MerchantOnboardingOwner, data: Dict[str, Any]) -> None: """Copy validated owner fields from `data` (snake_case) onto `row`, Fernet-encrypting ssn.""" for key, value in data.items(): if key == "id": continue if key in _OWNER_ENCRYPTED_FIELD_MAP: enc_attr = _OWNER_ENCRYPTED_FIELD_MAP[key] if value: setattr(row, enc_attr, encrypt_credential(str(value))) # else: leave the existing ssn_enc alone — None for a brand-new # row (no-op, defaults to None at construction), the # previously-stored value for an update (never overwrite a real # SSN with a blank resubmission — see _validate_ssn_requirement). continue if not hasattr(row, key): # Unknown key (e.g. a stray schema field with no matching column) — # skip rather than crash; the Pydantic schema is the shape gate. continue if key == "dob" and isinstance(value, str): # `services.save_owners` builds `owners_dicts` via # `model_dump(mode="json")`, which serializes the Pydantic `date` # field to an ISO string — parse it back so `row.dob` matches its # mapped `Date` column type. Otherwise `_owner_to_gp_payload`'s # `row.dob.isoformat()` crashes with AttributeError on the str. value = date.fromisoformat(value) setattr(row, key, value) def _apply_ownership_type_forcing(owners: List[Dict[str, Any]], ownership_type: Optional[str]) -> None: """ Mutates `owners` in place applying PRD-007 §2's forced-value rules. Must run BEFORE `_validate_owner_caps`/`_validate_ssn_requirement` so those checks see the already-forced values (e.g. SOLE's 100% and CNP's 0% are not "suggestions" — they're what actually gets saved/pushed). """ if not ownership_type: return if ownership_type == "SOLE": if len(owners) != 1: raise OwnerCapViolation( "Sole Proprietorship (ownershipType=SOLE) allows exactly one owner." ) owners[0]["app_signer"] = True owners[0]["personal_guarantor"] = True owners[0]["beneficial_owner"] = True owners[0]["individual_with_control"] = True owners[0]["owner_percent"] = 100 if ownership_type == "CNP": for owner in owners: if owner.get("individual_with_control"): owner["owner_percent"] = 0 def _validate_owner_caps(owners: List[Dict[str, Any]]) -> None: if not owners: raise OwnerCapViolation("At least one owner is required.") if len(owners) > MAX_OWNERS: raise OwnerCapViolation(f"Maximum {MAX_OWNERS} owners allowed.") for idx, owner in enumerate(owners): if not owner.get("beneficial_owner") and not owner.get("individual_with_control"): raise OwnerCapViolation( f"Owner {idx + 1}: each owner must be marked as Beneficial Owner " "and/or Individual with Control." ) app_signers = sum(1 for o in owners if o.get("app_signer")) personal_guarantors = sum(1 for o in owners if o.get("personal_guarantor")) beneficial_owners = sum(1 for o in owners if o.get("beneficial_owner")) individual_with_control = sum(1 for o in owners if o.get("individual_with_control")) if app_signers > MAX_APP_SIGNERS: raise OwnerCapViolation(f"Maximum {MAX_APP_SIGNERS} owners may be marked as Application Signer.") if personal_guarantors > MAX_PERSONAL_GUARANTORS: raise OwnerCapViolation( f"Maximum {MAX_PERSONAL_GUARANTORS} owners may be marked as Personal Guarantor." ) if beneficial_owners > MAX_BENEFICIAL_OWNERS: raise OwnerCapViolation(f"Maximum {MAX_BENEFICIAL_OWNERS} owners may be marked as Beneficial Owner.") if individual_with_control != REQUIRED_INDIVIDUAL_WITH_CONTROL: raise OwnerCapViolation( "Exactly one owner must be marked as Individual with Control " f"(currently {individual_with_control})." ) total_percent = sum(int(o.get("owner_percent") or 0) for o in owners) if total_percent != 100: raise OwnerCapViolation( f"Owner percentages must total 100% across all owners (currently {total_percent}%)." ) def _validate_ssn_requirement( owners: List[Dict[str, Any]], ownership_type: Optional[str], existing_by_id: Dict[int, MerchantOnboardingOwner], ) -> None: """ PRD-007 §1/§2.4 (C-13) — ssn optional only when ownershipType is exempt. A blank `ssn` on this submission is also allowed when it's an update (`id` matches an existing row) to an owner who already has one on file (`existing_row.ssn_enc`) — the merchant shouldn't be forced to retype an SSN that's never echoed back just to resubmit unrelated changes. """ exempt = bool(ownership_type) and ownership_type in OWNERSHIP_TYPES_SSN_EXEMPT for idx, owner in enumerate(owners): if owner.get("non_us_citizen"): continue # ssn is a US-citizen-branch-only field if owner.get("ssn"): continue # freshly provided — always fine owner_id = owner.get("id") existing_row = existing_by_id.get(owner_id) if owner_id else None if existing_row is not None and existing_row.ssn_enc: continue # blank on this submission, but already on file if not exempt: raise OwnerCapViolation( f"Owner {idx + 1}: ssn is required unless ownershipType is Financial Institution, " "Government Entity, or Public Corporation." ) def upsert_owners( db: Session, application_id: int, owners: List[Dict[str, Any]], ownership_type: Optional[str] = None, ) -> List[MerchantOnboardingOwner]: """ List-upsert (like accounts, PRD-HWONB-001 §7.3): each dict in `owners` with an `id` updates that existing row; each dict without one creates a new row. Any existing, non-soft-deleted owner row NOT present in `owners` (by id) is soft-deleted — the incoming list is the new authoritative set for the application, matching how the wizard's "+ Add another owner" / "Remove" repeat-item UI works (PRD-007 §5A). Raises OwnerCapViolation (never lets a bad cap reach the DB) if the resulting set would violate any of the caps documented on this module, or an ownershipType-dependent rule (SOLE/CNP forcing, ssn requirement). Does not commit — callers control the transaction boundary. """ # Work on shallow copies so forcing (SOLE/CNP) never mutates the # caller's own list/dicts out from under it (services.py may still want # to inspect what was originally submitted). owners = [dict(o) for o in owners] # Built before the ssn-requirement check so it can tell "update of an # owner who already has ssn_enc on file" apart from "new owner" (moved # up from below _validate_ssn_requirement for this reason). existing_by_id = {row.id: row for row in list_owners(db, application_id)} _apply_ownership_type_forcing(owners, ownership_type) _validate_owner_caps(owners) _validate_ssn_requirement(owners, ownership_type, existing_by_id) incoming_ids = {o["id"] for o in owners if o.get("id")} now = datetime.now(timezone.utc) for owner_id, row in existing_by_id.items(): if owner_id not in incoming_ids: row.deleted_at = now result: List[MerchantOnboardingOwner] = [] for data in owners: owner_id = data.get("id") if owner_id and owner_id in existing_by_id: row = existing_by_id[owner_id] _apply_owner_fields(row, data) else: row = MerchantOnboardingOwner(application_id=application_id) _apply_owner_fields(row, data) db.add(row) result.append(row) db.flush() return result def get_api_call_logs_for_merchant( db: Session, merchant_id: int, service: Optional[str] = None ) -> List[OnboardingApiCallLog]: """ Logged external API calls for this merchant, chronological — backs the Super Admin export buttons. `service` filters to one producer ("Base CRM" onboarding vs "TransIT" TSYS payment provider); None = all. """ stmt = select(OnboardingApiCallLog).where(OnboardingApiCallLog.merchant_id == merchant_id) if service is not None: stmt = stmt.where(OnboardingApiCallLog.service == service) stmt = stmt.order_by(OnboardingApiCallLog.created_at.asc()) return list(db.execute(stmt).scalars().all())