import html
import uuid
from contextlib import asynccontextmanager
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import HTMLResponse, RedirectResponse
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import AsyncSessionLocal, get_db
from app.core.permissions import deny_vendor, require_permission, resolve_portal_grants
from app.core.security import AuthContext, get_auth_context
from app.crud.payment_providers import PROVIDERS
from app.crud.payments import PaymentGatewayError, handle_callback, initiate_payment, verify_payment
from app.models.access import PageVersion
from app.models.navigation import Page
from app.models.reporting import PaymentTransaction
from app.schemas.payments import (
PaymentInitiateRequest,
PaymentInitiateResult,
PaymentProvider,
PaymentTransactionRead,
PaymentVerifyRequest,
PaymentVerifyResult,
)
router = APIRouter(prefix="/payments", tags=["payments"])
@router.get("/providers", response_model=list[PaymentProvider])
async def list_providers(_: AuthContext = Depends(get_auth_context)) -> list[dict]:
"""The supported gateways and each one's settings — drives the
Connectors screen's payment form."""
return PROVIDERS
def _walk_blocks(blocks: list) -> list[dict]:
out: list[dict] = []
for block in blocks or []:
if not isinstance(block, dict):
continue
out.append(block)
out.extend(_walk_blocks(block.get("children") or []))
for group in block.get("groups") or []:
if isinstance(group, dict):
out.extend(_walk_blocks(group.get("children") or []))
return out
async def _check_payment_access(
db: AsyncSession, ctx: AuthContext, *, block_type: str, connector_id: uuid.UUID, path_field: str, path_value: str | None
) -> None:
"""Who may start (/initiate) or check (/verify) a payment.
Internal (Studio) users: the "tasks":"edit" permission, same as running a
task from a Page/Form control — RBAC has no module/verb for payments.
Public Portal users: a portal role holds page grants, never action verbs,
so instead the call must match a Payment Initiator (/initiate) or Payment
Callback (/verify) block on a page their portal role grants — same
connector, and the block's own path override if the request sends one.
That's the button they actually clicked, so it's allowed; a portal login
can't aim a hand-made request at some other gateway connector or path."""
if ctx.vendor_id is None:
await require_permission("tasks", "edit")(ctx=ctx, db=db)
return
granted = (await resolve_portal_grants(db, ctx.tenant_id, ctx.user_id)).get("pages", set())
page_ids = []
for pid in granted:
try:
page_ids.append(uuid.UUID(pid))
except ValueError:
continue
if page_ids:
pages = (await db.execute(select(Page).where(Page.tenant_id == ctx.tenant_id, Page.id.in_(page_ids)))).scalars().all()
# What the portal user sees: a page's published version (GET
# /pages/{id}/live), not a draft being edited — the draft only for a
# page never published.
published = {
v.page_id: v.blocks
for v in (
await db.execute(
select(PageVersion).where(
PageVersion.page_id.in_([p.id for p in pages if p.published_version is not None]),
PageVersion.version.in_({p.published_version for p in pages if p.published_version is not None}),
)
)
).scalars()
if any(p.id == v.page_id and p.published_version == v.version for p in pages)
}
for page in pages:
for block in _walk_blocks(published.get(page.id, page.blocks)):
if (
block.get("type") == block_type
and str(block.get("payment_connector_id") or "").lower() == str(connector_id).lower()
and (not path_value or path_value == block.get(path_field))
):
return
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="This payment isn't available to your portal account — it has to be started from a payment control on a page your portal role can open.",
)
@router.post("/initiate", response_model=PaymentInitiateResult)
async def initiate(
payload: PaymentInitiateRequest,
ctx: AuthContext = Depends(get_auth_context),
db: AsyncSession = Depends(get_db),
) -> dict:
await _check_payment_access(
db, ctx, block_type="payment_initiator", connector_id=payload.connector_id, path_field="payment_checkout_path", path_value=payload.checkout_path
)
try:
return await initiate_payment(
db,
ctx.tenant_id,
payload.connector_id,
payload.amount,
currency=payload.currency,
reference=payload.reference,
checkout_path=payload.checkout_path,
return_url=payload.return_url,
created_by=ctx.user_id,
)
except PaymentGatewayError as exc:
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
@router.post("/verify", response_model=PaymentVerifyResult)
async def verify(
payload: PaymentVerifyRequest,
ctx: AuthContext = Depends(get_auth_context),
db: AsyncSession = Depends(get_db),
) -> dict:
await _check_payment_access(
db, ctx, block_type="payment_callback", connector_id=payload.connector_id, path_field="payment_status_path", path_value=payload.status_path
)
try:
return await verify_payment(db, ctx.tenant_id, payload.connector_id, payload.reference, status_path=payload.status_path)
except PaymentGatewayError as exc:
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
@router.get("/transactions", response_model=list[PaymentTransactionRead])
async def list_transactions(
connector_id: uuid.UUID | None = None,
limit: int = 20,
ctx: AuthContext = Depends(get_auth_context),
db: AsyncSession = Depends(get_db),
_: AuthContext = Depends(require_permission("connectors", "view")),
__: AuthContext = Depends(deny_vendor()),
) -> list[PaymentTransaction]:
stmt = select(PaymentTransaction).where(PaymentTransaction.tenant_id == ctx.tenant_id)
if connector_id:
stmt = stmt.where(PaymentTransaction.connector_id == connector_id)
stmt = stmt.order_by(PaymentTransaction.created_at.desc()).limit(min(max(limit, 1), 100))
return list((await db.execute(stmt)).scalars().all())
# ---- Browser-facing hops (no sign-in: the payer arrives from the gateway) ----
# Both are addressed by the transaction's unguessable id, and neither can
# change a payment's outcome on its own — the callback always re-checks the
# gateway's status API before reporting anything.
def _page(title: str, body: str) -> HTMLResponse:
return HTMLResponse(
""
f"{html.escape(title)}"
f"
{body}
"
)
@asynccontextmanager
async def _tenant_session(tenant_id: uuid.UUID):
"""A session scoped (RLS) to the tenant named in the URL — the payer
arriving from a gateway has no FlowFabric sign-in to take it from."""
async with AsyncSessionLocal() as session:
async with session.begin():
await session.execute(text("SELECT set_config('app.current_tenant_id', :v, true)"), {"v": str(tenant_id)})
yield session
@router.get("/launch/{tenant_id}/{transaction_id}", response_class=HTMLResponse, include_in_schema=False)
async def launch(tenant_id: uuid.UUID, transaction_id: uuid.UUID) -> HTMLResponse:
"""Hands the browser to gateways that take a form POST (e.g. Paytm)."""
async with _tenant_session(tenant_id) as db:
txn = await db.get(PaymentTransaction, transaction_id)
if txn is not None and txn.tenant_id != tenant_id:
txn = None
if txn is None or txn.status not in ("created", "pending") or (txn.checkout or {}).get("kind") != "form_post":
return _page("Payment", "
This payment link is no longer valid.
Please start the payment again.
")
checkout = txn.checkout
inputs = "".join(f"" for k, v in (checkout.get("fields") or {}).items())
return _page(
"Redirecting to payment…",
f"
Redirecting you to the payment page…
",
)
def _with_params(url: str, params: dict) -> str:
parts = urlsplit(url)
query = dict(parse_qsl(parts.query))
query.update(params)
return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))
@router.api_route("/callback/{tenant_id}/{transaction_id}", methods=["GET", "POST"], include_in_schema=False)
async def callback(tenant_id: uuid.UUID, transaction_id: uuid.UUID, request: Request):
params = dict(request.query_params)
if request.method == "POST":
try:
form = await request.form()
params.update({k: str(v) for k, v in form.items()})
except Exception: # noqa: BLE001 - a JSON or empty body just adds nothing
pass
async with _tenant_session(tenant_id) as db:
txn = await handle_callback(db, transaction_id, params)
if txn is not None and txn.tenant_id != tenant_id:
txn = None
if txn is None:
return _page("Payment", "