mirror of
https://github.com/Such-Software/wowcheckout.git
synced 2026-10-17 05:51:45 +02:00
jwinterm
0aa19ca6bf
Exposes the WOW→fiat spot rate the invoice path already uses internally (Nonlogs WOW-BTC × Kraken BTC-USD, 60s cache). No auth. Allows downstream services (Medusa storefront) to display a live USD reference price for WOW-denominated products without hard-coding a third-party rate provider on the storefront side.
1443 lines
50 KiB
Python
1443 lines
50 KiB
Python
import uuid
|
||
from typing import Any
|
||
from datetime import datetime, timedelta, timezone
|
||
from decimal import Decimal, ROUND_DOWN, ROUND_UP, InvalidOperation
|
||
import csv
|
||
import io
|
||
import json
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
|
||
from fastapi.responses import RedirectResponse, StreamingResponse
|
||
import requests
|
||
from requests import RequestException
|
||
from sqlalchemy import func, or_
|
||
from sqlalchemy.sql import cast
|
||
from sqlalchemy.types import String as SqlString
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .config import (
|
||
DONATION_ACTIVE_INVOICE_LIMIT,
|
||
DONATION_EXPIRY_MINUTES,
|
||
DONATIONS_ENABLED,
|
||
FOUNDER_PAYMENT_ADDRESS,
|
||
FOUNDER_VIEW_KEY,
|
||
INVOICE_DEFAULT_EXPIRY_HOURS,
|
||
INVOICE_RECONCILE_INTERVAL_SECONDS,
|
||
OPEN_REGISTRATION,
|
||
QR_STORAGE_DIR,
|
||
)
|
||
from .db import get_db
|
||
from .formatting import format_xmr_amount
|
||
from .models import Invoice, ProfileHistory, SystemStatus, User, Webhook, WebhookDelivery
|
||
from .rates import get_wow_rate
|
||
from .subaddress_allocator import MAX_SUBADDRESS_INDEX, create_subaddress_for_user
|
||
from .schemas import (
|
||
ApiCredentialsResetRequest,
|
||
ApiCredentialsResetResponse,
|
||
DonationCreate,
|
||
InvoiceCreateUser,
|
||
InvoiceListResponse,
|
||
InvoiceResponse,
|
||
InvoiceStatusResponse,
|
||
LoginRequest,
|
||
LoginResponse,
|
||
ProfileResponse,
|
||
ProfileUpdate,
|
||
SystemStatusResponse,
|
||
WebhookCreate,
|
||
WebhookDeliveryResponse,
|
||
WebhookResponse,
|
||
)
|
||
from .security import (
|
||
decrypt_secret,
|
||
decrypt_api_key,
|
||
encrypt_secret,
|
||
encrypt_api_key,
|
||
generate_api_key,
|
||
generate_webhook_secret,
|
||
hash_api_key,
|
||
require_api_key,
|
||
)
|
||
from .webhooks import build_webhook_payload, dispatch_webhooks
|
||
from .qr_codes import ensure_invoice_qr_png, invoice_qr_url, resolve_qr_settings
|
||
router = APIRouter()
|
||
|
||
WEBHOOK_EVENTS = (
|
||
"invoice.created",
|
||
"invoice.payment_detected",
|
||
"invoice.confirmed",
|
||
"invoice.expired",
|
||
)
|
||
|
||
|
||
def _require_donations_enabled() -> None:
|
||
if not DONATIONS_ENABLED:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Donations are disabled",
|
||
)
|
||
|
||
MAX_QR_LOGO_DATA_URL_LENGTH = 120_000
|
||
MONERO_CONNECTIVITY_STATUS_NAME = "monero_connectivity"
|
||
|
||
|
||
def _get_user_for_api_key(db: Session, api_key: str) -> User | None:
|
||
return db.query(User).filter(User.api_key_hash == hash_api_key(api_key)).first()
|
||
|
||
|
||
def _status_response_from_rows(
|
||
monero_status: SystemStatus | None,
|
||
reconciler_status: SystemStatus | None,
|
||
) -> SystemStatusResponse:
|
||
wallet_rpc = (
|
||
str(monero_status.wallet_rpc)
|
||
if monero_status and monero_status.wallet_rpc
|
||
else "unreachable"
|
||
)
|
||
daemon = (
|
||
str(monero_status.daemon)
|
||
if monero_status and monero_status.daemon
|
||
else "unknown"
|
||
)
|
||
daemon_height = monero_status.daemon_height if monero_status else None
|
||
return SystemStatusResponse(
|
||
wallet_rpc=wallet_rpc,
|
||
daemon=daemon,
|
||
daemon_height=daemon_height,
|
||
invoice_reconcile_interval_seconds=INVOICE_RECONCILE_INTERVAL_SECONDS,
|
||
last_reconcile_started_at=(
|
||
reconciler_status.last_reconcile_started_at if reconciler_status else None
|
||
),
|
||
last_reconcile_completed_at=(
|
||
reconciler_status.last_reconcile_completed_at if reconciler_status else None
|
||
),
|
||
last_reconcile_error=(
|
||
reconciler_status.last_reconcile_error if reconciler_status else None
|
||
),
|
||
)
|
||
|
||
|
||
def _load_public_system_status(db: Session) -> SystemStatusResponse:
|
||
reconciler_status = (
|
||
db.query(SystemStatus).filter(SystemStatus.name == "reconciler").first()
|
||
)
|
||
monero_status = (
|
||
db.query(SystemStatus)
|
||
.filter(SystemStatus.name == MONERO_CONNECTIVITY_STATUS_NAME)
|
||
.first()
|
||
)
|
||
return _status_response_from_rows(monero_status, reconciler_status)
|
||
|
||
|
||
def _get_founder_user(db: Session) -> User:
|
||
if not FOUNDER_PAYMENT_ADDRESS or not FOUNDER_VIEW_KEY:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||
detail="Donations are not configured",
|
||
)
|
||
_validate_payment_address_and_view_key(FOUNDER_PAYMENT_ADDRESS, FOUNDER_VIEW_KEY)
|
||
user = (
|
||
db.query(User)
|
||
.filter(User.payment_address == FOUNDER_PAYMENT_ADDRESS)
|
||
.first()
|
||
)
|
||
if user is None:
|
||
api_key = generate_api_key()
|
||
webhook_secret = generate_webhook_secret()
|
||
user = User(
|
||
payment_address=FOUNDER_PAYMENT_ADDRESS,
|
||
view_key_encrypted=encrypt_secret(FOUNDER_VIEW_KEY),
|
||
api_key_hash=hash_api_key(api_key),
|
||
api_key_encrypted=encrypt_api_key(api_key),
|
||
webhook_secret_encrypted=encrypt_secret(webhook_secret),
|
||
)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
return user
|
||
if decrypt_secret(user.view_key_encrypted) != FOUNDER_VIEW_KEY:
|
||
user.view_key_encrypted = encrypt_secret(FOUNDER_VIEW_KEY)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
_ensure_webhook_secret(db, user)
|
||
return user
|
||
|
||
|
||
def _default_invoice_expiry() -> datetime:
|
||
return datetime.now(timezone.utc) + timedelta(hours=INVOICE_DEFAULT_EXPIRY_HOURS)
|
||
|
||
|
||
def _donation_invoice_expiry() -> datetime:
|
||
return datetime.now(timezone.utc) + timedelta(minutes=DONATION_EXPIRY_MINUTES)
|
||
|
||
|
||
def _is_donation_invoice(invoice: Invoice) -> bool:
|
||
metadata = invoice.metadata_json or {}
|
||
return isinstance(metadata, dict) and metadata.get("origin") == "donation"
|
||
|
||
|
||
def _get_checkout_continue_url(invoice: Invoice) -> str | None:
|
||
metadata = invoice.metadata_json or {}
|
||
if not isinstance(metadata, dict):
|
||
return None
|
||
checkout = metadata.get("checkout")
|
||
if not isinstance(checkout, dict):
|
||
return None
|
||
value = checkout.get("continue_url")
|
||
if not isinstance(value, str):
|
||
return None
|
||
trimmed = value.strip()
|
||
return trimmed or None
|
||
|
||
|
||
def _ensure_webhook_secret(db: Session, user: User) -> str:
|
||
if user.webhook_secret_encrypted:
|
||
return decrypt_secret(user.webhook_secret_encrypted)
|
||
secret = generate_webhook_secret()
|
||
user.webhook_secret_encrypted = encrypt_secret(secret)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
return secret
|
||
|
||
|
||
def _resolve_webhook_payload(
|
||
payload: WebhookCreate,
|
||
) -> tuple[str | None, list[str], dict[str, str] | None]:
|
||
event_urls = {key: str(value) for key, value in (payload.event_urls or {}).items()}
|
||
events = list(payload.events or [])
|
||
if event_urls:
|
||
for key in event_urls:
|
||
if key not in WEBHOOK_EVENTS:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Webhook event is not supported",
|
||
)
|
||
if events:
|
||
for event in events:
|
||
if event not in WEBHOOK_EVENTS:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Webhook event is not supported",
|
||
)
|
||
events_set = set(events) | set(event_urls.keys())
|
||
if not events_set:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Select at least one webhook event",
|
||
)
|
||
if payload.url is None and not event_urls:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Webhook URL is required",
|
||
)
|
||
if payload.url is None:
|
||
missing = [event for event in events_set if event not in event_urls]
|
||
if missing:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Webhook URL is required for all selected events",
|
||
)
|
||
ordered_events = [event for event in WEBHOOK_EVENTS if event in events_set]
|
||
return (str(payload.url) if payload.url else None, ordered_events, event_urls or None)
|
||
|
||
|
||
def _invoice_url(invoice: Invoice, request: Request) -> str:
|
||
proto = request.headers.get("x-forwarded-proto", request.url.scheme)
|
||
host = request.headers.get("x-forwarded-host", request.headers.get("host", ""))
|
||
if not host:
|
||
return f"/invoice/{invoice.id}"
|
||
return f"{proto}://{host}/invoice/{invoice.id}"
|
||
|
||
|
||
def _qr_url(invoice: Invoice, request: Request) -> str:
|
||
proto = request.headers.get("x-forwarded-proto", request.url.scheme)
|
||
host = request.headers.get("x-forwarded-host", request.headers.get("host", ""))
|
||
path = invoice_qr_url(str(invoice.id))
|
||
if not host:
|
||
return path
|
||
return f"{proto}://{host}{path}"
|
||
|
||
|
||
def _invoice_response(
|
||
invoice: Invoice,
|
||
request: Request,
|
||
warnings: list[str] | None = None,
|
||
) -> InvoiceResponse:
|
||
response = InvoiceResponse.model_validate(invoice)
|
||
metadata = invoice.metadata_json or {}
|
||
quote = metadata.get("quote") if isinstance(metadata, dict) else None
|
||
qr = metadata.get("qr") if isinstance(metadata, dict) else None
|
||
qr_logo = qr.get("logo") if isinstance(qr, dict) else None
|
||
qr_logo_data_url = qr.get("logo_data_url") if isinstance(qr, dict) else None
|
||
if not isinstance(qr_logo, str):
|
||
qr_logo = None
|
||
if qr_logo not in ("monero", "none", "custom"):
|
||
qr_logo = None
|
||
qr_logo_data_url = None
|
||
if qr_logo != "custom":
|
||
qr_logo_data_url = None
|
||
if not isinstance(qr_logo_data_url, str):
|
||
qr_logo_data_url = None
|
||
return response.model_copy(
|
||
update={
|
||
"invoice_url": _invoice_url(invoice, request),
|
||
"qr_url": _qr_url(invoice, request),
|
||
"warnings": warnings,
|
||
"quote": quote,
|
||
"qr_logo": qr_logo,
|
||
"qr_logo_data_url": qr_logo_data_url,
|
||
}
|
||
)
|
||
|
||
|
||
def _public_invoice_status_response(
|
||
db: Session,
|
||
invoice: Invoice,
|
||
request: Request,
|
||
) -> InvoiceStatusResponse:
|
||
response = InvoiceStatusResponse.model_validate(invoice)
|
||
metadata = invoice.metadata_json or {}
|
||
update: dict[str, Any] = {}
|
||
if invoice.total_paid_atomic is not None:
|
||
# Wownero uses 11 decimal places (10^11), NOT 12 like Monero. Using
|
||
# 10^12 here under-reports amount_paid by a factor of 10.
|
||
total_paid = (Decimal(invoice.total_paid_atomic) / Decimal("100000000000")).quantize(
|
||
Decimal("0.00000000001"), rounding=ROUND_DOWN
|
||
)
|
||
update["amount_paid_xmr"] = format_xmr_amount(total_paid)
|
||
btcpay_data = metadata.get("btcpay") if isinstance(metadata, dict) else None
|
||
if isinstance(btcpay_data, dict):
|
||
amount = btcpay_data.get("amount")
|
||
currency = btcpay_data.get("currency")
|
||
if isinstance(amount, str) and isinstance(currency, str):
|
||
update["btcpay_amount"] = amount
|
||
update["btcpay_currency"] = currency
|
||
if invoice.user_id:
|
||
user = db.query(User).filter(User.id == invoice.user_id).first()
|
||
if user and user.btcpay_checkout_style:
|
||
update["btcpay_checkout_style"] = user.btcpay_checkout_style
|
||
checkout = btcpay_data.get("checkout")
|
||
if isinstance(checkout, dict):
|
||
redirect_url = checkout.get("redirectURL")
|
||
if isinstance(redirect_url, str) and redirect_url.strip():
|
||
update["btcpay_redirect_url"] = redirect_url.strip()
|
||
redirect_auto = checkout.get("redirectAutomatically")
|
||
if isinstance(redirect_auto, bool):
|
||
update["btcpay_redirect_automatically"] = redirect_auto
|
||
if isinstance(metadata, dict) and isinstance(metadata.get("quote"), dict):
|
||
update["quote"] = metadata.get("quote")
|
||
if isinstance(metadata, dict) and isinstance(metadata.get("posData"), str):
|
||
try:
|
||
pos_data = json.loads(metadata.get("posData", ""))
|
||
except (TypeError, ValueError, json.JSONDecodeError):
|
||
pos_data = None
|
||
if isinstance(pos_data, dict):
|
||
woo = pos_data.get("WooCommerce")
|
||
if isinstance(woo, dict):
|
||
order_id = woo.get("Order ID")
|
||
order_number = woo.get("Order Number")
|
||
if order_id is not None:
|
||
update["btcpay_order_id"] = str(order_id)
|
||
if order_number is not None:
|
||
update["btcpay_order_number"] = str(order_number)
|
||
qr = metadata.get("qr") if isinstance(metadata, dict) else None
|
||
if isinstance(qr, dict):
|
||
logo = qr.get("logo")
|
||
logo_data_url = qr.get("logo_data_url")
|
||
if isinstance(logo, str) and logo in ("monero", "none", "custom"):
|
||
update["qr_logo"] = logo
|
||
if logo == "custom" and isinstance(logo_data_url, str):
|
||
update["qr_logo_data_url"] = logo_data_url
|
||
continue_url = _get_checkout_continue_url(invoice)
|
||
update["checkout_continue_available"] = bool(
|
||
invoice.status == "confirmed" and continue_url
|
||
)
|
||
update["qr_url"] = _qr_url(invoice, request)
|
||
return response.model_copy(update=update)
|
||
|
||
|
||
def _resolve_invoice_amount(
|
||
db: Session,
|
||
*,
|
||
user: User,
|
||
requested_amount_xmr: Decimal | None = None,
|
||
requested_amount_fiat: Decimal | None = None,
|
||
requested_currency: str | None = None,
|
||
) -> tuple[Decimal, list[str] | None, dict[str, str] | None]:
|
||
if requested_amount_xmr is not None:
|
||
return requested_amount_xmr, None, None
|
||
if requested_amount_fiat is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="amount_xmr or amount_fiat is required",
|
||
)
|
||
if not requested_currency:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="currency is required when amount_fiat is provided",
|
||
)
|
||
try:
|
||
quote = get_wow_rate(requested_currency)
|
||
except ValueError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=str(exc),
|
||
) from exc
|
||
except Exception as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||
detail="Fiat quote service unavailable",
|
||
) from exc
|
||
# Round UP to 3 decimal places so we never undercharge.
|
||
# 3 decimals = 0.001 WOW — more than enough precision for Wownero.
|
||
# See also: btcpay_routes.py for the same rounding.
|
||
amount_xmr = (Decimal(requested_amount_fiat) / quote.rate).quantize(
|
||
Decimal("0.001"), rounding=ROUND_UP
|
||
)
|
||
warnings = ["Fiat conversion is an estimate and does not lock a rate."]
|
||
quote_payload = {
|
||
"fiat_amount": str(requested_amount_fiat),
|
||
"fiat_currency": quote.currency,
|
||
"rate": str(quote.rate),
|
||
"source": quote.source,
|
||
"quoted_at": quote.quoted_at.isoformat(),
|
||
}
|
||
return amount_xmr, warnings, quote_payload
|
||
|
||
|
||
def _validate_payment_address_and_view_key(payment_address: str, view_key: str) -> None:
|
||
"""Validate a Wownero primary address and its secret view key.
|
||
|
||
Wownero uses 2-byte varint prefixes (unlike Monero's 1-byte), so the
|
||
standard monero Python library cannot parse them correctly. We do
|
||
base58-decode + keccak checksum + ed25519 view-key verification here.
|
||
"""
|
||
from binascii import hexlify, unhexlify
|
||
from monero import base58, ed25519
|
||
from monero.keccak import keccak_256
|
||
|
||
# Wownero base58 prefix varints (first decoded byte)
|
||
_WOW_PRIMARY_BYTE0 = 178 # base58 prefix 4146 -> "Wo"
|
||
_WOW_SUBADDR_BYTE0 = 176 # base58 prefix 12208 -> "Ww"
|
||
_WOW_INTEGRATED_BYTE0 = 179 # base58 prefix 4147
|
||
|
||
try:
|
||
decoded = bytearray(unhexlify(base58.decode(payment_address)))
|
||
except Exception:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Primary address is invalid",
|
||
)
|
||
|
||
# Verify keccak-256 checksum (last 4 bytes)
|
||
if decoded[-4:] != keccak_256(decoded[:-4]).digest()[:4]:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Primary address is invalid",
|
||
)
|
||
|
||
first_byte = decoded[0]
|
||
if first_byte == _WOW_SUBADDR_BYTE0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Subaddresses are not supported. Use the primary address.",
|
||
)
|
||
if first_byte == _WOW_INTEGRATED_BYTE0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Integrated addresses are not supported. Use the primary address.",
|
||
)
|
||
if first_byte != _WOW_PRIMARY_BYTE0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Primary address is invalid",
|
||
)
|
||
|
||
# Wownero has 2-byte varint prefix -> keys start at offset 2
|
||
pub_spend = decoded[2:34]
|
||
pub_view = decoded[34:66]
|
||
|
||
# Derive public view key from the secret view key and compare
|
||
try:
|
||
derived_pub_view = ed25519.public_from_secret_hex(view_key)
|
||
if unhexlify(derived_pub_view) != bytes(pub_view):
|
||
raise ValueError("mismatch")
|
||
except Exception:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Secret view key does not match the primary address",
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/api/core/donations",
|
||
response_model=InvoiceResponse,
|
||
status_code=201,
|
||
include_in_schema=False,
|
||
)
|
||
def create_donation_invoice(
|
||
payload: DonationCreate,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_require_donations_enabled()
|
||
user = _get_founder_user(db)
|
||
expires_at = _donation_invoice_expiry()
|
||
active_limit = min(max(1, DONATION_ACTIVE_INVOICE_LIMIT), MAX_SUBADDRESS_INDEX)
|
||
now = datetime.now(timezone.utc)
|
||
active_count = (
|
||
db.query(func.count(Invoice.id))
|
||
.filter(
|
||
Invoice.user_id == user.id,
|
||
Invoice.status.in_(["pending", "payment_detected"]),
|
||
or_(Invoice.expires_at.is_(None), Invoice.expires_at > now),
|
||
)
|
||
.scalar()
|
||
or 0
|
||
)
|
||
if active_count >= active_limit:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
||
detail="Donation traffic is high. Try again soon.",
|
||
)
|
||
resolved_amount, warnings, _ = _resolve_invoice_amount(
|
||
db,
|
||
user=user,
|
||
requested_amount_xmr=payload.amount_xmr,
|
||
)
|
||
address, subaddress_index = create_subaddress_for_user(db, user=user)
|
||
invoice = Invoice(
|
||
user_id=user.id,
|
||
wallet_address=user.payment_address,
|
||
address=address,
|
||
subaddress_index=subaddress_index,
|
||
amount_xmr=resolved_amount,
|
||
status="pending",
|
||
confirmation_target=payload.confirmation_target,
|
||
metadata_json={"origin": "donation"},
|
||
expires_at=expires_at,
|
||
)
|
||
db.add(invoice)
|
||
db.commit()
|
||
db.refresh(invoice)
|
||
try:
|
||
ensure_invoice_qr_png(
|
||
invoice=invoice,
|
||
storage_dir=QR_STORAGE_DIR,
|
||
settings=resolve_qr_settings(invoice),
|
||
)
|
||
except HTTPException:
|
||
pass
|
||
return _invoice_response(invoice, request, warnings=warnings)
|
||
|
||
|
||
@router.get(
|
||
"/api/core/public/invoice/{invoice_id}",
|
||
response_model=InvoiceStatusResponse,
|
||
)
|
||
def get_invoice_status(
|
||
invoice_id: uuid.UUID,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
invoice = db.query(Invoice).filter(Invoice.id == invoice_id).first()
|
||
if invoice is None or _is_donation_invoice(invoice):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Invoice not found",
|
||
)
|
||
try:
|
||
ensure_invoice_qr_png(
|
||
invoice=invoice,
|
||
storage_dir=QR_STORAGE_DIR,
|
||
settings=resolve_qr_settings(invoice),
|
||
)
|
||
except HTTPException:
|
||
pass
|
||
return _public_invoice_status_response(db, invoice, request)
|
||
|
||
|
||
@router.get(
|
||
"/api/core/public/invoice/{invoice_id}/continue",
|
||
response_class=RedirectResponse,
|
||
)
|
||
def continue_invoice_after_confirmation(
|
||
invoice_id: uuid.UUID,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
invoice = db.query(Invoice).filter(Invoice.id == invoice_id).first()
|
||
if invoice is None or _is_donation_invoice(invoice):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Invoice not found",
|
||
)
|
||
if invoice.status != "confirmed":
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Continue link not available",
|
||
)
|
||
continue_url = _get_checkout_continue_url(invoice)
|
||
if not continue_url:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Continue link not available",
|
||
)
|
||
response = RedirectResponse(url=continue_url, status_code=status.HTTP_302_FOUND)
|
||
response.headers["Cache-Control"] = "no-store"
|
||
return response
|
||
|
||
|
||
@router.get(
|
||
"/api/core/public/donation/{invoice_id}",
|
||
response_model=InvoiceStatusResponse,
|
||
include_in_schema=False,
|
||
)
|
||
def get_donation_status(
|
||
invoice_id: uuid.UUID,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_require_donations_enabled()
|
||
founder = _get_founder_user(db)
|
||
invoice = db.query(Invoice).filter(Invoice.id == invoice_id).first()
|
||
if (
|
||
invoice is None
|
||
or not _is_donation_invoice(invoice)
|
||
or invoice.user_id != founder.id
|
||
):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Donation invoice not found",
|
||
)
|
||
try:
|
||
ensure_invoice_qr_png(
|
||
invoice=invoice,
|
||
storage_dir=QR_STORAGE_DIR,
|
||
settings=resolve_qr_settings(invoice),
|
||
)
|
||
except HTTPException:
|
||
pass
|
||
return _public_invoice_status_response(db, invoice, request)
|
||
|
||
|
||
@router.get("/api/core/public/system/status", response_model=SystemStatusResponse)
|
||
def get_public_system_status(db: Session = Depends(get_db)):
|
||
return _load_public_system_status(db)
|
||
|
||
|
||
@router.get("/api/core/public/rate")
|
||
def get_public_rate(currency: str = "usd"):
|
||
"""
|
||
Public WOW→fiat spot rate. Returns the same QuoteResult that the
|
||
invoice-creation paths use internally (60s in-process cache, sourced
|
||
from Nonlogs WOW-BTC × Kraken BTC-USD).
|
||
|
||
No auth — this is the same data the marketing storefront would
|
||
otherwise have to scrape from Nonlogs directly. Exposing it lets
|
||
downstream consumers (Medusa) get an authoritative-on-our-side
|
||
rate without hard-coding a third-party API.
|
||
"""
|
||
try:
|
||
quote = get_wow_rate(currency)
|
||
except ValueError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (RequestException, RuntimeError) as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||
detail=f"Rate temporarily unavailable: {exc}",
|
||
) from exc
|
||
return {
|
||
"rate": str(quote.rate),
|
||
"currency": quote.currency,
|
||
"source": quote.source,
|
||
"quoted_at": quote.quoted_at.isoformat(),
|
||
"base": "WOW",
|
||
}
|
||
|
||
|
||
@router.post("/api/core/webhooks", response_model=WebhookResponse)
|
||
def register_webhook(
|
||
payload: WebhookCreate,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
_ensure_webhook_secret(db, user)
|
||
url, events, event_urls = _resolve_webhook_payload(payload)
|
||
webhook = Webhook(
|
||
user_id=user.id,
|
||
url=url,
|
||
events=events,
|
||
event_urls=event_urls,
|
||
active=True,
|
||
)
|
||
db.add(webhook)
|
||
db.commit()
|
||
db.refresh(webhook)
|
||
return webhook
|
||
|
||
|
||
@router.get("/api/core/webhooks", response_model=list[WebhookResponse])
|
||
def list_webhooks(
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
return (
|
||
db.query(Webhook)
|
||
.filter(Webhook.user_id == user.id)
|
||
.order_by(Webhook.created_at.desc())
|
||
.all()
|
||
)
|
||
|
||
|
||
@router.get("/api/core/webhooks/history", response_model=list[WebhookDeliveryResponse])
|
||
def list_webhook_history(
|
||
limit: int = Query(50, ge=1, le=200),
|
||
offset: int = Query(0, ge=0),
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
return (
|
||
db.query(WebhookDelivery)
|
||
.filter(WebhookDelivery.user_id == user.id)
|
||
.order_by(WebhookDelivery.created_at.desc())
|
||
.limit(limit)
|
||
.offset(offset)
|
||
.all()
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/api/core/webhooks/deliveries/{delivery_id}/redeliver",
|
||
status_code=200,
|
||
)
|
||
def redeliver_webhook_delivery(
|
||
delivery_id: uuid.UUID,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
delivery = (
|
||
db.query(WebhookDelivery)
|
||
.filter(WebhookDelivery.id == delivery_id, WebhookDelivery.user_id == user.id)
|
||
.first()
|
||
)
|
||
if delivery is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Webhook delivery not found",
|
||
)
|
||
if delivery.http_status is not None and delivery.http_status < 400:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail="Only failed deliveries can be redelivered",
|
||
)
|
||
payload = delivery.payload_json
|
||
if not isinstance(payload, dict):
|
||
payload = None
|
||
if payload is None and delivery.invoice_id:
|
||
invoice = db.query(Invoice).filter(Invoice.id == delivery.invoice_id).first()
|
||
if invoice is not None:
|
||
payload = build_webhook_payload(delivery.event, invoice)
|
||
if payload is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail="Webhook payload is unavailable for this delivery",
|
||
)
|
||
|
||
webhook_secret = _ensure_webhook_secret(db, user)
|
||
status_code = None
|
||
error_message = None
|
||
try:
|
||
response = requests.post(
|
||
delivery.url,
|
||
json=payload,
|
||
headers={"X-Webhook-Secret": webhook_secret},
|
||
timeout=5,
|
||
)
|
||
status_code = response.status_code
|
||
except RequestException as exc:
|
||
error_message = str(exc)
|
||
|
||
redelivery = WebhookDelivery(
|
||
user_id=user.id,
|
||
webhook_id=delivery.webhook_id,
|
||
event=delivery.event,
|
||
url=delivery.url,
|
||
invoice_id=delivery.invoice_id,
|
||
invoice_address=delivery.invoice_address,
|
||
invoice_subaddress_index=delivery.invoice_subaddress_index,
|
||
invoice_amount_xmr=delivery.invoice_amount_xmr,
|
||
invoice_status=delivery.invoice_status,
|
||
payload_json=payload,
|
||
http_status=status_code,
|
||
error_message=error_message,
|
||
)
|
||
db.add(redelivery)
|
||
db.commit()
|
||
db.refresh(redelivery)
|
||
return {
|
||
"delivery_id": str(redelivery.id),
|
||
"http_status": redelivery.http_status,
|
||
"error_message": redelivery.error_message,
|
||
}
|
||
|
||
|
||
@router.delete(
|
||
"/api/core/webhooks/{webhook_id}",
|
||
status_code=204,
|
||
responses={204: {"description": "Webhook deleted"}},
|
||
)
|
||
def delete_webhook(
|
||
webhook_id: uuid.UUID,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
webhook = (
|
||
db.query(Webhook)
|
||
.filter(Webhook.id == webhook_id, Webhook.user_id == user.id)
|
||
.first()
|
||
)
|
||
if webhook is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Webhook not found",
|
||
)
|
||
db.delete(webhook)
|
||
db.commit()
|
||
return None
|
||
|
||
|
||
@router.post("/api/core/auth/login", response_model=LoginResponse)
|
||
def login_user(payload: LoginRequest, db: Session = Depends(get_db)):
|
||
payment_address = payload.payment_address.strip()
|
||
view_key = payload.view_key.strip()
|
||
_validate_payment_address_and_view_key(payment_address, view_key)
|
||
user = db.query(User).filter(User.payment_address == payment_address).first()
|
||
if user is None:
|
||
if not OPEN_REGISTRATION:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="registration_closed",
|
||
)
|
||
api_key = generate_api_key()
|
||
webhook_secret = generate_webhook_secret()
|
||
user = User(
|
||
payment_address=payment_address,
|
||
view_key_encrypted=encrypt_secret(view_key),
|
||
api_key_hash=hash_api_key(api_key),
|
||
api_key_encrypted=encrypt_api_key(api_key),
|
||
webhook_secret_encrypted=encrypt_secret(webhook_secret),
|
||
)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
return {
|
||
"api_key": api_key,
|
||
"webhook_secret": webhook_secret,
|
||
"store_id": user.id,
|
||
}
|
||
current_view_key = decrypt_secret(user.view_key_encrypted)
|
||
if current_view_key != view_key:
|
||
history_entry = ProfileHistory(
|
||
user_id=user.id,
|
||
field_name="view_key",
|
||
old_value=user.view_key_encrypted,
|
||
new_value=encrypt_secret(view_key),
|
||
value_encrypted=True,
|
||
)
|
||
user.view_key_encrypted = encrypt_secret(view_key)
|
||
db.add(history_entry)
|
||
db.commit()
|
||
db.refresh(user)
|
||
webhook_secret = _ensure_webhook_secret(db, user)
|
||
return {
|
||
"api_key": decrypt_api_key(user.api_key_encrypted),
|
||
"webhook_secret": webhook_secret,
|
||
"store_id": user.id,
|
||
}
|
||
|
||
|
||
@router.post("/api/core/auth/validate", status_code=status.HTTP_204_NO_CONTENT)
|
||
def validate_login(payload: LoginRequest) -> Response:
|
||
payment_address = payload.payment_address.strip()
|
||
view_key = payload.view_key.strip()
|
||
_validate_payment_address_and_view_key(payment_address, view_key)
|
||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||
|
||
|
||
@router.get("/api/core/auth/registration-status")
|
||
def registration_status():
|
||
return {"open": OPEN_REGISTRATION}
|
||
|
||
|
||
@router.post(
|
||
"/api/core/api-credentials/reset",
|
||
response_model=ApiCredentialsResetResponse,
|
||
)
|
||
def reset_api_credentials(
|
||
payload: ApiCredentialsResetRequest,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
if not payload.reset_api_key and not payload.reset_webhook_secret:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Select at least one credential to reset",
|
||
)
|
||
response: dict[str, str] = {}
|
||
if payload.reset_api_key:
|
||
new_key = generate_api_key()
|
||
new_key_encrypted = encrypt_api_key(new_key)
|
||
history_entry = ProfileHistory(
|
||
user_id=user.id,
|
||
field_name="api_key",
|
||
old_value=user.api_key_encrypted,
|
||
new_value=new_key_encrypted,
|
||
value_encrypted=True,
|
||
)
|
||
user.api_key_hash = hash_api_key(new_key)
|
||
user.api_key_encrypted = new_key_encrypted
|
||
db.add(history_entry)
|
||
response["api_key"] = new_key
|
||
if payload.reset_webhook_secret:
|
||
new_secret = generate_webhook_secret()
|
||
new_secret_encrypted = encrypt_secret(new_secret)
|
||
history_entry = ProfileHistory(
|
||
user_id=user.id,
|
||
field_name="webhook_secret",
|
||
old_value=user.webhook_secret_encrypted,
|
||
new_value=new_secret_encrypted,
|
||
value_encrypted=True,
|
||
)
|
||
user.webhook_secret_encrypted = new_secret_encrypted
|
||
db.add(history_entry)
|
||
response["webhook_secret"] = new_secret
|
||
db.commit()
|
||
db.refresh(user)
|
||
return response
|
||
|
||
|
||
@router.get("/api/core/profile", response_model=ProfileResponse)
|
||
def get_profile(
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
return user
|
||
|
||
|
||
@router.patch("/api/core/profile", response_model=ProfileResponse)
|
||
def update_profile(
|
||
payload: ProfileUpdate,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
if not payload.model_fields_set:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Select at least one profile field to update",
|
||
)
|
||
if (
|
||
"btcpay_checkout_style" in payload.model_fields_set
|
||
and payload.btcpay_checkout_style != user.btcpay_checkout_style
|
||
):
|
||
history_entry = ProfileHistory(
|
||
user_id=user.id,
|
||
field_name="btcpay_checkout_style",
|
||
old_value=user.btcpay_checkout_style,
|
||
new_value=payload.btcpay_checkout_style,
|
||
value_encrypted=False,
|
||
)
|
||
user.btcpay_checkout_style = payload.btcpay_checkout_style
|
||
db.add(history_entry)
|
||
if (
|
||
"default_confirmation_target" in payload.model_fields_set
|
||
and payload.default_confirmation_target is not None
|
||
and payload.default_confirmation_target != user.default_confirmation_target
|
||
):
|
||
history_entry = ProfileHistory(
|
||
user_id=user.id,
|
||
field_name="default_confirmation_target",
|
||
old_value=str(user.default_confirmation_target),
|
||
new_value=str(payload.default_confirmation_target),
|
||
value_encrypted=False,
|
||
)
|
||
user.default_confirmation_target = payload.default_confirmation_target
|
||
db.add(history_entry)
|
||
if "default_qr_logo" in payload.model_fields_set:
|
||
if payload.default_qr_logo is not None and payload.default_qr_logo != user.default_qr_logo:
|
||
history_entry = ProfileHistory(
|
||
user_id=user.id,
|
||
field_name="default_qr_logo",
|
||
old_value=user.default_qr_logo,
|
||
new_value=payload.default_qr_logo,
|
||
value_encrypted=False,
|
||
)
|
||
user.default_qr_logo = payload.default_qr_logo
|
||
db.add(history_entry)
|
||
if "default_qr_logo_data_url" in payload.model_fields_set:
|
||
if payload.default_qr_logo_data_url is not None:
|
||
value = payload.default_qr_logo_data_url.strip()
|
||
if value and len(value) > MAX_QR_LOGO_DATA_URL_LENGTH:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="QR logo image is too large",
|
||
)
|
||
if value == "":
|
||
payload.default_qr_logo_data_url = None
|
||
if payload.default_qr_logo_data_url != user.default_qr_logo_data_url:
|
||
history_entry = ProfileHistory(
|
||
user_id=user.id,
|
||
field_name="default_qr_logo_data_url",
|
||
old_value=user.default_qr_logo_data_url,
|
||
new_value=payload.default_qr_logo_data_url,
|
||
value_encrypted=False,
|
||
)
|
||
user.default_qr_logo_data_url = payload.default_qr_logo_data_url
|
||
db.add(history_entry)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
return user
|
||
|
||
|
||
@router.post("/api/core/invoices", response_model=InvoiceResponse, status_code=201)
|
||
def create_invoice_for_user(
|
||
payload: InvoiceCreateUser,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
if not user.payment_address or not user.view_key_encrypted:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail="Primary address is not configured for this user",
|
||
)
|
||
confirmation_target = payload.confirmation_target
|
||
if "confirmation_target" not in payload.model_fields_set:
|
||
confirmation_target = user.default_confirmation_target
|
||
expires_at = payload.expires_at or _default_invoice_expiry()
|
||
resolved_amount, warnings, quote = _resolve_invoice_amount(
|
||
db,
|
||
user=user,
|
||
requested_amount_xmr=payload.amount_xmr,
|
||
requested_amount_fiat=payload.amount_fiat,
|
||
requested_currency=payload.currency,
|
||
)
|
||
if warnings is not None and len(warnings) == 0:
|
||
warnings = None
|
||
metadata = payload.metadata or {}
|
||
if quote:
|
||
metadata = {**metadata, "quote": quote}
|
||
if isinstance(metadata, dict):
|
||
qr = metadata.get("qr")
|
||
if not isinstance(qr, dict):
|
||
qr = None
|
||
if qr is None:
|
||
qr = {"logo": user.default_qr_logo}
|
||
if user.default_qr_logo == "custom" and user.default_qr_logo_data_url:
|
||
qr["logo_data_url"] = user.default_qr_logo_data_url
|
||
metadata = {**metadata, "qr": qr}
|
||
else:
|
||
logo = qr.get("logo")
|
||
logo_data_url = qr.get("logo_data_url")
|
||
if isinstance(logo, str) and logo in ("monero", "none", "custom"):
|
||
normalized = {"logo": logo}
|
||
if logo == "custom":
|
||
if isinstance(logo_data_url, str):
|
||
trimmed = logo_data_url.strip()
|
||
if trimmed and len(trimmed) > MAX_QR_LOGO_DATA_URL_LENGTH:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="QR logo image is too large",
|
||
)
|
||
if trimmed:
|
||
normalized["logo_data_url"] = trimmed
|
||
metadata = {**metadata, "qr": normalized}
|
||
else:
|
||
metadata = {"qr": {"logo": user.default_qr_logo}}
|
||
if user.default_qr_logo == "custom" and user.default_qr_logo_data_url:
|
||
metadata["qr"]["logo_data_url"] = user.default_qr_logo_data_url
|
||
if payload.checkout_continue_url is not None:
|
||
checkout = metadata.get("checkout") if isinstance(metadata, dict) else None
|
||
if not isinstance(checkout, dict):
|
||
checkout = {}
|
||
checkout = {**checkout, "continue_url": str(payload.checkout_continue_url)}
|
||
metadata = {**(metadata if isinstance(metadata, dict) else {}), "checkout": checkout}
|
||
address, subaddress_index = create_subaddress_for_user(db, user=user)
|
||
invoice = Invoice(
|
||
user_id=user.id,
|
||
wallet_address=user.payment_address,
|
||
address=address,
|
||
subaddress_index=subaddress_index,
|
||
amount_xmr=resolved_amount,
|
||
status="pending",
|
||
confirmation_target=confirmation_target,
|
||
total_paid_atomic=0,
|
||
metadata_json=metadata,
|
||
expires_at=expires_at,
|
||
)
|
||
db.add(invoice)
|
||
db.commit()
|
||
db.refresh(invoice)
|
||
try:
|
||
ensure_invoice_qr_png(
|
||
invoice=invoice,
|
||
storage_dir=QR_STORAGE_DIR,
|
||
settings=resolve_qr_settings(invoice),
|
||
)
|
||
except HTTPException:
|
||
pass
|
||
dispatch_webhooks(db, str(user.id), "invoice.created", invoice)
|
||
return _invoice_response(invoice, request, warnings=warnings)
|
||
|
||
|
||
@router.get("/api/core/invoices", response_model=InvoiceListResponse)
|
||
def list_invoices_for_user(
|
||
request: Request,
|
||
limit: int = Query(default=50, ge=1, le=100),
|
||
offset: int = Query(default=0, ge=0),
|
||
status_filter: str | None = Query(default=None, alias="status"),
|
||
include_archived: bool = Query(default=False, alias="include_archived"),
|
||
q: str | None = Query(default=None),
|
||
sort: str = Query(default="created_at"),
|
||
order: str = Query(default="desc"),
|
||
created_from: datetime | None = Query(default=None),
|
||
created_to: datetime | None = Query(default=None),
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
query = db.query(Invoice).filter(Invoice.user_id == user.id)
|
||
if not include_archived:
|
||
query = query.filter(Invoice.archived_at.is_(None))
|
||
if status_filter:
|
||
query = query.filter(Invoice.status == status_filter)
|
||
if q:
|
||
needle = q.strip()
|
||
if needle:
|
||
try:
|
||
invoice_uuid = uuid.UUID(needle)
|
||
except ValueError:
|
||
invoice_uuid = None
|
||
try:
|
||
amount_xmr = Decimal(needle)
|
||
except (InvalidOperation, ValueError):
|
||
amount_xmr = None
|
||
matchers = []
|
||
if invoice_uuid is not None:
|
||
matchers.append(Invoice.id == invoice_uuid)
|
||
if amount_xmr is not None:
|
||
matchers.append(Invoice.amount_xmr == amount_xmr)
|
||
matchers.append(Invoice.address.ilike(f"%{needle}%"))
|
||
matchers.append(cast(Invoice.metadata_json, SqlString).ilike(f"%{needle}%"))
|
||
query = query.filter(or_(*matchers))
|
||
if created_from:
|
||
query = query.filter(Invoice.created_at >= created_from)
|
||
if created_to:
|
||
query = query.filter(Invoice.created_at <= created_to)
|
||
|
||
sort_keys = {
|
||
"created_at": Invoice.created_at,
|
||
"expires_at": Invoice.expires_at,
|
||
"amount_xmr": Invoice.amount_xmr,
|
||
"status": Invoice.status,
|
||
"confirmations": Invoice.confirmations,
|
||
"confirmation_target": Invoice.confirmation_target,
|
||
}
|
||
sort_column = sort_keys.get(sort, Invoice.created_at)
|
||
if order.lower() == "asc":
|
||
query = query.order_by(sort_column.asc())
|
||
else:
|
||
query = query.order_by(sort_column.desc())
|
||
items = (
|
||
query.offset(offset)
|
||
.limit(limit)
|
||
.all()
|
||
)
|
||
invoice_items = [_invoice_response(invoice, request) for invoice in items]
|
||
total_query = db.query(func.count(Invoice.id)).filter(Invoice.user_id == user.id)
|
||
if not include_archived:
|
||
total_query = total_query.filter(Invoice.archived_at.is_(None))
|
||
if status_filter:
|
||
total_query = total_query.filter(Invoice.status == status_filter)
|
||
if q:
|
||
needle = q.strip()
|
||
if needle:
|
||
try:
|
||
invoice_uuid = uuid.UUID(needle)
|
||
except ValueError:
|
||
invoice_uuid = None
|
||
try:
|
||
amount_xmr = Decimal(needle)
|
||
except (InvalidOperation, ValueError):
|
||
amount_xmr = None
|
||
matchers = []
|
||
if invoice_uuid is not None:
|
||
matchers.append(Invoice.id == invoice_uuid)
|
||
if amount_xmr is not None:
|
||
matchers.append(Invoice.amount_xmr == amount_xmr)
|
||
matchers.append(Invoice.address.ilike(f"%{needle}%"))
|
||
matchers.append(cast(Invoice.metadata_json, SqlString).ilike(f"%{needle}%"))
|
||
total_query = total_query.filter(or_(*matchers))
|
||
if created_from:
|
||
total_query = total_query.filter(Invoice.created_at >= created_from)
|
||
if created_to:
|
||
total_query = total_query.filter(Invoice.created_at <= created_to)
|
||
total = total_query.scalar() or 0
|
||
return {"items": invoice_items, "total": total}
|
||
|
||
|
||
@router.get(
|
||
"/api/core/invoices/export.csv",
|
||
responses={200: {"content": {"text/csv": {}}}},
|
||
)
|
||
def export_invoices_csv(
|
||
include_archived: bool = Query(default=False, alias="include_archived"),
|
||
status_filter: str | None = Query(default=None, alias="status"),
|
||
q: str | None = Query(default=None),
|
||
sort: str = Query(default="created_at"),
|
||
order: str = Query(default="desc"),
|
||
created_from: datetime | None = Query(default=None),
|
||
created_to: datetime | None = Query(default=None),
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
|
||
query = db.query(Invoice).filter(Invoice.user_id == user.id)
|
||
if not include_archived:
|
||
query = query.filter(Invoice.archived_at.is_(None))
|
||
if status_filter:
|
||
query = query.filter(Invoice.status == status_filter)
|
||
if q:
|
||
needle = q.strip()
|
||
if needle:
|
||
try:
|
||
invoice_uuid = uuid.UUID(needle)
|
||
except ValueError:
|
||
invoice_uuid = None
|
||
matchers = []
|
||
if invoice_uuid is not None:
|
||
matchers.append(Invoice.id == invoice_uuid)
|
||
matchers.append(Invoice.address.ilike(f"%{needle}%"))
|
||
matchers.append(cast(Invoice.metadata_json, SqlString).ilike(f"%{needle}%"))
|
||
query = query.filter(or_(*matchers))
|
||
if created_from:
|
||
query = query.filter(Invoice.created_at >= created_from)
|
||
if created_to:
|
||
query = query.filter(Invoice.created_at <= created_to)
|
||
|
||
sort_keys = {
|
||
"created_at": Invoice.created_at,
|
||
"expires_at": Invoice.expires_at,
|
||
"amount_xmr": Invoice.amount_xmr,
|
||
"status": Invoice.status,
|
||
"confirmations": Invoice.confirmations,
|
||
"confirmation_target": Invoice.confirmation_target,
|
||
}
|
||
sort_column = sort_keys.get(sort, Invoice.created_at)
|
||
if order.lower() == "asc":
|
||
query = query.order_by(sort_column.asc())
|
||
else:
|
||
query = query.order_by(sort_column.desc())
|
||
|
||
def _row_writer():
|
||
buffer = io.StringIO()
|
||
writer = csv.writer(buffer)
|
||
writer.writerow(
|
||
[
|
||
"id",
|
||
"status",
|
||
"amount_xmr",
|
||
"confirmation_target",
|
||
"confirmations",
|
||
"paid_after_expiry",
|
||
"paid_after_expiry_at",
|
||
"address",
|
||
"subaddress_index",
|
||
"created_at",
|
||
"expires_at",
|
||
"detected_at",
|
||
"confirmed_at",
|
||
"archived_at",
|
||
"metadata",
|
||
]
|
||
)
|
||
yield buffer.getvalue()
|
||
buffer.seek(0)
|
||
buffer.truncate(0)
|
||
|
||
for invoice in query.yield_per(500):
|
||
metadata = invoice.metadata_json
|
||
if metadata is None:
|
||
metadata_value = ""
|
||
else:
|
||
metadata_value = str(metadata)
|
||
writer.writerow(
|
||
[
|
||
str(invoice.id),
|
||
invoice.status,
|
||
format_xmr_amount(invoice.amount_xmr),
|
||
invoice.confirmation_target,
|
||
invoice.confirmations or 0,
|
||
bool(invoice.paid_after_expiry),
|
||
invoice.paid_after_expiry_at.isoformat()
|
||
if invoice.paid_after_expiry_at
|
||
else "",
|
||
invoice.address,
|
||
invoice.subaddress_index if invoice.subaddress_index is not None else "",
|
||
invoice.created_at.isoformat() if invoice.created_at else "",
|
||
invoice.expires_at.isoformat() if invoice.expires_at else "",
|
||
invoice.detected_at.isoformat() if invoice.detected_at else "",
|
||
invoice.confirmed_at.isoformat() if invoice.confirmed_at else "",
|
||
invoice.archived_at.isoformat() if invoice.archived_at else "",
|
||
metadata_value,
|
||
]
|
||
)
|
||
yield buffer.getvalue()
|
||
buffer.seek(0)
|
||
buffer.truncate(0)
|
||
|
||
filename = f"invoices-{datetime.now(timezone.utc).date().isoformat()}.csv"
|
||
headers = {"Content-Disposition": f'attachment; filename="{filename}"'}
|
||
return StreamingResponse(_row_writer(), media_type="text/csv", headers=headers)
|
||
|
||
|
||
@router.get("/api/core/invoices/{invoice_id}", response_model=InvoiceResponse)
|
||
def get_invoice_for_user(
|
||
invoice_id: uuid.UUID,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
invoice = (
|
||
db.query(Invoice)
|
||
.filter(
|
||
Invoice.id == invoice_id,
|
||
Invoice.user_id == user.id,
|
||
Invoice.archived_at.is_(None),
|
||
)
|
||
.first()
|
||
)
|
||
if invoice is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Invoice not found",
|
||
)
|
||
return _invoice_response(invoice, request)
|
||
|
||
|
||
@router.delete(
|
||
"/api/core/invoices/{invoice_id}",
|
||
status_code=204,
|
||
responses={204: {"description": "Invoice archived"}},
|
||
)
|
||
def archive_invoice_for_user(
|
||
invoice_id: uuid.UUID,
|
||
db: Session = Depends(get_db),
|
||
api_key: str = Depends(require_api_key),
|
||
):
|
||
user = _get_user_for_api_key(db, api_key)
|
||
if user is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="User API key required",
|
||
)
|
||
invoice = (
|
||
db.query(Invoice)
|
||
.filter(
|
||
Invoice.id == invoice_id,
|
||
Invoice.user_id == user.id,
|
||
Invoice.archived_at.is_(None),
|
||
)
|
||
.first()
|
||
)
|
||
if invoice is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Invoice not found",
|
||
)
|
||
if invoice.status not in ("pending", "expired", "invalid"):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail="Only pending, expired, or invalid invoices can be archived",
|
||
)
|
||
invoice.archived_at = datetime.now(timezone.utc)
|
||
db.add(invoice)
|
||
db.commit()
|
||
return None
|