""" Daily Fee Settlement Sweep Task Runs nightly at 23:00 UTC. Collects all unsettled platform-fee transactions for the current calendar day and submits a single ACH debit to the HubWallet platform bank account via TSYS TransIT. Idempotency: - At task entry, query fee_settlement_runs for today's date. - If a row already exists with status != 'failed', skip. - A 'failed' run is retried (retry_count incremented). Design notes: - platform_fee_amount is stored as dollars (float). The sweep SUMs and multiplies by 100 to obtain cents for the ACH charge. - Transactions are batch-assigned to the run via a single UPDATE WHERE fee_settlement_run_id IS NULL. - TSYS credentials come from HW_TSYS_* env vars (platform account, not merchant credentials). """ import asyncio import logging from datetime import date, datetime, timezone from decimal import Decimal from typing import Optional from sqlalchemy import Date, cast, func, select, update from sqlalchemy.exc import IntegrityError from src.core.database import SessionLocal from src.worker.celery_app import celery_app logger = logging.getLogger(__name__) MAX_SWEEP_RETRIES = 3 @celery_app.task( name="fee_settlement.run_daily_fee_sweep", bind=True, max_retries=0, # retries handled in-task via run record ) def run_daily_fee_sweep( self, _triggered_by_admin: bool = False, _triggered_by_admin_id: Optional[int] = None, _parent_log_id: Optional[int] = None, **kwargs, ): """ Daily fee settlement sweep Celery task. Eligible transactions: status=PAID, category=CHARGE, platform_fee_amount > 0, fee_settlement_run_id IS NULL, occurred_at <= today. """ from src.apps.admin.models.fee_settlement_run import FeeSettlementRun, SettlementRunStatus, SettlementTrigger from src.apps.transactions.models.transactions import Transactions from src.core.utils.enums import TransactionStatusTypes, TransactionCategories triggered_by = SettlementTrigger.MANUAL if _triggered_by_admin else SettlementTrigger.SCHEDULER settlement_date = date.today() with SessionLocal() as db: # ── Idempotency check ────────────────────────────────────────────────── existing_run = db.execute( select(FeeSettlementRun).where( FeeSettlementRun.settlement_date == settlement_date, ) ).scalar_one_or_none() if existing_run and existing_run.status != SettlementRunStatus.FAILED: logger.info( "FEE_SWEEP: run for %s already exists with status=%s — skipping", settlement_date, existing_run.status, ) return {"status": "skipped", "reason": f"run exists with status={existing_run.status}"} # ── Check retry cap ──────────────────────────────────────────────────── if existing_run and existing_run.status == SettlementRunStatus.FAILED: if (existing_run.retry_count or 0) >= MAX_SWEEP_RETRIES: logger.error( "FEE_SWEEP: run id=%d for %s has exhausted %d retries — not retrying. " "Admin must investigate and manually reset the run.", existing_run.id, settlement_date, MAX_SWEEP_RETRIES, ) return { "status": "exhausted", "run_id": existing_run.id, "reason": f"exhausted {MAX_SWEEP_RETRIES} retries", } # ── Load active settlement account ───────────────────────────────────── from src.apps.fee_settlement.crud import get_active_settlement_account from src.apps.fee_settlement.services import get_decrypted_account_credentials account = get_active_settlement_account(db) if not account: logger.warning("FEE_SWEEP: no active settlement account configured — aborting") return {"status": "skipped", "reason": "no active settlement account"} # ── Create or reuse run record ───────────────────────────────────────── if existing_run and existing_run.status == SettlementRunStatus.FAILED: # Retry the failed run — increment retry_count and reset status run = existing_run run.retry_count = (run.retry_count or 0) + 1 run.status = SettlementRunStatus.PROCESSING run.last_error = None run.triggered_by = triggered_by db.flush() logger.info("FEE_SWEEP: retrying failed run id=%d (attempt %d)", run.id, run.retry_count) else: from src.apps.fee_settlement.crud import create_settlement_run try: run = create_settlement_run( db=db, settlement_date=settlement_date, settlement_account_id=account.id, triggered_by=triggered_by, ) run.status = SettlementRunStatus.PROCESSING db.flush() except IntegrityError: # Another concurrent worker already claimed this settlement date. # The settlement_date UNIQUE constraint is the authoritative lock. db.rollback() logger.info( "FEE_SWEEP: concurrent worker claimed settlement for %s — skipping", settlement_date, ) return {"status": "skipped", "reason": "concurrent worker claimed this date"} logger.info("FEE_SWEEP: created run id=%d for date=%s", run.id, settlement_date) db.commit() # ── Batch-assign eligible transactions ───────────────────────────────── db.execute( update(Transactions) .where( Transactions.fee_settlement_run_id.is_(None), Transactions.platform_fee_amount > 0, Transactions.txn_status == TransactionStatusTypes.PAID, Transactions.category == TransactionCategories.CHARGE, Transactions.deleted_at.is_(None), cast(Transactions.ocurred_at, Date) <= settlement_date, ) .values(fee_settlement_run_id=run.id) ) db.commit() # ── Aggregate totals ─────────────────────────────────────────────────── agg = db.execute( select( func.count(Transactions.id).label("txn_count"), func.coalesce(func.sum(Transactions.platform_fee_amount), 0).label("total_fee_dollars"), ).where(Transactions.fee_settlement_run_id == run.id) ).one() txn_count = agg.txn_count or 0 # Use Decimal throughout to avoid IEEE 754 float accumulation errors. # PostgreSQL SUM on NUMERIC columns returns a Decimal; convert 0 via str # to avoid float contamination. total_fee_decimal = Decimal(str(agg.total_fee_dollars or 0)) total_fee_cents = int( (total_fee_decimal * 100).to_integral_value() ) # amount_dollars passed to TSYS — keep as float only at the API boundary total_fee_dollars = float(total_fee_decimal) run.transaction_count = txn_count run.total_fee_cents = total_fee_cents db.commit() if txn_count == 0 or total_fee_cents == 0: run.status = SettlementRunStatus.SKIPPED db.commit() logger.info( "FEE_SWEEP: no unsettled fees for %s (txn_count=%d, total_cents=%d) — SKIPPED", settlement_date, txn_count, total_fee_cents, ) return {"status": "skipped", "reason": "no fees to settle", "transaction_count": txn_count} # ── Decrypt ACH credentials ──────────────────────────────────────────── try: routing_number, account_number = get_decrypted_account_credentials(account) except Exception as exc: run.status = SettlementRunStatus.FAILED run.last_error = f"credential_decrypt_failed: {exc}" db.commit() logger.error("FEE_SWEEP: credential decryption failed: %s", exc) return {"status": "failed", "reason": str(exc)} # ── Submit ACH via TSYS ──────────────────────────────────────────────── try: tsys_txn_id = _submit_ach_to_tsys( routing_number=routing_number, account_number=account_number, account_type=account.account_type, account_holder_name=account.account_holder_name, amount_dollars=total_fee_dollars, ) except Exception as exc: run.status = SettlementRunStatus.FAILED run.last_error = str(exc)[:500] db.commit() logger.error("FEE_SWEEP: TSYS ACH submission failed: %s", exc) return {"status": "failed", "reason": str(exc)} # ── Mark settled ─────────────────────────────────────────────────────── run.status = SettlementRunStatus.SETTLED run.tsys_transaction_id = tsys_txn_id run.settled_at = datetime.now(timezone.utc) db.commit() logger.info( "FEE_SWEEP: SETTLED run_id=%d txns=%d total_cents=%d tsys_txn_id=%s", run.id, txn_count, total_fee_cents, tsys_txn_id, ) return { "status": "settled", "run_id": run.id, "settlement_date": str(settlement_date), "transaction_count": txn_count, "total_fee_cents": total_fee_cents, "tsys_transaction_id": tsys_txn_id, "records_processed": txn_count, } def _submit_ach_to_tsys( routing_number: str, account_number: str, account_type: str, account_holder_name: str, amount_dollars: float, ) -> str: """ Submit an ACH debit to the HubWallet platform bank account via TSYS TransIT. Uses HW_TSYS_* env vars for platform credentials. Returns the provider transaction ID on success. Raises RuntimeError on failure. """ from src.core.config import settings from src.core.providers.base import ProviderConfig from src.core.providers.implementations.tsys import TSYSProvider if settings.PAYMENT_PROVIDER_STUB: import uuid stub_id = f"STUB-ACH-{uuid.uuid4().hex[:12].upper()}" logger.info("FEE_SWEEP: stub mode — skipping real TSYS ACH, returning %s", stub_id) return stub_id merchant_id = getattr(settings, "HW_TSYS_MERCHANT_ID", None) device_id = getattr(settings, "HW_TSYS_DEVICE_ID", None) transaction_key = getattr(settings, "HW_TSYS_TRANSACTION_KEY", None) developer_id = getattr(settings, "HW_TSYS_DEVELOPER_ID", None) api_base_url = getattr(settings, "HW_TSYS_API_BASE_URL", "https://stagegw.transnox.com/transit-api") if not merchant_id or not device_id or not transaction_key: raise RuntimeError( "HW_TSYS_MERCHANT_ID, HW_TSYS_DEVICE_ID, and HW_TSYS_TRANSACTION_KEY " "must all be set for ACH settlement" ) # Build a standalone ProviderConfig using platform HW_TSYS credentials config = ProviderConfig( provider_slug="tsys", credentials={ "merchant_id": merchant_id, "device_id": device_id, "transaction_key": transaction_key, "developer_id": developer_id or "", "api_base_url": api_base_url, }, ) # Split account_holder_name into first/last for TSYS fields name_parts = account_holder_name.strip().split(" ", 1) first_name = name_parts[0] last_name = name_parts[1] if len(name_parts) > 1 else first_name tsys = TSYSProvider() result = asyncio.run( tsys.submit_ach_charge( config=config, routing_number=routing_number, account_number=account_number, account_type=account_type, first_name=first_name, last_name=last_name, amount=amount_dollars, ) ) if result.status != "succeeded": raw = result.raw_response err_msg = ( raw.get("AchResponse", {}).get("responseText") or raw.get("AchResponse", {}).get("responseCode") or f"TSYS ACH declined: status={result.status}" ) raise RuntimeError(err_msg) return result.transaction_id