bonaminandClaude Opus 5.5
13a451a756
- services/license.py: verifies the cloud's Ed25519-signed license token
(public key built in) and decides purely: valid → licensed until expiry,
then a 5-day grace, then blocked - never mid-service (deferred while a
workday is open, applied at close). Works offline for as long as the
license lasts: the "unlicensed after 72h without heartbeat" rule is gone.
- Tamper resistance: an edited token fails the signature ("unverified");
a clock earlier than the latest provable time (token issued_at, newest
order in the DB, stored high-water mark; 1 day tolerance) → "clock".
- apply_license() re-evaluates from the stored token at startup, after
every heartbeat attempt and when a workday closes. Cloud lock/unlock from
the token keeps the workday-deferred behaviour. Transition: a cloud
without tokens is trusted 72h per successful heartbeat.
- FIX: the promised 5-day grace after expiry never happened - the cloud's
licensed=false was applied immediately (402 on everything).
- FIX: license_state.json lived inside the container and was lost on every
re-creation; it now lives in the data volume (old path read once).
- /api/system/status: offline_days, license_verified, license_problem,
grace_over; lock_reason "clock"/"unverified"; grace days from the license
module (rounded up).
Tests: 18 unit checks (signature, tamper, other site/key, 364 days
offline, grace ±workday, inactive, clock rollback, transition) + 17 E2E
checks with a real cloud + site process (400 days offline, tampered file →
402, clock behind newest order, expiry deferred until workday close,
renewal, remote lock/unlock).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
533 lines
20 KiB
Python
533 lines
20 KiB
Python
"""
|
|
Periodic cloud check-in. Runs every 5 minutes as an asyncio background task.
|
|
|
|
Licensing (KI-006, see services/license.py): the cloud is needed to RENEW a
|
|
license, not to RUN one. Each heartbeat returns a signed license token; the
|
|
site stores it and enforces it offline — until expiry (+5 days grace), however
|
|
long it has no internet. apply_license() re-evaluates the stored token without
|
|
network: at startup, after every heartbeat attempt and when a workday closes.
|
|
|
|
Lock behaviour (unchanged):
|
|
- cloud sets locked=true → lock_pending while a workday is open, locked once it closes
|
|
- expiry past grace / site deactivated → likewise never mid-service
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from config import settings
|
|
from middleware.license_check import license_state
|
|
from services.license import evaluate, parse_dt, verify_token
|
|
|
|
ORDER_POLL_INTERVAL = settings.CONNECT_SYNC_INTERVAL_SECONDS
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SYNC_INTERVAL_SECONDS = 5 * 60 # 5 minutes
|
|
|
|
|
|
def _data_dir() -> Path:
|
|
"""Directory of the SQLite database — the persistent data volume in Docker."""
|
|
from services.tls_identity import tls_dir
|
|
return tls_dir().parent
|
|
|
|
|
|
# In the data volume: it must survive container re-creation (image updates,
|
|
# compose changes), or an offline site would lose its license. Earlier versions
|
|
# kept it inside the container at local_backend/license_state.json.
|
|
STATE_FILE = _data_dir() / "license_state.json"
|
|
LEGACY_STATE_FILE = Path(__file__).parent.parent / "license_state.json"
|
|
|
|
|
|
def _load_persisted_state():
|
|
for path in (STATE_FILE, LEGACY_STATE_FILE):
|
|
if path.exists():
|
|
try:
|
|
license_state.update(json.loads(path.read_text()))
|
|
logger.info("Loaded persisted license state from %s", path)
|
|
return
|
|
except Exception as e:
|
|
logger.warning("Could not load license state file %s: %s", path, e)
|
|
|
|
|
|
def _persist_state():
|
|
try:
|
|
STATE_FILE.write_text(json.dumps(license_state))
|
|
except Exception as e:
|
|
logger.warning("Could not persist license state: %s", e)
|
|
|
|
|
|
def _latest_activity(db) -> datetime | None:
|
|
"""Newest order timestamp in the database — the clock can't be set before it."""
|
|
from sqlalchemy import func
|
|
from models.order import Order, OrderItem
|
|
stamps = [db.query(func.max(Order.opened_at)).scalar(), db.query(func.max(OrderItem.added_at)).scalar()]
|
|
stamps = [parse_dt(x) for x in stamps if x]
|
|
return max(stamps, default=None)
|
|
|
|
|
|
def apply_license(now: datetime | None = None) -> None:
|
|
"""Re-evaluate the license from the stored signed token (no network)."""
|
|
if not settings.SITE_ID:
|
|
return # dev / unregistered install: licensing off, as before
|
|
from database import SessionLocal
|
|
from models.business_day import BusinessDay
|
|
|
|
now = now or datetime.now(timezone.utc)
|
|
payload = verify_token(license_state.get("license_token"), settings.SITE_ID)
|
|
db = SessionLocal()
|
|
try:
|
|
workday_open = db.query(BusinessDay).filter(BusinessDay.status == "open").first() is not None
|
|
latest_activity = _latest_activity(db)
|
|
finally:
|
|
db.close()
|
|
|
|
floors = [parse_dt(license_state.get("time_high_water")), latest_activity,
|
|
parse_dt(payload["issued_at"]) if payload else None]
|
|
floor = max((f for f in floors if f), default=None)
|
|
result = evaluate(payload, now, floor, workday_open, parse_dt(license_state.get("legacy_ok_at")))
|
|
if result["license_problem"] != "clock":
|
|
license_state["time_high_water"] = max(floor or now, now).isoformat()
|
|
|
|
if payload: # lock requested by the cloud — deferred while a workday is open
|
|
if payload.get("locked"):
|
|
if workday_open and not license_state.get("locked"):
|
|
license_state["lock_pending"] = True
|
|
else:
|
|
license_state["lock_pending"] = False
|
|
license_state["locked"] = True
|
|
else:
|
|
license_state["lock_pending"] = False
|
|
license_state["locked"] = False
|
|
|
|
last_sync = parse_dt(license_state.get("last_sync"))
|
|
license_state.update({**result, "offline_days": (now - last_sync).days if last_sync else None})
|
|
_persist_state()
|
|
if result["license_problem"]:
|
|
logger.warning("License problem: %s (licensed=%s)", result["license_problem"], result["licensed"])
|
|
|
|
|
|
def _get_local_ip() -> str | None:
|
|
"""The server's LAN IP as phones see it — same resolver as the manager's
|
|
pairing QR (override → live detection → HOST_IP), see services/lan_ip.py.
|
|
No socket tricks: inside Docker they return the unreachable bridge IP."""
|
|
from database import SessionLocal
|
|
from services.lan_ip import resolve_lan_ip
|
|
db = SessionLocal()
|
|
try:
|
|
return resolve_lan_ip(db)["effective"]
|
|
except Exception:
|
|
return None
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
async def _sync_once():
|
|
if not settings.SITE_ID or not settings.CLOUD_URL:
|
|
logger.debug("No SITE_ID/CLOUD_URL configured — skipping cloud sync")
|
|
return
|
|
|
|
try:
|
|
local_ip = _get_local_ip()
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.post(
|
|
f"{settings.CLOUD_URL}/api/heartbeat/",
|
|
headers={
|
|
"X-Site-ID": settings.SITE_ID,
|
|
"X-Site-Key": settings.SITE_KEY,
|
|
},
|
|
json={"version": settings.VERSION, "uptime_seconds": 0, "local_ip": local_ip},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
token = data.get("license_token")
|
|
if token:
|
|
if verify_token(token, settings.SITE_ID):
|
|
license_state["license_token"] = token
|
|
else:
|
|
logger.error("Cloud sent a license token that does not verify - keeping the previous one")
|
|
else:
|
|
# Cloud without license signing (transition): trust this answer for 72h,
|
|
# keep the old immediate lock handling
|
|
if data.get("licensed", True):
|
|
license_state["legacy_ok_at"] = datetime.now(timezone.utc).isoformat()
|
|
license_state["expires_at"] = data.get("expires_at")
|
|
if data.get("locked"):
|
|
license_state["lock_pending"] = True
|
|
else:
|
|
license_state["lock_pending"] = False
|
|
license_state["locked"] = False
|
|
|
|
license_state.update({
|
|
"latest_version": data.get("latest_version"),
|
|
"waiter_domain": data.get("waiter_domain"),
|
|
"site_numeric_id": data.get("site_numeric_id"),
|
|
"last_sync": datetime.now(timezone.utc).isoformat(),
|
|
"sync_failed": False,
|
|
})
|
|
logger.info("Cloud sync OK (signed license: %s)", bool(token))
|
|
|
|
except Exception as e:
|
|
logger.warning("Cloud sync failed: %s", e)
|
|
license_state["sync_failed"] = True
|
|
|
|
# Online or not: the stored signed license decides (no more 72h offline rule)
|
|
apply_license()
|
|
|
|
IMAGE_DIR = Path("/app/data/product_images")
|
|
|
|
|
|
async def _push_product_images(db, products):
|
|
"""Upload the local image file for each product whose picture changed since
|
|
the last successful push (tracked via Product.cloud_image_hash), so the
|
|
public QR menu can show it without needing a manually-set digital_image_url.
|
|
Skips products that already have a manual digital_image_url override."""
|
|
import hashlib
|
|
|
|
site_numeric_id = license_state.get("site_numeric_id")
|
|
if not site_numeric_id:
|
|
return
|
|
|
|
for p in products:
|
|
if p.digital_image_url or not p.image_url:
|
|
continue
|
|
|
|
filename = os.path.basename(p.image_url)
|
|
filepath = IMAGE_DIR / filename
|
|
if not filepath.exists():
|
|
continue
|
|
|
|
try:
|
|
contents = filepath.read_bytes()
|
|
image_hash = hashlib.sha256(contents).hexdigest()
|
|
if image_hash == p.cloud_image_hash:
|
|
continue
|
|
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
resp = await client.post(
|
|
f"{settings.CLOUD_URL}/api/menu/sync-image",
|
|
headers={"X-Site-ID": settings.SITE_ID, "X-Site-Key": settings.SITE_KEY},
|
|
data={"product_id": str(p.id)},
|
|
files={"file": (filename, contents)},
|
|
)
|
|
resp.raise_for_status()
|
|
|
|
p.cloud_image_hash = image_hash
|
|
db.commit()
|
|
logger.info("Pushed image for product %d to cloud", p.id)
|
|
|
|
except Exception as e:
|
|
logger.warning("Image push failed for product %d: %s", p.id, e)
|
|
|
|
|
|
async def _push_menu_snapshot():
|
|
"""Serialize all digital-visible products+categories and POST to cloud."""
|
|
if not settings.SITE_ID or not settings.CLOUD_URL:
|
|
return
|
|
|
|
try:
|
|
from database import SessionLocal
|
|
from models.product import Category, Product
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
categories = db.query(Category).filter(Category.parent_id == None).all()
|
|
payload_categories = []
|
|
all_products = []
|
|
for cat in categories:
|
|
products = (
|
|
db.query(Product)
|
|
.filter(
|
|
Product.category_id == cat.id,
|
|
Product.digital_visible == 1,
|
|
Product.lifecycle_status == "active",
|
|
)
|
|
.order_by(Product.sort_order)
|
|
.all()
|
|
)
|
|
all_products.extend(products)
|
|
product_list = []
|
|
for p in products:
|
|
product_list.append({
|
|
"id": p.id,
|
|
"name": p.name,
|
|
"digital_name": p.digital_name,
|
|
"digital_description": p.digital_description,
|
|
"digital_price": p.digital_price,
|
|
"base_price": p.base_price,
|
|
"digital_discount": p.digital_discount,
|
|
"digital_available": bool(p.digital_available),
|
|
"digital_image_url": p.digital_image_url,
|
|
"image_url": p.image_url,
|
|
"quick_options": [
|
|
{"id": o.id, "name": o.name, "price": o.price}
|
|
for o in p.quick_options
|
|
],
|
|
})
|
|
if product_list:
|
|
payload_categories.append({
|
|
"id": cat.id,
|
|
"name": cat.name,
|
|
"sort_order": cat.sort_order,
|
|
"products": product_list,
|
|
})
|
|
|
|
snapshot_json = json.dumps({"categories": payload_categories})
|
|
|
|
# Resolve numeric site_id from license state (set by heartbeat response)
|
|
site_numeric_id = license_state.get("site_numeric_id")
|
|
if not site_numeric_id:
|
|
logger.debug("Menu push skipped — site_numeric_id not yet known")
|
|
return
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.post(
|
|
f"{settings.CLOUD_URL}/api/menu/sync",
|
|
headers={"X-Site-ID": settings.SITE_ID, "X-Site-Key": settings.SITE_KEY},
|
|
json={"site_id": site_numeric_id, "snapshot_json": snapshot_json},
|
|
)
|
|
resp.raise_for_status()
|
|
logger.info("Menu snapshot pushed (%d categories)", len(payload_categories))
|
|
|
|
await _push_product_images(db, all_products)
|
|
finally:
|
|
db.close()
|
|
|
|
except Exception as e:
|
|
logger.warning("Menu snapshot push failed: %s", e)
|
|
|
|
|
|
async def _pull_pending_orders():
|
|
"""Fetch online orders from cloud that haven't been synced to local yet."""
|
|
if not settings.SITE_ID or not settings.CLOUD_URL:
|
|
return
|
|
|
|
site_numeric_id = license_state.get("site_numeric_id")
|
|
if not site_numeric_id:
|
|
return
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(
|
|
f"{settings.CLOUD_URL}/api/orders/pending/{site_numeric_id}",
|
|
headers={"X-Site-ID": settings.SITE_ID, "X-Site-Key": settings.SITE_KEY},
|
|
)
|
|
resp.raise_for_status()
|
|
orders = resp.json()
|
|
|
|
if not orders:
|
|
return
|
|
|
|
from database import SessionLocal
|
|
from models.order import Order, OrderItem
|
|
from models.product import Product
|
|
from services.sse_bus import broadcast_sync
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Use a system user id=1 (first manager/sysadmin) as the opener
|
|
from models.user import User
|
|
system_user = db.query(User).filter(User.perm_access_dashboard == True).first()
|
|
opener_id = system_user.id if system_user else 1
|
|
|
|
for cloud_order in orders:
|
|
# Create local Order row — no table_id for online orders
|
|
local_order = Order(
|
|
table_id=None,
|
|
opened_by=opener_id,
|
|
status="open",
|
|
source="online",
|
|
online_order_ref=cloud_order["public_ref"],
|
|
online_order_cloud_id=cloud_order["id"],
|
|
online_status="pending_acceptance",
|
|
online_customer_name=cloud_order.get("customer_name"),
|
|
online_customer_phone=cloud_order.get("customer_phone"),
|
|
online_customer_address=cloud_order.get("customer_address"),
|
|
online_customer_notes=cloud_order.get("customer_notes"),
|
|
online_order_type=cloud_order.get("order_type"),
|
|
)
|
|
db.add(local_order)
|
|
db.flush() # get local_order.id
|
|
|
|
# Create OrderItem rows
|
|
items = json.loads(cloud_order.get("items_json", "[]"))
|
|
for item in items:
|
|
product = db.query(Product).filter(
|
|
Product.id == item.get("product_id")
|
|
).first()
|
|
db.add(OrderItem(
|
|
order_id=local_order.id,
|
|
product_id=item.get("product_id"),
|
|
added_by=opener_id,
|
|
quantity=item.get("quantity", 1),
|
|
unit_price=item.get("unit_price", 0.0),
|
|
selected_options=json.dumps(item.get("options", [])),
|
|
status="active",
|
|
printed=False,
|
|
))
|
|
|
|
db.commit()
|
|
|
|
# Mark synced on cloud
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
await client.post(
|
|
f"{settings.CLOUD_URL}/api/orders/{cloud_order['id']}/synced",
|
|
headers={"X-Site-ID": settings.SITE_ID, "X-Site-Key": settings.SITE_KEY},
|
|
json={"local_order_id": local_order.id},
|
|
)
|
|
|
|
logger.info("Online order %s pulled and created as local order %d",
|
|
cloud_order["public_ref"], local_order.id)
|
|
|
|
db.close()
|
|
|
|
# Broadcast SSE so the manager dashboard lights up
|
|
broadcast_sync("online_order_received", {"count": len(orders)})
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
db.close()
|
|
logger.error("Failed to process pulled orders: %s", e)
|
|
|
|
except Exception as e:
|
|
logger.warning("Order pull failed: %s", e)
|
|
|
|
|
|
async def _sync_loop():
|
|
_load_persisted_state()
|
|
apply_license() # decide from the stored license before the first network attempt
|
|
while True:
|
|
await _sync_once()
|
|
await asyncio.sleep(SYNC_INTERVAL_SECONDS)
|
|
|
|
|
|
async def _push_stats_snapshot():
|
|
"""Collect live operational stats and POST to cloud for the remote manager dashboard."""
|
|
if not settings.SITE_ID or not settings.CLOUD_URL:
|
|
return
|
|
|
|
site_numeric_id = license_state.get("site_numeric_id")
|
|
if not site_numeric_id:
|
|
return
|
|
|
|
try:
|
|
from database import SessionLocal
|
|
from models.order import Order, OrderItem
|
|
from models.business_day import BusinessDay
|
|
from models.shift import WaiterShift
|
|
from sqlalchemy import func
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
now = datetime.now(timezone.utc)
|
|
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
|
# Open tables = orders currently in status open or partially_paid
|
|
open_tables = db.query(Order).filter(
|
|
Order.status.in_(["open", "partially_paid"]),
|
|
Order.source == "pos",
|
|
).count()
|
|
|
|
# Today's completed POS orders (paid or closed today)
|
|
today_pos_orders = db.query(Order).filter(
|
|
Order.status.in_(["paid", "closed"]),
|
|
Order.source == "pos",
|
|
Order.closed_at >= today_start,
|
|
).all()
|
|
today_order_count = len(today_pos_orders)
|
|
|
|
# Today's revenue: sum of paid items on those orders
|
|
today_revenue = 0.0
|
|
for o in today_pos_orders:
|
|
for item in o.items:
|
|
if item.status in ("active", "paid"):
|
|
today_revenue += item.unit_price * item.quantity
|
|
|
|
# Online orders today
|
|
online_orders_today = db.query(Order).filter(
|
|
Order.source == "online",
|
|
Order.opened_at >= today_start,
|
|
).count()
|
|
|
|
online_orders_pending = db.query(Order).filter(
|
|
Order.source == "online",
|
|
Order.online_status == "pending_acceptance",
|
|
).count()
|
|
|
|
# Active business day
|
|
open_day = db.query(BusinessDay).filter(BusinessDay.status == "open").first()
|
|
current_shift = None
|
|
if open_day:
|
|
active_shift = db.query(WaiterShift).filter(
|
|
WaiterShift.business_day_id == open_day.id,
|
|
WaiterShift.ended_at == None,
|
|
).first()
|
|
if active_shift:
|
|
current_shift = {
|
|
"waiter_id": active_shift.waiter_id,
|
|
"started_at": active_shift.started_at.isoformat(),
|
|
}
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
snapshot = {
|
|
"open_tables": open_tables,
|
|
"today_revenue": round(today_revenue, 2),
|
|
"today_orders": today_order_count,
|
|
"online_orders_pending": online_orders_pending,
|
|
"online_orders_today": online_orders_today,
|
|
"current_shift": current_shift,
|
|
"as_of": now.isoformat(),
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.post(
|
|
f"{settings.CLOUD_URL}/api/remote/snapshot",
|
|
headers={"X-Site-ID": settings.SITE_ID, "X-Site-Key": settings.SITE_KEY},
|
|
json={"site_id": site_numeric_id, "snapshot_json": json.dumps(snapshot)},
|
|
)
|
|
resp.raise_for_status()
|
|
logger.info("Stats snapshot pushed: %s", snapshot)
|
|
|
|
except Exception as e:
|
|
logger.warning("Stats snapshot push failed: %s", e)
|
|
|
|
|
|
async def _connect_loop():
|
|
"""Faster loop: pulls pending online orders every CONNECT_SYNC_INTERVAL_SECONDS.
|
|
Also pushes menu snapshot and stats every 5 minutes (piggybacking on this loop).
|
|
On startup, waits for the first heartbeat to populate site_numeric_id then
|
|
immediately pushes menu + stats without waiting a full cycle."""
|
|
push_every = max(1, (5 * 60) // ORDER_POLL_INTERVAL)
|
|
|
|
# Wait for the heartbeat loop to get site_numeric_id, then do an immediate push
|
|
for _ in range(30): # wait up to 30 seconds
|
|
if license_state.get("site_numeric_id"):
|
|
break
|
|
await asyncio.sleep(1)
|
|
await _push_menu_snapshot()
|
|
await _push_stats_snapshot()
|
|
|
|
tick = 0
|
|
while True:
|
|
await asyncio.sleep(ORDER_POLL_INTERVAL)
|
|
await _pull_pending_orders()
|
|
tick += 1
|
|
if tick % push_every == 0:
|
|
await _push_menu_snapshot()
|
|
await _push_stats_snapshot()
|
|
|
|
|
|
async def start_cloud_sync() -> asyncio.Task:
|
|
heartbeat_task = asyncio.create_task(_sync_loop())
|
|
asyncio.create_task(_connect_loop())
|
|
return heartbeat_task
|