feat(loyalty): pair POS terminal devices with one-time setup QR
Some checks failed
CI / ruff (push) Successful in 47s
CI / validate (push) Has been cancelled
CI / dependency-scanning (push) Has been cancelled
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled
CI / pytest (push) Has been cancelled

Adds the backend half of the Android tablet rollout. Merchants can
pair tablets to specific stores from /merchants/loyalty/devices (or
admins can pair on behalf from the merchant detail page). Each
pairing issues a long-lived JWT shown ONCE in the response with a
server-rendered QR PNG containing {api_url, store_code, auth_token} —
the tablet scans it on first boot and persists the three fields.

The store API (/api/v1/store/loyalty/*) now accepts these device JWTs
alongside user JWTs. Revoking a device row immediately rejects its
token (401 TERMINAL_DEVICE_REVOKED). Tokens expire after 1 year;
re-pair to renew.

- Migration loyalty_010 + TerminalDevice model
- create_device_token / verify_device_token JWT helpers
- 5 endpoints x 2 portals (merchant + admin on-behalf)
- Bearer-auth wiring in app/api/deps.py
- Pages, shared list partial with one-time pairing-QR modal,
  Alpine.js factories
- Locale strings (en authoritative; fr/de/lb seeded with EN copy
  for translation)
- 6 integration tests covering pair, list, revoke, idempotency,
  cross-merchant rejection, store-API auth via device JWT

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-05 20:18:57 +02:00
parent c7ab5eb900
commit 6276e9e3ac
28 changed files with 1971 additions and 7 deletions

View File

@@ -34,6 +34,10 @@ from app.modules.loyalty.services.stamp_service import (
StampService,
stamp_service,
)
from app.modules.loyalty.services.terminal_device_service import (
TerminalDeviceService,
terminal_device_service,
)
from app.modules.loyalty.services.wallet_service import (
WalletService,
wallet_service,
@@ -56,4 +60,6 @@ __all__ = [
"google_wallet_service",
"AppleWalletService",
"apple_wallet_service",
"TerminalDeviceService",
"terminal_device_service",
]

View File

@@ -0,0 +1,255 @@
# app/modules/loyalty/services/terminal_device_service.py
"""Terminal device service.
Pairs Android (or other) POS tablets to a single store. Issues a long-lived
JWT at pair time — the token itself is shown once and never persisted; only
the `jti` lives on the device row, which remains the source of truth for
revocation.
Two failure modes render a token unusable:
- Explicit `revoked_at` (merchant action, row stays for audit)
- `expires_at < now` (1y default, no auto-renewal — re-pair the tablet)
"""
import base64
import io
import json
import logging
import uuid
from datetime import UTC, datetime, timedelta
import qrcode
from sqlalchemy.orm import Session
from app.core.config import settings
from app.modules.loyalty.exceptions import TerminalDeviceNotFoundException
from app.modules.loyalty.models import TerminalDevice
from app.modules.loyalty.schemas.terminal_device import (
TerminalDeviceCreate,
TerminalDeviceUpdate,
)
from middleware.auth import AuthManager
logger = logging.getLogger(__name__)
DEFAULT_TOKEN_LIFETIME = timedelta(days=365)
class TerminalDeviceService:
"""Service for pairing and managing terminal devices."""
def __init__(self):
self._auth = AuthManager()
# =========================================================================
# Reads
# =========================================================================
def get_device(self, db: Session, device_id: int) -> TerminalDevice | None:
return (
db.query(TerminalDevice)
.filter(TerminalDevice.id == device_id)
.first()
)
def require_device(self, db: Session, device_id: int) -> TerminalDevice:
device = self.get_device(db, device_id)
if not device:
raise TerminalDeviceNotFoundException(str(device_id))
return device
def require_device_for_merchant(
self, db: Session, device_id: int, merchant_id: int
) -> TerminalDevice:
"""Fetch a device but enforce it belongs to the given merchant."""
device = self.require_device(db, device_id)
if device.merchant_id != merchant_id:
raise TerminalDeviceNotFoundException(str(device_id))
return device
def get_device_by_jti(self, db: Session, jti: str) -> TerminalDevice | None:
return db.query(TerminalDevice).filter(TerminalDevice.jti == jti).first()
def list_devices(
self,
db: Session,
merchant_id: int,
*,
store_id: int | None = None,
include_revoked: bool = False,
) -> list[TerminalDevice]:
query = db.query(TerminalDevice).filter(
TerminalDevice.merchant_id == merchant_id
)
if store_id is not None:
query = query.filter(TerminalDevice.store_id == store_id)
if not include_revoked:
query = query.filter(TerminalDevice.revoked_at.is_(None))
return query.order_by(
TerminalDevice.store_id, TerminalDevice.created_at.desc()
).all()
# =========================================================================
# Writes
# =========================================================================
def create_device(
self,
db: Session,
*,
merchant_id: int,
data: TerminalDeviceCreate,
created_by_user_id: int | None,
) -> tuple[TerminalDevice, str, dict, str]:
"""Pair a new device and issue its setup token.
Returns ``(device, setup_token, setup_payload, qr_png_base64)``.
The caller must NOT persist ``setup_token`` — only the device row's
``jti`` is kept.
"""
from app.modules.tenancy.models import Store
store = (
db.query(Store)
.filter(Store.id == data.store_id, Store.merchant_id == merchant_id)
.first()
)
if not store:
raise TerminalDeviceNotFoundException(f"store:{data.store_id}")
jti = str(uuid.uuid4())
expires_at = datetime.now(UTC) + DEFAULT_TOKEN_LIFETIME
device = TerminalDevice(
merchant_id=merchant_id,
store_id=store.id,
label=data.label,
jti=jti,
expires_at=expires_at,
created_by_id=created_by_user_id,
)
db.add(device)
db.commit()
db.refresh(device)
token = self._auth.create_device_token(
device_id=device.id,
jti=jti,
merchant_id=merchant_id,
store_id=store.id,
store_code=store.store_code,
expires_at=expires_at,
)
payload = {
"api_url": settings.app_base_url,
"store_code": store.store_code,
"auth_token": token,
}
qr_png_base64 = self._render_qr_png_base64(json.dumps(payload))
logger.info(
f"Paired terminal device {device.id} ('{device.label}') for "
f"merchant {merchant_id}, store {store.id} ({store.store_code})"
)
return device, token, payload, qr_png_base64
def update_device(
self,
db: Session,
device: TerminalDevice,
data: TerminalDeviceUpdate,
) -> TerminalDevice:
if data.label is not None:
device.label = data.label
db.commit()
db.refresh(device)
return device
def revoke_device(
self,
db: Session,
device: TerminalDevice,
*,
revoked_by_user_id: int | None,
reason: str | None = None,
) -> TerminalDevice:
if device.revoked_at is not None:
return device
device.revoked_at = datetime.now(UTC)
device.revoked_by_id = revoked_by_user_id
device.revoked_reason = reason
db.commit()
db.refresh(device)
logger.info(
f"Revoked terminal device {device.id} by user {revoked_by_user_id}"
)
return device
def delete_device(self, db: Session, device: TerminalDevice) -> None:
device_id = device.id
db.delete(device)
db.commit()
logger.info(f"Deleted terminal device {device_id}")
def touch_last_seen(
self,
db: Session,
device: TerminalDevice,
ip: str | None = None,
) -> None:
"""Bump last_seen_at/ip on each authenticated request. Best-effort."""
device.last_seen_at = datetime.now(UTC)
if ip is not None:
device.last_seen_ip = ip
try:
db.commit()
except Exception: # noqa: EXC003 - best-effort heartbeat, must not break the request
db.rollback()
logger.warning(
f"Failed to update last_seen for device {device.id}", exc_info=True
)
# =========================================================================
# Helpers
# =========================================================================
@staticmethod
def _render_qr_png_base64(data: str) -> str:
qr = qrcode.QRCode(
version=None,
error_correction=qrcode.constants.ERROR_CORRECT_M,
box_size=8,
border=2,
)
qr.add_data(data)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buffer = io.BytesIO()
img.save(buffer, format="PNG")
encoded = base64.b64encode(buffer.getvalue()).decode("ascii")
return f"data:image/png;base64,{encoded}"
@staticmethod
def build_response(device: TerminalDevice) -> dict:
"""Hydrate a response dict including derived status + store name."""
return {
"id": device.id,
"merchant_id": device.merchant_id,
"store_id": device.store_id,
"store_name": device.store.name if device.store else None,
"label": device.label,
"status": device.status,
"expires_at": device.expires_at,
"last_seen_at": device.last_seen_at,
"last_seen_ip": device.last_seen_ip,
"revoked_at": device.revoked_at,
"revoked_by_id": device.revoked_by_id,
"revoked_reason": device.revoked_reason,
"created_by_id": device.created_by_id,
"created_at": device.created_at,
}
terminal_device_service = TerminalDeviceService()