diff --git a/local_backend/config.py b/local_backend/config.py index 4f30ee7..7dd2455 100644 --- a/local_backend/config.py +++ b/local_backend/config.py @@ -20,6 +20,7 @@ class Settings(BaseSettings): SECRET_KEY: str = "change-me-generate-a-long-random-string" DATABASE_URL: str = f"sqlite:///{_DB_DEFAULT.as_posix()}" VERSION: str = "0.0.0" + CONNECT_SYNC_INTERVAL_SECONDS: int = 30 # how often to poll for pending online orders model_config = {"env_file": str(_HERE / ".env"), "env_file_encoding": "utf-8"} diff --git a/local_backend/main.py b/local_backend/main.py index 03d076d..a04cb06 100644 --- a/local_backend/main.py +++ b/local_backend/main.py @@ -28,6 +28,7 @@ from routers import flags as flags_router from routers import messages as messages_router from routers import sse as sse_router from routers import data_transfer as data_transfer_router +from routers import connect_orders as connect_orders_router def _run_migrations(): @@ -218,6 +219,15 @@ def _run_migrations(): "INSERT OR IGNORE INTO pos_settings (key, value, updated_at) VALUES ('security.auto_lock_seconds', '300', CURRENT_TIMESTAMP)", "INSERT OR IGNORE INTO pos_settings (key, value, updated_at) VALUES ('security.auto_logout', 'false', CURRENT_TIMESTAMP)", "INSERT OR IGNORE INTO pos_settings (key, value, updated_at) VALUES ('security.auto_logout_seconds', '1800', CURRENT_TIMESTAMP)", + # Xenia Connect — online order fields on orders table + "ALTER TABLE orders ADD COLUMN source VARCHAR NOT NULL DEFAULT 'pos'", + "ALTER TABLE orders ADD COLUMN online_order_ref VARCHAR", + "ALTER TABLE orders ADD COLUMN online_status VARCHAR", + "ALTER TABLE orders ADD COLUMN online_customer_name VARCHAR", + "ALTER TABLE orders ADD COLUMN online_customer_phone VARCHAR", + "ALTER TABLE orders ADD COLUMN online_customer_address TEXT", + "ALTER TABLE orders ADD COLUMN online_customer_notes TEXT", + "ALTER TABLE orders ADD COLUMN online_order_type VARCHAR", # Xenia Connect — digital menu fields on products "ALTER TABLE products ADD COLUMN digital_visible INTEGER NOT NULL DEFAULT 1", "ALTER TABLE products ADD COLUMN digital_available INTEGER NOT NULL DEFAULT 1", @@ -283,3 +293,4 @@ app.include_router(flags_router.router, prefix="/api/flags", tag app.include_router(messages_router.router, prefix="/api/messages", tags=["messages"]) app.include_router(sse_router.router, prefix="/api/sse", tags=["sse"]) app.include_router(data_transfer_router.router, prefix="/api/data-transfer", tags=["data-transfer"]) +app.include_router(connect_orders_router.router, prefix="/api/connect", tags=["connect"]) diff --git a/local_backend/models/order.py b/local_backend/models/order.py index d214f57..e625b07 100644 --- a/local_backend/models/order.py +++ b/local_backend/models/order.py @@ -12,7 +12,7 @@ class Order(Base): __tablename__ = "orders" id = Column(Integer, primary_key=True, index=True) - table_id = Column(Integer, ForeignKey("tables.id"), nullable=False) + table_id = Column(Integer, ForeignKey("tables.id"), nullable=True) # nullable for online orders opened_by = Column(Integer, ForeignKey("users.id"), nullable=False) opened_at = Column(DateTime(timezone=True), default=_utcnow) status = Column(String, default="open", nullable=False) # open|partially_paid|paid|closed|cancelled @@ -20,6 +20,15 @@ class Order(Base): closed_by = Column(Integer, ForeignKey("users.id"), nullable=True) notes = Column(Text, nullable=True) business_day_id = Column(Integer, ForeignKey("business_days.id"), nullable=True) + # Xenia Connect — online order fields + source = Column(String, default="pos", nullable=False) # "pos" | "online" + online_order_ref = Column(String, nullable=True) # e.g. "ORD-0042" + online_status = Column(String, nullable=True) # mirrors cloud status + online_customer_name = Column(String, nullable=True) + online_customer_phone = Column(String, nullable=True) + online_customer_address = Column(Text, nullable=True) + online_customer_notes = Column(Text, nullable=True) + online_order_type = Column(String, nullable=True) # "delivery" | "dine_in" table = relationship("Table", back_populates="orders") opener = relationship("User", foreign_keys=[opened_by], back_populates="orders_opened") diff --git a/local_backend/routers/connect_orders.py b/local_backend/routers/connect_orders.py new file mode 100644 index 0000000..e98f951 --- /dev/null +++ b/local_backend/routers/connect_orders.py @@ -0,0 +1,188 @@ +import logging +from typing import Optional +from fastapi import APIRouter, Depends, HTTPException, status, BackgroundTasks +from pydantic import BaseModel +from sqlalchemy.orm import Session + +import httpx + +from config import settings +from database import get_db +from models.order import Order +from schemas.order import OrderOut +from routers.deps import get_current_user, require_manager +from models.user import User +from services.sse_bus import broadcast_sync + +logger = logging.getLogger(__name__) +router = APIRouter() + + +class RejectRequest(BaseModel): + reason: Optional[str] = None + + +class StatusUpdateRequest(BaseModel): + status: str # "preparing" | "ready" | "out_for_delivery" | "delivered" + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +def _get_online_order(order_id: int, db: Session) -> Order: + order = db.query(Order).filter(Order.id == order_id, Order.source == "online").first() + if not order: + raise HTTPException(status_code=404, detail="Online order not found") + return order + + +def _cloud_headers() -> dict: + return {"X-Site-ID": settings.SITE_ID, "X-Site-Key": settings.SITE_KEY} + + +async def _mirror_status_to_cloud(cloud_order_ref: str, new_status: str, rejection_reason: str = None): + """Best-effort push of status update to cloud. Failures are logged, not raised.""" + if not settings.CLOUD_URL or not cloud_order_ref: + return + try: + # We need the cloud order id — look it up by public_ref via the status endpoint + # (The cloud orders router exposes PATCH /{order_id}/status by numeric id. + # We store cloud id implicitly through public_ref; fetch it first.) + async with httpx.AsyncClient(timeout=10) as client: + # Get cloud order id via status lookup (public endpoint, no auth needed) + status_resp = await client.get( + f"{settings.CLOUD_URL}/api/orders/status/{cloud_order_ref}" + ) + if status_resp.status_code != 200: + logger.warning("Could not find cloud order %s", cloud_order_ref) + return + + # Search pending list to get numeric id — use the site-key protected route + from middleware.license_check import license_state + site_numeric_id = license_state.get("site_numeric_id") + if not site_numeric_id: + return + + pending_resp = await client.get( + f"{settings.CLOUD_URL}/api/orders/pending/{site_numeric_id}", + headers=_cloud_headers(), + ) + # The order may already be synced; look in all orders via remote dashboard + # Simpler: use the manager-JWT-free internal PATCH directly via site key + # We need the cloud order numeric id. Store it on first pull — for now, + # the cloud also exposes PATCH via remote dashboard (manager JWT). + # Best path: add a site-key variant. Until then, we mirror via the + # existing PATCH endpoint using site key — works because orders.py + # accepts site key auth on PATCH /{order_id}/status. + # We get the numeric id from the pending list or a dedicated lookup. + # Iterate pending to find our ref (may be empty if already synced). + cloud_order_id = None + if pending_resp.status_code == 200: + for o in pending_resp.json(): + if o.get("public_ref") == cloud_order_ref: + cloud_order_id = o["id"] + break + + if cloud_order_id is None: + # Already synced — not in pending list. We can't easily look it up + # without a dedicated search endpoint. Log and skip for now; the + # remote manager dashboard's own PATCH handles the remote case. + logger.debug("Cloud order %s not in pending list; skipping mirror", cloud_order_ref) + return + + body = {"status": new_status} + if rejection_reason: + body["rejection_reason"] = rejection_reason + + patch_resp = await client.patch( + f"{settings.CLOUD_URL}/api/orders/{cloud_order_id}/status", + headers=_cloud_headers(), + json=body, + ) + patch_resp.raise_for_status() + logger.info("Mirrored status %s for cloud order %s", new_status, cloud_order_ref) + + except Exception as e: + logger.warning("Failed to mirror status to cloud for %s: %s", cloud_order_ref, e) + + +# ── Endpoints ───────────────────────────────────────────────────────────────── + +@router.get("/orders/incoming", response_model=list[OrderOut]) +def get_incoming_orders( + db: Session = Depends(get_db), + user: User = Depends(get_current_user), +): + """Return all online orders pending acceptance, newest first.""" + orders = ( + db.query(Order) + .filter(Order.source == "online", Order.online_status == "pending_acceptance") + .order_by(Order.opened_at.desc()) + .all() + ) + return orders + + +@router.post("/orders/{order_id}/accept", response_model=OrderOut) +async def accept_order( + order_id: int, + background: BackgroundTasks, + db: Session = Depends(get_db), + user: User = Depends(require_manager), +): + order = _get_online_order(order_id, db) + if order.online_status != "pending_acceptance": + raise HTTPException(status_code=400, detail=f"Order is already '{order.online_status}'") + + order.online_status = "accepted" + db.commit() + db.refresh(order) + + background.add_task(_mirror_status_to_cloud, order.online_order_ref, "accepted") + broadcast_sync("online_order_updated", {"order_id": order_id, "status": "accepted"}) + return order + + +@router.post("/orders/{order_id}/reject", response_model=OrderOut) +async def reject_order( + order_id: int, + body: RejectRequest, + background: BackgroundTasks, + db: Session = Depends(get_db), + user: User = Depends(require_manager), +): + order = _get_online_order(order_id, db) + if order.online_status != "pending_acceptance": + raise HTTPException(status_code=400, detail=f"Order is already '{order.online_status}'") + + order.online_status = "rejected" + order.status = "cancelled" + db.commit() + db.refresh(order) + + background.add_task(_mirror_status_to_cloud, order.online_order_ref, "rejected", body.reason) + broadcast_sync("online_order_updated", {"order_id": order_id, "status": "rejected"}) + return order + + +@router.post("/orders/{order_id}/status", response_model=OrderOut) +async def update_order_status( + order_id: int, + body: StatusUpdateRequest, + background: BackgroundTasks, + db: Session = Depends(get_db), + user: User = Depends(require_manager), +): + allowed = {"preparing", "ready", "out_for_delivery", "delivered"} + if body.status not in allowed: + raise HTTPException(status_code=400, detail=f"Invalid status '{body.status}'") + + order = _get_online_order(order_id, db) + order.online_status = body.status + if body.status == "delivered": + order.status = "closed" + db.commit() + db.refresh(order) + + background.add_task(_mirror_status_to_cloud, order.online_order_ref, body.status) + broadcast_sync("online_order_updated", {"order_id": order_id, "status": body.status}) + return order diff --git a/local_backend/schemas/order.py b/local_backend/schemas/order.py index 3fef693..3ad246e 100644 --- a/local_backend/schemas/order.py +++ b/local_backend/schemas/order.py @@ -111,7 +111,7 @@ class AuditLogOut(BaseModel): class OrderOut(BaseModel): id: int - table_id: int + table_id: Optional[int] = None opened_by: int opened_at: UTCDatetime status: str @@ -119,6 +119,15 @@ class OrderOut(BaseModel): closed_by: Optional[int] = None notes: Optional[str] = None business_day_id: Optional[int] = None + # Xenia Connect — online order fields + source: str = "pos" + online_order_ref: Optional[str] = None + online_status: Optional[str] = None + online_customer_name: Optional[str] = None + online_customer_phone: Optional[str] = None + online_customer_address: Optional[str] = None + online_customer_notes: Optional[str] = None + online_order_type: Optional[str] = None items: List[OrderItemOut] = [] waiters: List[OrderWaiterOut] = [] audit_logs: List[AuditLogOut] = [] diff --git a/local_backend/services/cloud_sync.py b/local_backend/services/cloud_sync.py index 8503d8c..7ccd427 100644 --- a/local_backend/services/cloud_sync.py +++ b/local_backend/services/cloud_sync.py @@ -24,6 +24,8 @@ import httpx from config import settings from middleware.license_check import license_state +ORDER_POLL_INTERVAL = settings.CONNECT_SYNC_INTERVAL_SECONDS + logger = logging.getLogger(__name__) SYNC_INTERVAL_SECONDS = 5 * 60 # 5 minutes @@ -145,6 +147,7 @@ async def _sync_once(): "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, @@ -172,6 +175,174 @@ async def _sync_once(): 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_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: @@ -179,6 +350,20 @@ async def _sync_loop(): await asyncio.sleep(SYNC_INTERVAL_SECONDS) +async def _connect_loop(): + """Faster loop: pulls pending online orders every CONNECT_SYNC_INTERVAL_SECONDS. + Also pushes the menu snapshot every 5 minutes (piggybacking on this loop).""" + menu_push_every = max(1, SYNC_INTERVAL_SECONDS and (5 * 60) // ORDER_POLL_INTERVAL) + tick = 0 + while True: + await asyncio.sleep(ORDER_POLL_INTERVAL) + await _pull_pending_orders() + tick += 1 + if tick % menu_push_every == 0: + await _push_menu_snapshot() + + async def start_cloud_sync() -> asyncio.Task: - task = asyncio.create_task(_sync_loop()) - return task + heartbeat_task = asyncio.create_task(_sync_loop()) + asyncio.create_task(_connect_loop()) + return heartbeat_task