mirror of
https://github.com/Such-Software/xmrcheckout.git
synced 2026-10-17 05:51:34 +02:00
jwinterm
b8efe702c1
Exposes the XMR→fiat spot rate the invoice path already uses internally (Kraken XMR-USD, 60s cache). No auth. Allows downstream services (Medusa storefront) to display a live USD reference price for XMR-denominated products without hard-coding a third-party rate provider on the storefront side.
1407 lines
49 KiB
Python
1407 lines
49 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 monero.address import Address, IntegratedAddress, SubAddress
|
|
from .rates import get_xmr_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:
|
|
total_paid = (Decimal(invoice.total_paid_atomic) / Decimal("1000000000000")).quantize(
|
|
Decimal("0.000000000001"), 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_xmr_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 5 decimal places so we never undercharge.
|
|
# 5 decimals = 0.00001 XMR ≈ $0.002 at $200/XMR — more than enough precision.
|
|
# See also: btcpay_routes.py for the same rounding.
|
|
# To change precision: adjust the Decimal quantize string here and in btcpay_routes.py.
|
|
amount_xmr = (Decimal(requested_amount_fiat) / quote.rate).quantize(
|
|
Decimal("0.00001"), 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:
|
|
try:
|
|
address = Address(payment_address)
|
|
except ValueError as exc:
|
|
try:
|
|
SubAddress(payment_address)
|
|
except ValueError:
|
|
try:
|
|
IntegratedAddress(payment_address)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Primary address is invalid",
|
|
) from exc
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Integrated addresses are not supported. Use the primary address.",
|
|
) from exc
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Subaddresses are not supported. Use the primary address.",
|
|
) from exc
|
|
if not address.check_private_view_key(view_key):
|
|
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 XMR→fiat spot rate. Returns the same QuoteResult that the
|
|
invoice-creation paths use internally (60s in-process cache, sourced
|
|
from Kraken XMR/USD).
|
|
|
|
No auth — exposes the same data invoice creation already uses,
|
|
so downstream consumers (Medusa) can get an authoritative-on-our-
|
|
side rate without hard-coding Kraken / a third-party API.
|
|
"""
|
|
try:
|
|
quote = get_xmr_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": "XMR",
|
|
}
|
|
|
|
|
|
@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
|