""" Periodic cloud check-in. Runs every 5 minutes as an asyncio background task. Grace period: 72 hours (3 days) before marking unlicensed on connectivity failure. Lock behaviour: - cloud sets locked=true → set lock_pending=true in state - lock_pending is enforced at workday-close time (see business_day router) - while a workday is open, the site keeps running; lock applies once it closes License expiry behaviour: - 5 days before expiry → warning only (days_until_expiry in state) - on expiry → 5-day grace period begins (grace_expires_at in state) - after grace + no open workday → licensed=False enforced by business_day router """ import asyncio import json import logging import socket from datetime import datetime, timedelta, timezone from pathlib import Path import httpx from config import settings from middleware.license_check import license_state ORDER_POLL_INTERVAL = settings.CONNECT_SYNC_INTERVAL_SECONDS logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) SYNC_INTERVAL_SECONDS = 5 * 60 # 5 minutes GRACE_HOURS = 72 # 3 days offline grace EXPIRY_GRACE_DAYS = 5 # days after expiry before blocking EXPIRY_WARNING_DAYS = 5 # days before expiry to show warning STATE_FILE = Path(__file__).parent.parent / "license_state.json" def _load_persisted_state(): if STATE_FILE.exists(): try: data = json.loads(STATE_FILE.read_text()) license_state.update(data) logger.info("Loaded persisted license state: %s", data) except Exception as e: logger.warning("Could not load license state file: %s", 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 _compute_expiry_fields(expires_at_str: str | None) -> dict: """Return days_until_expiry and grace_expires_at derived from expires_at.""" if not expires_at_str: return {"days_until_expiry": None, "grace_expires_at": None} try: expires_at = datetime.fromisoformat(expires_at_str) if expires_at.tzinfo is None: expires_at = expires_at.replace(tzinfo=timezone.utc) except ValueError: return {"days_until_expiry": None, "grace_expires_at": None} now = datetime.now(timezone.utc) days_until = (expires_at - now).days # negative once expired grace_expires_at = None if days_until < 0: grace_expires_at = (expires_at + timedelta(days=EXPIRY_GRACE_DAYS)).isoformat() return { "days_until_expiry": days_until, "grace_expires_at": grace_expires_at, } def _get_local_ip() -> str | None: """Best-effort detection of the host machine's LAN IP address. When running inside Docker the socket trick returns the container/bridge IP, so we honour HOST_IP if it is explicitly provided via the environment.""" import os if host_ip := os.environ.get("HOST_IP", "").strip(): return host_ip try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: s.connect(("8.8.8.8", 80)) return s.getsockname()[0] except Exception: return None 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() licensed = data.get("licensed", True) cloud_locked = data.get("locked", False) expires_at = data.get("expires_at") expiry_fields = _compute_expiry_fields(expires_at) # If cloud says locked, check whether a workday is currently open. # No open workday → lock immediately. # Open workday → defer to workday close (business_day router enforces it). if cloud_locked: from database import SessionLocal from models.business_day import BusinessDay db = SessionLocal() try: open_day = db.query(BusinessDay).filter(BusinessDay.status == "open").first() finally: db.close() if open_day: if not license_state.get("lock_pending"): license_state["lock_pending"] = True logger.info("Cloud requested lock — workday open, deferring to workday close") else: license_state["lock_pending"] = False license_state["locked"] = True logger.info("Cloud requested lock — no open workday, locking immediately") # If cloud lifts the lock, clear pending too if not cloud_locked: license_state["lock_pending"] = False license_state["locked"] = False license_state.update({ "licensed": licensed, "expires_at": expires_at, "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, **expiry_fields, }) _persist_state() logger.info("Cloud sync OK: licensed=%s locked=%s expires_at=%s", licensed, cloud_locked, expires_at) except Exception as e: logger.warning("Cloud sync failed: %s", e) license_state["sync_failed"] = True last_sync_str = license_state.get("last_sync") if last_sync_str: try: last_sync = datetime.fromisoformat(last_sync_str) grace_expires = last_sync + timedelta(hours=GRACE_HOURS) if datetime.now(timezone.utc) > grace_expires: logger.error("72-hour offline grace period expired — marking unlicensed") license_state["licensed"] = False except ValueError: pass # Recompute expiry fields from cached expires_at even when offline expiry_fields = _compute_expiry_fields(license_state.get("expires_at")) license_state.update(expiry_fields) 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 = [] 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() ) 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, }) finally: db.close() 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)) 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.role.in_(["sysadmin", "manager"])).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() 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