API.py

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import logging
import math
import os
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path
from threading import Lock
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlparse
from uuid import uuid4

import cv2
import numpy as np
import requests
from fastapi import FastAPI, Body, File, Form, Header, HTTPException, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse
from ultralytics import YOLO


try:
	from dotenv import load_dotenv
except Exception:  # pragma: no cover
	load_dotenv = None


BASE_DIR = Path(__file__).resolve().parent
MODELS_DIR = BASE_DIR / "Models"

MODEL_PATHS = {
    "plate": MODELS_DIR / "car_plate_best.pt",
    "ocr": MODELS_DIR / "model_ocr.pt",
}





LOGGER = logging.getLogger("ain_el_aql_api")
if not LOGGER.handlers:
	logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")


def _bool_env(var_name: str, default: bool = False) -> bool:
	raw = os.getenv(var_name)
	if raw is None:
		return default
	return raw.strip().lower() in {"1", "true", "yes", "on"}


def _float_env(var_name: str, default: float) -> float:
	raw = os.getenv(var_name)
	if raw is None:
		return default

	value = raw.strip()
	if not value:
		return default

	try:
		return float(value)
	except ValueError:
		LOGGER.warning("Invalid float env for %s=%r. Falling back to %s", var_name, raw, default)
		return default


def _int_env(var_name: str, default: int) -> int:
	raw = os.getenv(var_name)
	if raw is None:
		return default

	value = raw.strip()
	if not value:
		return default

	try:
		return int(value)
	except ValueError:
		LOGGER.warning("Invalid int env for %s=%r. Falling back to %s", var_name, raw, default)
		return default


def _parse_csv_env_values(raw: str) -> List[str]:
	if not raw or not raw.strip():
		return []

	values: List[str] = []
	for part in raw.split(","):
		normalized = part.strip().lower()
		if normalized:
			values.append(normalized)

	return values


def _normalize_location_key_for_config(value: Optional[str]) -> Optional[str]:
	if value is None:
		return None
	raw = str(value).strip().upper()
	if not raw:
		return None
	key = "".join(raw.split())
	return key or None


def _parse_billing_mode(raw: Optional[str], *, default: str = "hourly", env_name: str = "") -> str:
	mode = str(raw or "").strip().lower()
	if mode in {"hourly", "daily"}:
		return mode
	if mode:
		label = env_name or "billing_mode"
		LOGGER.warning("Invalid %s=%r. Falling back to %s", label, raw, default)
	return default


def _parse_location_hourly_rates(raw: str) -> Dict[str, float]:
	result: Dict[str, float] = {}
	if not raw or not raw.strip():
		return result

	for item in raw.split(","):
		part = item.strip()
		if not part:
			continue

		location_part, separator, rate_part = part.partition(":")
		if not separator:
			LOGGER.warning(
				"Ignoring invalid PARKING_LOCATION_HOURLY_RATES item '%s'. Expected format LOCATION:RATE",
				part,
			)
			continue

		location_key = _normalize_location_key_for_config(location_part)
		if not location_key:
			LOGGER.warning("Ignoring empty parking location key in PARKING_LOCATION_HOURLY_RATES item '%s'", part)
			continue

		try:
			rate_value = float(rate_part.strip())
		except ValueError:
			LOGGER.warning("Ignoring invalid rate in PARKING_LOCATION_HOURLY_RATES item '%s'", part)
			continue

		if rate_value < 0:
			LOGGER.warning("Ignoring negative rate in PARKING_LOCATION_HOURLY_RATES item '%s'", part)
			continue

		result[location_key] = rate_value

	return result


def _parse_location_int_values(raw: str, *, env_name: str) -> Dict[str, int]:
	result: Dict[str, int] = {}
	if not raw or not raw.strip():
		return result

	for item in raw.split(","):
		part = item.strip()
		if not part:
			continue

		location_part, separator, value_part = part.partition(":")
		if not separator:
			LOGGER.warning("Ignoring invalid %s item '%s'. Expected format LOCATION:VALUE", env_name, part)
			continue

		location_key = _normalize_location_key_for_config(location_part)
		if not location_key:
			LOGGER.warning("Ignoring empty location key in %s item '%s'", env_name, part)
			continue

		try:
			parsed_value = int(value_part.strip())
		except ValueError:
			LOGGER.warning("Ignoring invalid int value in %s item '%s'", env_name, part)
			continue

		if parsed_value < 0:
			LOGGER.warning("Ignoring negative value in %s item '%s'", env_name, part)
			continue

		result[location_key] = parsed_value

	return result


def _parse_location_pricing_configs(raw: str) -> Dict[str, Dict[str, Any]]:
	result: Dict[str, Dict[str, Any]] = {}
	if not raw or not raw.strip():
		return result

	for item in raw.split(","):
		part = item.strip()
		if not part:
			continue

		pieces = [piece.strip() for piece in part.split(":")]
		if len(pieces) != 5:
			LOGGER.warning(
				"Ignoring invalid PARKING_LOCATION_PRICING item '%s'. Expected LOCATION:MODE:BASE:EXTRA_AFTER_FIRST_HOUR:CAPACITY",
				part,
			)
			continue

		location_key = _normalize_location_key_for_config(pieces[0])
		if not location_key:
			LOGGER.warning("Ignoring empty location key in PARKING_LOCATION_PRICING item '%s'", part)
			continue

		billing_mode = _parse_billing_mode(pieces[1], default="hourly", env_name="PARKING_LOCATION_PRICING.mode")

		try:
			base_rate_egp = float(pieces[2])
			extra_after_first_hour_egp = float(pieces[3])
			capacity = int(pieces[4])
		except ValueError:
			LOGGER.warning("Ignoring invalid numeric values in PARKING_LOCATION_PRICING item '%s'", part)
			continue

		if base_rate_egp < 0 or extra_after_first_hour_egp < 0 or capacity < 0:
			LOGGER.warning("Ignoring negative values in PARKING_LOCATION_PRICING item '%s'", part)
			continue

		result[location_key] = {
			"billing_mode": billing_mode,
			"base_rate_egp": base_rate_egp,
			"extra_after_first_hour_egp": extra_after_first_hour_egp,
			"capacity": capacity,
		}

	return result


SUPABASE_URL = os.getenv("SUPABASE_URL", "").strip().rstrip("/")
SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "").strip()
SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "").strip()
SUPABASE_ENFORCE_AUTH = _bool_env("SUPABASE_ENFORCE_AUTH", default=False)
SUPABASE_RAW_BUCKET = os.getenv("SUPABASE_RAW_BUCKET", "car-raw-images").strip()
SUPABASE_PROCESSED_BUCKET = os.getenv("SUPABASE_PROCESSED_BUCKET", "car-processed-images").strip()
SUPABASE_TIMEOUT_SECONDS = float(os.getenv("SUPABASE_TIMEOUT_SECONDS", "20"))
BARRIER_API_TOKEN = os.getenv("BARRIER_API_TOKEN", "").strip()
_raw_barrier_tokens = [BARRIER_API_TOKEN]
_raw_barrier_tokens.extend(part.strip() for part in os.getenv("BARRIER_API_TOKENS", "").split(","))
BARRIER_API_TOKENS = {token for token in _raw_barrier_tokens if token}
del _raw_barrier_tokens
BARRIER_SYNTHETIC_USER_ID = os.getenv("BARRIER_SYNTHETIC_USER_ID", "barrier-device").strip() or "barrier-device"
PASSWORD_RESET_REDIRECT_URL = os.getenv("PASSWORD_RESET_REDIRECT_URL", "").strip()
INTERNAL_LOGIN_EMAIL_SUFFIX = "@ainel-aql.local"
REDIRECT_ALLOWED_HTTP_HOSTS = set(_parse_csv_env_values(os.getenv("ALLOWED_REDIRECT_HOSTS", "")))
_configured_password_reset_redirect_host = (urlparse(PASSWORD_RESET_REDIRECT_URL).hostname or "").strip().lower()
if _configured_password_reset_redirect_host:
	REDIRECT_ALLOWED_HTTP_HOSTS.add(_configured_password_reset_redirect_host)
del _configured_password_reset_redirect_host
REDIRECT_ALLOWED_DEEPLINK_SCHEMES = set(
	_parse_csv_env_values(os.getenv("ALLOWED_REDIRECT_DEEPLINK_SCHEMES", "ainelaql"))
)
if not REDIRECT_ALLOWED_DEEPLINK_SCHEMES:
	REDIRECT_ALLOWED_DEEPLINK_SCHEMES = {"ainelaql"}
PARKING_HOURLY_RATE_EGP = _float_env("PARKING_HOURLY_RATE_EGP", 30.0)
PARKING_LOCATION_HOURLY_RATES = _parse_location_hourly_rates(os.getenv("PARKING_LOCATION_HOURLY_RATES", ""))
PARKING_BILLING_MODE = _parse_billing_mode(os.getenv("PARKING_BILLING_MODE", "hourly"), default="hourly", env_name="PARKING_BILLING_MODE")
PARKING_DAILY_RATE_EGP = _float_env("PARKING_DAILY_RATE_EGP", 200.0)
PARKING_EXTRA_AFTER_FIRST_HOUR_EGP = _float_env("PARKING_EXTRA_AFTER_FIRST_HOUR_EGP", PARKING_HOURLY_RATE_EGP)
GARAGE_TOTAL_CAPACITY = max(0, _int_env("GARAGE_TOTAL_CAPACITY", 1000))
PARKING_LOCATION_CAPACITIES = _parse_location_int_values(
	os.getenv("PARKING_LOCATION_CAPACITIES", ""),
	env_name="PARKING_LOCATION_CAPACITIES",
)
PARKING_LOCATION_PRICING = _parse_location_pricing_configs(os.getenv("PARKING_LOCATION_PRICING", ""))
APP_SERVICE_FEE_EGP = _float_env("APP_SERVICE_FEE_EGP", 0.0)
CORS_ALLOW_ORIGINS = [origin.strip() for origin in os.getenv("CORS_ALLOW_ORIGINS", "*").split(",") if origin.strip()]

PAYMOB_BASE_URL = os.getenv("PAYMOB_BASE_URL", "https://accept.paymob.com/api").strip().rstrip("/")
PAYMOB_API_KEY = os.getenv("PAYMOB_API_KEY", "").strip()
PAYMOB_INTEGRATION_ID = os.getenv("PAYMOB_INTEGRATION_ID", "").strip()
PAYMOB_IFRAME_ID = os.getenv("PAYMOB_IFRAME_ID", "").strip()
PAYMOB_CURRENCY = os.getenv("PAYMOB_CURRENCY", "EGP").strip().upper() or "EGP"

MODEL_INFERENCE_PROVIDER_RAW = os.getenv("MODEL_INFERENCE_PROVIDER", "local").strip().lower()
if MODEL_INFERENCE_PROVIDER_RAW in {"local", "remote"}:
	MODEL_INFERENCE_PROVIDER = MODEL_INFERENCE_PROVIDER_RAW
else:
	LOGGER.warning(
		"Invalid MODEL_INFERENCE_PROVIDER=%r. Falling back to 'local'.",
		MODEL_INFERENCE_PROVIDER_RAW,
	)
	MODEL_INFERENCE_PROVIDER = "local"

MODEL_SERVICE_URL = os.getenv("MODEL_SERVICE_URL", "https://pant0x-ealpr-ocr.hf.space").strip().rstrip("/")
MODEL_SERVICE_INFER_PATH = os.getenv("MODEL_SERVICE_INFER_PATH", "/model/infer").strip() or "/model/infer"
MODEL_SERVICE_TIMEOUT_SECONDS = _float_env("MODEL_SERVICE_TIMEOUT_SECONDS", 45.0)
MODEL_SERVICE_SHARED_SECRET = os.getenv("MODEL_SERVICE_SHARED_SECRET", "").strip()
MODEL_SERVICE_HEADER_NAME = os.getenv("MODEL_SERVICE_HEADER_NAME", "X-Model-Service-Secret").strip() or "X-Model-Service-Secret"
MODEL_SERVICE_ENABLE_LOCAL_ENDPOINT = _bool_env(
	"MODEL_SERVICE_ENABLE_LOCAL_ENDPOINT",
	default=MODEL_INFERENCE_PROVIDER == "local",
)

STAFF_ROLES = {"admin", "security"}
BARRIER_ROLES = {"barrier"}
DEVELOPER_ROLES = {"developer"}
CORE_TABLES = {"profiles", "vehicles", "parking_sessions", "car_events"}
PARKING_LOCATION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 _-]{0,31}(?:\s*,\s*[A-Za-z0-9][A-Za-z0-9 _-]{0,31})*$")
CAMERA_SOURCE_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9_.-]{3,32}$")
PHONE_PATTERN = re.compile(r"^\+?[0-9]{7,15}$")
NATIONAL_ID_PATTERN = re.compile(r"^[A-Za-z0-9-]{6,32}$")
PROFILE_SELECT_FIELDS = "id,first_name,last_name,full_name,username,phone_number,email,national_id,role,staff_id"
VEHICLE_SELECT_FIELDS = "id,owner_id,plate_letters_ar,plate_numbers_ar,plate_letters_en,plate_numbers_en,car_name,car_model,car_color,is_active,is_verified,verified_at,verified_by,created_at,updated_at"
SUPPORTED_EVENT_TYPES = {"entry", "exit", "ocr_scan"}
SUPPORTED_OTP_CHANNELS = {"email", "phone"}
MOBILE_REDIRECT_SCHEME_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*$")


def _supabase_configured() -> bool:
	return bool(SUPABASE_URL and SUPABASE_ANON_KEY and SUPABASE_SERVICE_ROLE_KEY)


def _paymob_configured() -> bool:
	return bool(PAYMOB_BASE_URL and PAYMOB_API_KEY and PAYMOB_INTEGRATION_ID and PAYMOB_IFRAME_ID)


def _model_service_configured() -> bool:
	return bool(MODEL_SERVICE_URL)


def _resolve_model_service_infer_url() -> Optional[str]:
	if not _model_service_configured():
		return None

	path = MODEL_SERVICE_INFER_PATH
	if not path.startswith("/"):
		path = f"/{path}"
	return f"{MODEL_SERVICE_URL}{path}"


def _should_load_local_models() -> bool:
	return MODEL_INFERENCE_PROVIDER == "local" or MODEL_SERVICE_ENABLE_LOCAL_ENDPOINT


_PLATE_MODEL: Optional[YOLO] = None
_OCR_MODEL: Optional[YOLO] = None


def _ensure_models_loaded() -> None:
	global _PLATE_MODEL, _OCR_MODEL

	with _model_lock:
		if _PLATE_MODEL is not None and _OCR_MODEL is not None:
			return

		try:
			if _PLATE_MODEL is None:
				_PLATE_MODEL = YOLO(str(MODEL_PATHS["plate"]))
			if _OCR_MODEL is None:
				_OCR_MODEL = YOLO(str(MODEL_PATHS["ocr"]))
		except Exception as exc:
			LOGGER.error("Failed to load local YOLO models: %s", exc)
			raise RuntimeError(f"Model initialization failed: {exc}") from exc


def _normalize_spaces(value: str) -> str:
	return " ".join((value or "").split())


def _normalize_name_part(value: Any) -> str:
	return _normalize_spaces(str(value or ""))


def _compose_full_name(first_name: Any, last_name: Any) -> str:
	return _normalize_spaces(f"{_normalize_name_part(first_name)} {_normalize_name_part(last_name)}")


def _split_full_name_parts(full_name: Any) -> Tuple[str, str]:
	normalized_full_name = _normalize_name_part(full_name)
	if not normalized_full_name:
		return "", ""

	parts = normalized_full_name.split(" ", 1)
	if len(parts) == 1:
		return parts[0], ""

	return parts[0], _normalize_spaces(parts[1])


def _resolve_profile_name_fields(
	payload: Optional[Dict[str, Any]],
	*,
	fallback_full_name: Optional[Any] = None,
	fallback_first_name: Optional[Any] = None,
	fallback_last_name: Optional[Any] = None,
) -> Tuple[str, str, str]:
	body = payload if isinstance(payload, dict) else {}

	first_name = _normalize_name_part(body.get("first_name") or body.get("firstName") or fallback_first_name)
	last_name = _normalize_name_part(body.get("last_name") or body.get("lastName") or fallback_last_name)
	legacy_full_name = _normalize_name_part(body.get("full_name") or body.get("name") or fallback_full_name)

	if not first_name and not last_name and legacy_full_name:
		first_name, last_name = _split_full_name_parts(legacy_full_name)

	full_name = _compose_full_name(first_name, last_name)
	if not full_name and legacy_full_name:
		full_name = legacy_full_name
		if not first_name and not last_name:
			first_name, last_name = _split_full_name_parts(legacy_full_name)

	return first_name, last_name, full_name


def _is_valid_http_url(url: str) -> bool:
	if not url:
		return False
	parsed = urlparse(url)
	return parsed.scheme in {"http", "https"} and bool(parsed.netloc)


def _is_valid_redirect_url(url: str, *, allowed_http_hosts: Optional[List[str]] = None) -> bool:
	if not url:
		return False

	parsed = urlparse(url)
	scheme = (parsed.scheme or "").strip()
	if not scheme or not MOBILE_REDIRECT_SCHEME_PATTERN.fullmatch(scheme):
		return False
	normalized_scheme = scheme.lower()

	if normalized_scheme in {"http", "https"}:
		hostname = (parsed.hostname or "").strip().lower()
		if not hostname:
			return False

		effective_allowed_hosts = set(REDIRECT_ALLOWED_HTTP_HOSTS)
		if allowed_http_hosts:
			for host in allowed_http_hosts:
				normalized_host = (host or "").strip().lower()
				if normalized_host:
					effective_allowed_hosts.add(normalized_host)

		if not effective_allowed_hosts:
			return False

		return hostname in effective_allowed_hosts

	if normalized_scheme not in REDIRECT_ALLOWED_DEEPLINK_SCHEMES:
		return False

	# Mobile deep links such as ainelaql://auth/reset are allowed.
	return bool(parsed.netloc or parsed.path)


def _resolve_public_base_url(request: Request) -> str:
	forwarded_proto = _normalize_spaces((request.headers.get("x-forwarded-proto") or "").split(",")[0])
	forwarded_host = _normalize_spaces((request.headers.get("x-forwarded-host") or "").split(",")[0])
	if forwarded_proto and forwarded_host:
		return f"{forwarded_proto}://{forwarded_host}".rstrip("/")
	return str(request.base_url).rstrip("/")


def _resolve_default_password_reset_redirect_url(request: Request) -> str:
	configured = _normalize_spaces(PASSWORD_RESET_REDIRECT_URL)
	if configured:
		parsed = urlparse(configured)
		if parsed.scheme and parsed.scheme not in {"http", "https"}:
			# Supabase can always deliver web callbacks; the bridge then opens the mobile deep link.
			return f"{_resolve_public_base_url(request)}/auth/reset-password-bridge"
		return configured

	base_url = _resolve_public_base_url(request)
	return f"{base_url}/auth/reset-password-page"


def _resolve_password_reset_deep_link_url() -> str:
	configured = _normalize_spaces(PASSWORD_RESET_REDIRECT_URL)
	if _is_valid_redirect_url(configured):
		parsed = urlparse(configured)
		if parsed.scheme not in {"http", "https"}:
			return configured

	return "ainelaql://auth/reset"


def _split_arabic_plate(plate_text_ar: str) -> Tuple[Optional[str], Optional[str]]:
	raw = _normalize_spaces(plate_text_ar)
	if not raw or raw.upper() == "N/A" or "|" not in raw:
		return None, None

	letters, numbers = [part.strip() for part in raw.split("|", 1)]
	letters = _normalize_spaces(letters)
	numbers = _normalize_spaces(numbers)
	if not letters or not numbers:
		return None, None
	return letters, numbers


def _split_english_plate(plate_text_en: str) -> Tuple[Optional[str], Optional[str]]:
	raw = _normalize_spaces(plate_text_en)
	if not raw or raw.upper() == "N/A" or "|" not in raw:
		return None, None

	letters, numbers = [part.strip() for part in raw.split("|", 1)]
	letters = _normalize_spaces(letters)
	numbers = numbers.replace(" ", "")
	if not letters and not numbers:
		return None, None
	return letters or None, numbers or None


def _normalize_plate_text(text: Optional[str]) -> str:
	"""Normalize plate text for matching: remove spaces, convert Arabic-Indic -> Western digits."""
	if not text:
		return ''
	# Convert to string and handle case
	result = str(text).strip().lower()
	
	# Strip common OCR debug labels
	for prefix in ["car", "vehicle", "license", "plate", "lp", "ar", "en"]:
		if result.startswith(prefix):
			result = result[len(prefix):].strip()
			
	# Remove all whitespace and non-alphanumeric separators
	result = "".join(c for c in result if c.isalnum() or c in "|-")
	
	# Convert Arabic-Indic digits to Western
	_ARABIC_INDIC = '\u0660\u0661\u0662\u0663\u0664\u0665\u0666\u0667\u0668\u0669'
	_WESTERN      = '0123456789'
	for ar, en in zip(_ARABIC_INDIC, _WESTERN):
		result = result.replace(ar, en)
	return result.strip().upper()


def _format_vehicle_plate_label(vehicle_row: Optional[Dict[str, Any]]) -> str:
	if not isinstance(vehicle_row, dict):
		return ""

	letters_ar = _normalize_spaces(str(vehicle_row.get("plate_letters_ar") or ""))
	numbers_ar = _normalize_spaces(str(vehicle_row.get("plate_numbers_ar") or ""))
	if letters_ar or numbers_ar:
		return _normalize_spaces(f"{letters_ar} {numbers_ar}")

	letters_en = _normalize_spaces(str(vehicle_row.get("plate_letters_en") or ""))
	numbers_en = _normalize_spaces(str(vehicle_row.get("plate_numbers_en") or ""))
	if letters_en or numbers_en:
		return _normalize_spaces(f"{letters_en} {numbers_en}")

	return ""


def _resolve_plate_label_for_session(session_row: Optional[Dict[str, Any]]) -> str:
	if not isinstance(session_row, dict):
		return ""

	plate_text = _normalize_spaces(str(session_row.get("plate_arabic") or ""))
	if plate_text:
		return plate_text

	plate_text = _normalize_spaces(str(session_row.get("plate_english") or ""))
	if plate_text:
		return plate_text

	vehicle_id = str(session_row.get("vehicle_id") or "").strip()
	if not vehicle_id:
		return ""

	vehicle_map = _fetch_vehicle_map_by_ids([vehicle_id])
	return _format_vehicle_plate_label(vehicle_map.get(vehicle_id))


def _error_response(status_code: int, message: str, code: str) -> JSONResponse:
	return JSONResponse(
		status_code=status_code,
		content={
			"status": "error",
			"message": message,
			"code": code,
		},
	)


def _plate_letters_variants(text: Optional[str]) -> List[str]:
	"""Return deduplicated letter variants to tolerate spacing differences in DB/OCR."""
	raw = _normalize_spaces(str(text or ""))
	if not raw:
		return []

	variants: List[str] = [raw]
	no_space = "".join(raw.split())
	if no_space and no_space not in variants:
		variants.append(no_space)

	return variants


def _plate_numbers_variants(text: Optional[str]) -> List[str]:
	"""Return deduplicated number variants (Arabic-Indic and Western)."""
	raw = _normalize_spaces(str(text or ""))
	if not raw:
		return []

	normalized = _normalize_plate_text(raw)
	variants: List[str] = [raw]
	if normalized and normalized not in variants:
		variants.append(normalized)

	return variants


def _normalize_event_type(value: Optional[str]) -> str:
	raw = (value or "ocr_scan").strip().lower()
	if raw not in SUPPORTED_EVENT_TYPES:
		raise HTTPException(
			status_code=400,
			detail=f"Invalid event_type '{raw}'. Allowed values: {sorted(SUPPORTED_EVENT_TYPES)}",
		)
	return raw


def _build_plate_payload(plate_info: Dict[str, Any]) -> Dict[str, Any]:
	arabic_text = str(plate_info.get("arabic") or "N/A")
	english_text = str(plate_info.get("english") or "N/A")

	ar_letters, ar_numbers = _split_arabic_plate(arabic_text)
	en_letters, en_numbers = _split_english_plate(english_text)

	return {
		"arabic_text": arabic_text,
		"english_text": english_text,
		"arabic": {
			"letters": ar_letters,
			"numbers": ar_numbers,
		},
		"english": {
			"letters": en_letters,
			"numbers": en_numbers,
		},
	}


def _normalize_parking_location(value: Optional[str]) -> Optional[str]:
	raw = _normalize_spaces(value or "")
	if not raw:
		return None

	if not PARKING_LOCATION_PATTERN.fullmatch(raw):
		raise HTTPException(
			status_code=400,
			detail="Invalid parking_location format. Use values like OPERA or A1, and comma-separated forms like OPERA,CINEMA.",
		)

	parts = [
		location_key
		for location_key in [_normalize_location_key_for_config(part) for part in raw.split(",")]
		if location_key
	]
	return ",".join(parts) if parts else None


def _normalize_camera_source(value: Optional[str]) -> Optional[str]:
	raw = _normalize_spaces(value or "")
	if not raw:
		return None

	if not CAMERA_SOURCE_PATTERN.fullmatch(raw):
		raise HTTPException(
			status_code=400,
			detail="Invalid camera_source. Use alphanumeric, underscore, or dash (max 64 chars).",
		)

	return raw


def _is_plate_detected(plate_info: Dict[str, Any]) -> bool:
	if not isinstance(plate_info, dict):
		return False

	arabic = str(plate_info.get("arabic") or "").strip().upper()
	english = str(plate_info.get("english") or "").strip().upper()
	
	if arabic not in {"", "N/A"} or english not in {"", "N/A"}:
		return True
		
	# Even if chars are detected, if there's no readable string, we can't track it.
	return False


def _is_staff_role(role: Optional[str]) -> bool:
	return (role or "").strip().lower() in STAFF_ROLES


def _is_barrier_role(role: Optional[str]) -> bool:
	return (role or "").strip().lower() in BARRIER_ROLES


def _can_view_global_records(role: Optional[str]) -> bool:
	return _is_staff_role(role) or _is_barrier_role(role)


def _is_developer_role(role: Optional[str]) -> bool:
	return (role or "").strip().lower() in DEVELOPER_ROLES


def _extract_created_by_user_id(request_user: Optional[Dict[str, Any]]) -> Optional[str]:
	if not isinstance(request_user, dict):
		return None
	if str(request_user.get("auth_type") or "").strip().lower() != "supabase":
		return None
	user_id = str(request_user.get("id") or "").strip()
	return user_id or None


def _extract_bearer_token(authorization: Optional[str]) -> Optional[str]:
	if not authorization:
		return None
	parts = authorization.strip().split(" ", 1)
	if len(parts) != 2 or parts[0].lower() != "bearer":
		return None
	token = parts[1].strip()
	return token if token else None


def _looks_like_jwt(token: str) -> bool:
	parts = token.split(".")
	if len(parts) != 3:
		return False
	return bool(parts[0] and parts[1])


def _extract_response_error_message(response: requests.Response) -> str:
	try:
		payload = response.json()
	except ValueError:
		payload = None

	if isinstance(payload, dict):
		for key in ["error_description", "message", "error", "msg"]:
			value = payload.get(key)
			if value is not None:
				text = str(value).strip()
				if text:
					return text

	return response.text[:250].strip()


def _supabase_headers(*, api_key: str, bearer: str, content_type: Optional[str] = None, prefer: Optional[str] = None) -> Dict[str, str]:
	headers: Dict[str, str] = {
		"apikey": api_key,
		"Authorization": f"Bearer {bearer}",
	}
	if content_type:
		headers["Content-Type"] = content_type
	if prefer:
		headers["Prefer"] = prefer
	return headers



def _verify_barrier_token(token: str) -> Optional[Dict[str, Any]]:
	if not token or not BARRIER_API_TOKENS:
		return None

	for configured_token in BARRIER_API_TOKENS:
		if hmac.compare_digest(token, configured_token):
			return {
				"id": BARRIER_SYNTHETIC_USER_ID,
				"email": None,
				"role": "barrier",
				"auth_type": "barrier_token",
			}

	return None


def _verify_supabase_user_token(token: str) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase auth is not configured on backend.")

	if not _looks_like_jwt(token):
		raise HTTPException(
			status_code=401,
			detail="Invalid Supabase access token format. Send session.access_token as Bearer token.",
		)

	url = f"{SUPABASE_URL}/auth/v1/user"
	try:
		response = requests.get(
			url,
			headers={
				"apikey": SUPABASE_ANON_KEY,
				"Authorization": f"Bearer {token}",
			},
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
	except requests.RequestException as exc:
		raise HTTPException(status_code=503, detail=f"Supabase auth is unreachable: {exc}") from exc

	if response.status_code != 200:
		supabase_error = _extract_response_error_message(response)
		err_lc = supabase_error.lower()
		if "expired" in err_lc:
			detail = "Supabase access token expired. Refresh session and retry."
		elif any(hint in err_lc for hint in ["jwt", "signature", "malformed", "invalid"]):
			detail = "Invalid Supabase access token. Send session.access_token as Bearer token."
		else:
			detail = "Invalid or expired Supabase token."
		LOGGER.info("Supabase token verify failed with status %s: %s", response.status_code, supabase_error)
		raise HTTPException(status_code=401, detail=detail)

	payload = response.json()
	if not isinstance(payload, dict):
		raise HTTPException(status_code=401, detail="Invalid Supabase auth response payload.")

	user_id = payload.get("id")
	if not user_id:
		raise HTTPException(status_code=401, detail="Supabase token did not return a user id.")

	user_context = {
		"id": user_id,
		"email": payload.get("email"),
		"auth_type": "supabase",
		"user_metadata": payload.get("user_metadata") if isinstance(payload.get("user_metadata"), dict) else {},
		"app_metadata": payload.get("app_metadata") if isinstance(payload.get("app_metadata"), dict) else {},
	}
	_ensure_profile_row_exists_for_verified_user(user_context)
	return user_context


def _require_authenticated_user(authorization: Optional[str]) -> Optional[Dict[str, Any]]:
	token = _extract_bearer_token(authorization)

	if SUPABASE_ENFORCE_AUTH and not token:
		raise HTTPException(status_code=401, detail="Missing Bearer token.")

	if not token:
		return None

	barrier_user = _verify_barrier_token(token)
	if barrier_user is not None:
		return barrier_user

	return _verify_supabase_user_token(token)


def _upload_to_supabase_storage(bucket: str, object_path: str, content: bytes, content_type: str = "image/jpeg") -> None:
	if not _supabase_configured():
		raise RuntimeError("Supabase is not configured.")

	endpoint = f"{SUPABASE_URL}/storage/v1/object/{bucket}/{object_path}"
	headers = _supabase_headers(
		api_key=SUPABASE_SERVICE_ROLE_KEY,
		bearer=SUPABASE_SERVICE_ROLE_KEY,
		content_type=content_type,
	)
	headers["x-upsert"] = "true"

	response = requests.post(
		endpoint,
		data=content,
		headers=headers,
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201}:
		raise RuntimeError(f"Storage upload failed ({response.status_code}): {response.text[:400]}")


def _delete_from_supabase_storage(bucket: str, object_path: str) -> None:
	if not _supabase_configured() or not bucket or not object_path:
		return

	endpoint = f"{SUPABASE_URL}/storage/v1/object/{bucket}/{object_path}"
	response = requests.delete(
		endpoint,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204, 404}:
		raise RuntimeError(f"Storage delete failed ({response.status_code}): {response.text[:400]}")


def _supabase_list_buckets() -> List[str]:
	if not _supabase_configured():
		return []

	response = requests.get(
		f"{SUPABASE_URL}/storage/v1/bucket",
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		raise RuntimeError(f"Bucket list request failed ({response.status_code}): {response.text[:250]}")

	rows = response.json()
	if not isinstance(rows, list):
		return []

	return [str(row.get("id")) for row in rows if isinstance(row, dict) and row.get("id")]


def _supabase_table_exists(table_name: str) -> Tuple[bool, Optional[str]]:
	if not _supabase_configured():
		return False, "Supabase not configured"

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/{table_name}",
		params={"select": "*", "limit": "1"},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)

	if response.status_code == 200:
		return True, None

	return False, f"HTTP {response.status_code}: {response.text[:220]}"


def _supabase_get_profile_role(user_id: Optional[str]) -> Optional[str]:
	normalized_user_id = str(user_id or "").strip()
	if normalized_user_id and normalized_user_id == BARRIER_SYNTHETIC_USER_ID:
		return "barrier"

	profile = _supabase_get_profile_by_user_id(user_id)
	if not isinstance(profile, dict):
		return None

	role = str(profile.get("role") or "").strip().lower()
	return role or None


def _resolve_requester_role(request_user: Optional[Dict[str, Any]]) -> Optional[str]:
	if not isinstance(request_user, dict):
		return None

	inline_role = str(request_user.get("role") or "").strip().lower()
	if inline_role:
		return inline_role

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		return None

	return _supabase_get_profile_role(requester_id)


def _require_developer_user(authorization: Optional[str]) -> Tuple[Dict[str, Any], str]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_role = _resolve_requester_role(request_user)
	if not _is_developer_role(requester_role):
		raise HTTPException(status_code=403, detail="Only developers can access this endpoint.")

	return request_user, (requester_role or "").strip().lower()


def _supabase_get_profile_by_user_id(user_id: Optional[str]) -> Optional[Dict[str, Any]]:
	if not user_id or not _supabase_configured():
		return None

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/profiles",
		params={
			"select": PROFILE_SELECT_FIELDS,
			"id": f"eq.{user_id}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch profile for %s: %s", user_id, response.text[:240])
		return None

	rows = response.json()
	if not isinstance(rows, list) or not rows:
		return None
	first_row = rows[0]
	if not isinstance(first_row, dict):
		return None
	return first_row


def _ensure_profile_row_exists_for_verified_user(auth_user: Dict[str, Any]) -> None:
	if not _supabase_configured() or not isinstance(auth_user, dict):
		return

	user_id = str(auth_user.get("id") or "").strip()
	if not user_id:
		return

	existing_profile = _supabase_get_profile_by_user_id(user_id)
	if isinstance(existing_profile, dict):
		return

	email = _normalize_spaces(str(auth_user.get("email") or "")).lower()
	payload: Dict[str, Any] = {
		"id": user_id,
	}
	if email:
		payload["email"] = email

	try:
		response = requests.post(
			f"{SUPABASE_URL}/rest/v1/profiles",
			json=payload,
			headers=_supabase_headers(
				api_key=SUPABASE_SERVICE_ROLE_KEY,
				bearer=SUPABASE_SERVICE_ROLE_KEY,
				content_type="application/json",
				prefer="resolution=merge-duplicates,return=minimal",
			),
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
		if response.status_code not in {200, 201, 204}:
			LOGGER.warning(
				"Failed to auto-create profile for authenticated user %s: %s",
				user_id,
				response.text[:240],
			)
	except requests.RequestException as exc:
		LOGGER.warning("Failed to auto-create profile for authenticated user %s: %s", user_id, exc)


def _supabase_get_profile_by_field(
	field: str,
	value: str,
	*,
	case_insensitive: bool = False,
) -> Optional[Dict[str, Any]]:
	if not _supabase_configured() or not value:
		return None

	operator = "ilike" if case_insensitive else "eq"
	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/profiles",
		params={
			"select": PROFILE_SELECT_FIELDS,
			field: f"{operator}.{value}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch profile by %s: %s", field, response.text[:240])
		return None

	rows = response.json()
	if not isinstance(rows, list) or not rows:
		return None
	first_row = rows[0]
	if not isinstance(first_row, dict):
		return None
	return first_row


def _supabase_find_profile_by_identifier(identifier: Optional[str]) -> Optional[Dict[str, Any]]:
	if not _supabase_configured():
		return None

	normalized_identifier = _normalize_spaces(str(identifier or ""))
	if not normalized_identifier:
		return None

	lookup_chain: List[Tuple[str, str, bool]] = []
	if "@" in normalized_identifier:
		lookup_chain.append(("email", normalized_identifier.lower(), True))

	lookup_chain.extend(
		[
			("username", normalized_identifier, False),
			("phone_number", normalized_identifier, False),
			("national_id", normalized_identifier, False),
		]
	)

	if "@" not in normalized_identifier:
		lookup_chain.append(("email", normalized_identifier.lower(), True))

	seen: set = set()
	for field_name, field_value, case_insensitive in lookup_chain:
		key = (field_name, field_value, case_insensitive)
		if key in seen:
			continue
		seen.add(key)

		profile = _supabase_get_profile_by_field(
			field_name,
			field_value,
			case_insensitive=case_insensitive,
		)
		if isinstance(profile, dict):
			return profile

	return None


def _supabase_password_login(
	*,
	password: str,
	email: Optional[str] = None,
	phone_number: Optional[str] = None,
) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
	if not _supabase_configured():
		return None, "Supabase is not configured."

	login_payload: Dict[str, str] = {"password": password}
	if email:
		login_payload["email"] = email.strip().lower()
	elif phone_number:
		login_payload["phone"] = phone_number.strip()
	else:
		return None, "No login identifier was provided."

	response = requests.post(
		f"{SUPABASE_URL}/auth/v1/token?grant_type=password",
		json=login_payload,
		headers={
			"apikey": SUPABASE_ANON_KEY,
			"Content-Type": "application/json",
		},
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)

	if response.status_code == 200:
		payload = response.json()
		if isinstance(payload, dict):
			return payload, None
		return None, "Supabase login returned an unexpected response payload."

	try:
		error_data = response.json()
	except ValueError:
		error_data = {}

	error_message = ""
	if isinstance(error_data, dict):
		error_message = str(error_data.get("error_description") or error_data.get("message") or "").strip()
	if not error_message:
		error_message = response.text[:200]

	return None, error_message or f"Login failed with status {response.status_code}."


def _is_supabase_duplicate_auth_user_error(error_message: Optional[str]) -> bool:
	normalized = _normalize_spaces(str(error_message or "")).lower()
	if not normalized:
		return False

	duplicate_hints = [
		"already registered",
		"already exists",
		"already in use",
		"user already registered",
		"email address already in use",
		"duplicate key value",
	]
	return any(hint in normalized for hint in duplicate_hints)


def _find_supabase_auth_user_by_email(email: str) -> Optional[Dict[str, Any]]:
	if not _supabase_configured():
		return None

	normalized_email = _normalize_spaces(email).lower()
	if not normalized_email or "@" not in normalized_email:
		return None

	max_pages = 20
	per_page = 200
	for page in range(1, max_pages + 1):
		try:
			response = requests.get(
				f"{SUPABASE_URL}/auth/v1/admin/users",
				params={
					"page": str(page),
					"per_page": str(per_page),
				},
				headers={
					"apikey": SUPABASE_SERVICE_ROLE_KEY,
					"Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}",
				},
				timeout=SUPABASE_TIMEOUT_SECONDS,
			)
		except requests.RequestException as exc:
			LOGGER.warning("Failed to list Supabase auth users while searching for %s: %s", normalized_email, exc)
			return None

		if response.status_code != 200:
			LOGGER.warning(
				"Supabase auth users listing failed (%s) while searching for %s: %s",
				response.status_code,
				normalized_email,
				_extract_response_error_message(response),
			)
			return None

		try:
			payload = response.json()
		except ValueError:
			LOGGER.warning("Supabase auth users listing returned invalid JSON while searching for %s", normalized_email)
			return None

		users_payload: Any
		if isinstance(payload, dict):
			users_payload = payload.get("users")
		else:
			users_payload = payload

		users = users_payload if isinstance(users_payload, list) else []
		for entry in users:
			if not isinstance(entry, dict):
				continue
			candidate_email = _normalize_spaces(str(entry.get("email") or "")).lower()
			if candidate_email == normalized_email:
				return entry

		total_count: Optional[int] = None
		if isinstance(payload, dict):
			total_raw = payload.get("total")
			if total_raw is not None:
				try:
					total_count = int(total_raw)
				except (TypeError, ValueError):
					total_count = None

		if total_count is not None and page * per_page >= total_count:
			break
		if len(users) < per_page:
			break

	return None


def _update_supabase_auth_user(user_id: str, patch_payload: Dict[str, Any]) -> None:
	if not user_id or not patch_payload:
		return

	try:
		response = requests.put(
			f"{SUPABASE_URL}/auth/v1/admin/users/{user_id}",
			json=patch_payload,
			headers={
				"apikey": SUPABASE_SERVICE_ROLE_KEY,
				"Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}",
				"Content-Type": "application/json",
			},
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
	except requests.RequestException as exc:
		raise RuntimeError(f"Supabase auth update failed: {exc}") from exc

	if response.status_code not in {200, 201}:
		error_msg = _extract_response_error_message(response)
		raise RuntimeError(f"Supabase auth update failed ({response.status_code}): {error_msg}")


def _extract_supabase_session_tokens(payload: Dict[str, Any]) -> Dict[str, Any]:
	session_obj = payload.get("session") if isinstance(payload.get("session"), dict) else {}
	access_token = _normalize_spaces(str(payload.get("access_token") or session_obj.get("access_token") or ""))
	refresh_token = _normalize_spaces(str(payload.get("refresh_token") or session_obj.get("refresh_token") or ""))

	expires_in_raw = payload.get("expires_in")
	if expires_in_raw is None:
		expires_in_raw = session_obj.get("expires_in")

	return {
		"access_token": access_token,
		"refresh_token": refresh_token,
		"expires_in": expires_in_raw,
	}

def _build_internal_login_email(username: str, phone_number: str) -> str:
	seed = re.sub(r"[^a-z0-9]+", "", f"{username}-{phone_number}".lower())
	if not seed:
		seed = uuid4().hex[:12]
	return f"{seed[:24]}-{uuid4().hex[:8]}@ainel-aql.local"


def _is_internal_login_email(email: Optional[str]) -> bool:
	normalized_email = _normalize_spaces(str(email or "")).lower()
	if not normalized_email:
		return False
	return normalized_email.endswith(INTERNAL_LOGIN_EMAIL_SUFFIX)


def _missing_schema_hint(raw_error: str) -> str:
	err = (raw_error or "").strip()
	if "42703" in err:
		return (
			"Database schema is missing new auth/profile columns. "
			"Run supabase/06_profile_name_parts_upgrade.sql (or rerun supabase/01_schema.sql) first."
		)
	return err[:280] if err else "Schema validation failed."


def _ensure_auth_profile_schema_ready() -> None:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/profiles",
		params={
			"select": "id,first_name,last_name,full_name,username,national_id,phone_number,email,role",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code == 200:
		return

	raise HTTPException(
		status_code=503,
		detail=f"Auth schema check failed: {_missing_schema_hint(response.text)}",
	)


def _extract_registration_vehicle_payload(payload: Dict[str, Any]) -> Optional[Dict[str, str]]:
	raw_vehicle = payload.get("vehicle") if isinstance(payload.get("vehicle"), dict) else {}

	def _pick(*keys: str) -> Optional[str]:
		for key in keys:
			value = raw_vehicle.get(key) if isinstance(raw_vehicle, dict) else None
			if value is None:
				value = payload.get(key)
			if value is None:
				continue
			normalized_value = _normalize_spaces(str(value))
			if normalized_value:
				return normalized_value
		return None

	plate_letters_ar = _pick("plate_letters_ar", "plate_letters")
	plate_numbers_ar = _pick("plate_numbers_ar", "plate_numbers")
	plate_letters_en = _pick("plate_letters_en")
	plate_numbers_en = _pick("plate_numbers_en")
	car_name = _pick("car_name", "nickname")
	car_model = _pick("car_model", "model")
	car_color = _pick("car_color", "color")

	has_any_vehicle_field = any(
		[
			plate_letters_ar,
			plate_numbers_ar,
			plate_letters_en,
			plate_numbers_en,
			car_name,
			car_model,
			car_color,
		]
	)
	if not has_any_vehicle_field:
		return None

	if not plate_letters_ar or not plate_numbers_ar:
		raise HTTPException(
			status_code=400,
			detail="Vehicle plate details are incomplete. Provide plate_letters_ar and plate_numbers_ar.",
		)

	vehicle_payload = {
		"plate_letters_ar": plate_letters_ar,
		"plate_numbers_ar": plate_numbers_ar,
	}
	if plate_letters_en:
		vehicle_payload["plate_letters_en"] = plate_letters_en
	if plate_numbers_en:
		vehicle_payload["plate_numbers_en"] = plate_numbers_en
	if car_name:
		vehicle_payload["car_name"] = car_name
	if car_model:
		vehicle_payload["car_model"] = car_model
	if car_color:
		vehicle_payload["car_color"] = car_color

	return vehicle_payload


def _upsert_profile_after_register(
	*,
	user_id: str,
	first_name: str,
	last_name: str,
	username: str,
	phone_number: str,
	email: str,
	national_id: Optional[str],
	role: str = "user",
) -> Optional[Dict[str, Any]]:
	resolved_first_name = _normalize_name_part(first_name)
	resolved_last_name = _normalize_name_part(last_name)
	resolved_full_name = _compose_full_name(resolved_first_name, resolved_last_name) or _normalize_name_part(username)

	payload: Dict[str, Any] = {
		"id": user_id,
		"first_name": resolved_first_name,
		"last_name": resolved_last_name,
		"full_name": resolved_full_name,
		"username": username,
		"phone_number": phone_number,
		"email": email,
	}
	if national_id:
		payload["national_id"] = national_id
	if role:
		payload["role"] = role.lower()

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/profiles",
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="resolution=merge-duplicates,return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201}:
		raise RuntimeError(f"Failed to upsert profile ({response.status_code}): {_missing_schema_hint(response.text)}")

	rows = response.json()
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		return rows[0]
	return _supabase_get_profile_by_user_id(user_id)


def _create_vehicle_for_owner(
	*,
	owner_id: str,
	vehicle_payload: Dict[str, str],
) -> Optional[Dict[str, Any]]:
	if not owner_id or not vehicle_payload:
		return None

	payload: Dict[str, Any] = {
		"owner_id": owner_id,
		"plate_letters_ar": vehicle_payload["plate_letters_ar"],
		"plate_numbers_ar": vehicle_payload["plate_numbers_ar"],
	}
	for optional_key in ["plate_letters_en", "plate_numbers_en", "car_name", "car_model", "car_color"]:
		optional_value = vehicle_payload.get(optional_key)
		if optional_value:
			payload[optional_key] = optional_value

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201}:
		hint = _missing_schema_hint(response.text)
		raise HTTPException(
			status_code=400,
			detail=f"Failed to save vehicle details: {hint}",
		)

	rows = response.json()
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		return rows[0]
	return None


def _extract_vehicle_payload(payload: Dict[str, Any], *, require_plate: bool) -> Dict[str, str]:
	def _pick(*keys: str) -> Optional[str]:
		for key in keys:
			value = payload.get(key)
			if value is None:
				continue
			normalized_value = _normalize_spaces(str(value))
			if normalized_value:
				return normalized_value
		return None

	plate_letters_ar = _pick("plate_letters_ar", "plate_letters")
	plate_numbers_ar = _pick("plate_numbers_ar", "plate_numbers")
	plate_letters_en = _pick("plate_letters_en")
	plate_numbers_en = _pick("plate_numbers_en")
	car_name = _pick("car_name", "nickname")
	car_model = _pick("car_model", "model")
	car_color = _pick("car_color", "color")

	payload_out: Dict[str, str] = {}
	if plate_letters_ar:
		payload_out["plate_letters_ar"] = plate_letters_ar
	if plate_numbers_ar:
		payload_out["plate_numbers_ar"] = plate_numbers_ar
	if plate_letters_en:
		payload_out["plate_letters_en"] = plate_letters_en
	if plate_numbers_en:
		payload_out["plate_numbers_en"] = plate_numbers_en
	if car_name:
		payload_out["car_name"] = car_name
	if car_model:
		payload_out["car_model"] = car_model
	if car_color:
		payload_out["car_color"] = car_color

	if require_plate and (not plate_letters_ar or not plate_numbers_ar):
		raise HTTPException(
			status_code=400,
			detail="plate_letters_ar and plate_numbers_ar are required.",
		)

	if not payload_out:
		raise HTTPException(status_code=400, detail="No vehicle fields were provided.")

	return payload_out


def _fetch_vehicle_for_owner(owner_id: str, vehicle_id: str) -> Optional[Dict[str, Any]]:
	if not owner_id or not vehicle_id:
		return None

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"select": VEHICLE_SELECT_FIELDS,
			"id": f"eq.{vehicle_id}",
			"owner_id": f"eq.{owner_id}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		raise HTTPException(status_code=502, detail=f"Failed to fetch vehicle: {response.text[:260]}")

	rows = response.json()
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		return rows[0]
	return None


def _fetch_vehicle_by_plate_for_owner(
	owner_id: str,
	plate_letters_ar: str,
	plate_numbers_ar: str,
) -> Optional[Dict[str, Any]]:
	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"select": VEHICLE_SELECT_FIELDS,
			"owner_id": f"eq.{owner_id}",
			"plate_letters_ar": f"eq.{plate_letters_ar}",
			"plate_numbers_ar": f"eq.{plate_numbers_ar}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		raise HTTPException(status_code=502, detail=f"Failed to fetch vehicle by plate: {response.text[:260]}")

	rows = response.json()
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		return rows[0]
	return None


def _list_vehicles_for_owner(owner_id: str) -> List[Dict[str, Any]]:
	if not owner_id:
		return []

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"select": VEHICLE_SELECT_FIELDS,
			"owner_id": f"eq.{owner_id}",
			"order": "created_at.desc",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		raise HTTPException(status_code=502, detail=f"Failed to list vehicles: {response.text[:260]}")

	rows = response.json()
	if not isinstance(rows, list):
		return []
	return [row for row in rows if isinstance(row, dict)]


def _update_vehicle_for_owner(owner_id: str, vehicle_id: str, update_payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
	if not owner_id or not vehicle_id:
		return None

	response = requests.patch(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"id": f"eq.{vehicle_id}",
			"owner_id": f"eq.{owner_id}",
			"select": VEHICLE_SELECT_FIELDS,
		},
		json=update_payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		error_hint = _missing_schema_hint(response.text)
		raise HTTPException(status_code=400, detail=f"Failed to update vehicle: {error_hint}")

	if not response.text:
		return _fetch_vehicle_for_owner(owner_id, vehicle_id)

	rows = response.json()
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		return rows[0]

	return _fetch_vehicle_for_owner(owner_id, vehicle_id)


def _delete_vehicle_for_owner(owner_id: str, vehicle_id: str) -> bool:
	if not owner_id or not vehicle_id:
		return False

	response = requests.delete(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"id": f"eq.{vehicle_id}",
			"owner_id": f"eq.{owner_id}",
			"select": "id",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		error_hint = _missing_schema_hint(response.text)
		raise HTTPException(status_code=400, detail=f"Failed to delete vehicle: {error_hint}")

	if not response.text:
		return True

	rows = response.json()
	return isinstance(rows, list) and len(rows) > 0


def _normalize_otp_channel(raw_value: Optional[str]) -> str:
	raw = _normalize_spaces(str(raw_value or "")).lower()
	if raw in {"phone", "sms", "phone_number"}:
		return "phone"
	if raw in {"email", "mail"}:
		return "email"
	raise HTTPException(status_code=400, detail="channel must be 'email' or 'phone'.")


def _resolve_otp_identifier(
	*,
	channel: str,
	payload: Dict[str, Any],
	profile: Optional[Dict[str, Any]],
) -> str:
	if channel not in SUPPORTED_OTP_CHANNELS:
		raise HTTPException(status_code=400, detail="Unsupported OTP channel.")

	if channel == "email":
		candidate = _normalize_spaces(str(payload.get("email") or (profile or {}).get("email") or "")).lower()
		if not candidate or "@" not in candidate:
			raise HTTPException(status_code=400, detail="A valid email is required for email OTP.")
		return candidate

	candidate = _normalize_spaces(
		str(
			payload.get("phone")
			or payload.get("phone_number")
			or (profile or {}).get("phone_number")
			or ""
		)
	)
	if not candidate or not PHONE_PATTERN.fullmatch(candidate):
		raise HTTPException(status_code=400, detail="A valid phone number is required for phone OTP.")
	return candidate


def _request_supabase_otp(
	*,
	channel: str,
	identifier: str,
	create_user: bool,
	redirect_to: Optional[str],
	purpose: Optional[str],
) -> None:
	request_payload: Dict[str, Any] = {
		"create_user": bool(create_user),
	}

	if channel == "email":
		request_payload["email"] = identifier
		if redirect_to:
			if not _is_valid_redirect_url(redirect_to):
				raise HTTPException(status_code=400, detail="redirect_to must be a valid URL or deep-link.")
			request_payload["email_redirect_to"] = redirect_to
			request_payload["options"] = {
				"emailRedirectTo": redirect_to,
			}
	else:
		request_payload["phone"] = identifier
		request_payload["channel"] = "sms"

	try:
		response = requests.post(
			f"{SUPABASE_URL}/auth/v1/otp",
			json=request_payload,
			headers={
				"apikey": SUPABASE_ANON_KEY,
				"Content-Type": "application/json",
			},
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
	except requests.RequestException as exc:
		raise HTTPException(status_code=503, detail=f"Supabase OTP request failed: {exc}") from exc

	if response.status_code not in {200, 201}:
		error_data = response.json() if response.text else {}
		error_msg = ""
		if isinstance(error_data, dict):
			error_msg = str(
				error_data.get("message")
				or error_data.get("error_description")
				or error_data.get("msg")
				or ""
			).strip()
		if not error_msg:
			error_msg = response.text[:220]

		normalized_error = _normalize_spaces(error_msg).lower()
		normalized_purpose = _normalize_spaces(str(purpose or "")).lower()
		if "otp_disabled" in normalized_error or "signups not allowed for otp" in normalized_error:
			if normalized_purpose in {"register", "signup"}:
				raise HTTPException(
					status_code=422,
					detail=(
						"OTP signup is disabled in backend auth configuration (otp_disabled). "
						"purpose=register cannot proceed with OTP until OTP signups are enabled."
					),
				)
			raise HTTPException(
				status_code=422,
				detail="OTP provider is disabled in backend auth configuration (otp_disabled).",
			)

		raise HTTPException(status_code=response.status_code, detail=f"OTP request failed: {error_msg}")


def _verify_supabase_otp(
	*,
	channel: str,
	identifier: str,
	token: str,
	verify_type: Optional[str],
) -> Dict[str, Any]:
	token_value = _normalize_spaces(token)
	if not token_value:
		raise HTTPException(status_code=400, detail="otp_token is required.")

	resolved_type = _normalize_spaces(str(verify_type or "")).lower()
	if not resolved_type:
		resolved_type = "email" if channel == "email" else "sms"

	request_payload: Dict[str, Any] = {
		"token": token_value,
		"type": resolved_type,
	}
	if channel == "email":
		request_payload["email"] = identifier
	else:
		request_payload["phone"] = identifier

	try:
		response = requests.post(
			f"{SUPABASE_URL}/auth/v1/verify",
			json=request_payload,
			headers={
				"apikey": SUPABASE_ANON_KEY,
				"Content-Type": "application/json",
			},
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
	except requests.RequestException as exc:
		raise HTTPException(status_code=503, detail=f"Supabase OTP verify failed: {exc}") from exc

	if response.status_code != 200:
		error_data = response.json() if response.text else {}
		error_msg = ""
		if isinstance(error_data, dict):
			error_msg = str(error_data.get("message") or error_data.get("error_description") or "").strip()
		if not error_msg:
			error_msg = response.text[:220]
		raise HTTPException(status_code=401, detail=f"OTP verification failed: {error_msg}")

	response_payload = response.json() if response.text else {}
	if not isinstance(response_payload, dict):
		return {}
	return response_payload


def _upsert_notification_device_token(
	*,
	user_id: str,
	fcm_token: str,
	platform: Optional[str],
	device_id: Optional[str],
) -> Optional[Dict[str, Any]]:
	payload: Dict[str, Any] = {
		"user_id": user_id,
		"fcm_token": fcm_token,
		"is_active": True,
		"last_seen_at": datetime.now(timezone.utc).isoformat(),
	}
	if platform:
		payload["platform"] = platform
	if device_id:
		payload["device_id"] = device_id

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/notification_device_tokens",
		params={"on_conflict": "fcm_token"},
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="resolution=merge-duplicates,return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201}:
		raise HTTPException(status_code=502, detail=f"Failed to save device token: {response.text[:220]}")

	rows = response.json() if response.text else []
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		return rows[0]
	return None


def _insert_notification_event(
	*,
	user_id: Optional[str],
	event_type: str,
	title: str,
	body: str,
	data: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
	if not _supabase_configured() or not user_id:
		return None

	payload: Dict[str, Any] = {
		"user_id": user_id,
		"event_type": event_type,
		"title": title,
		"body": body,
		"data": data or {},
	}

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/notification_events",
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201}:
		LOGGER.warning("Failed to insert notification event: %s", response.text[:220])
		return None

	rows = response.json() if response.text else []
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		notification_id = rows[0].get("id")
		if notification_id:
			return str(notification_id)
	return None


def _list_security_user_ids() -> List[str]:
	if not _supabase_configured():
		return []

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/profiles",
		params={
			"select": "id",
			"role": "eq.security",
			"limit": "1000",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch security profiles: %s", response.text[:240])
		return []

	rows = response.json()
	if not isinstance(rows, list):
		return []

	return [str(row.get("id")) for row in rows if isinstance(row, dict) and row.get("id")]


def _list_security_user_ids_with_tokens() -> List[str]:
	security_user_ids = _list_security_user_ids()
	if not security_user_ids:
		return []

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/notification_device_tokens",
		params={
			"select": "user_id",
			"user_id": f"in.({','.join(sorted(set(security_user_ids)))})",
			"is_active": "eq.true",
			"limit": "10000",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch security device tokens: %s", response.text[:240])
		return []

	rows = response.json()
	if not isinstance(rows, list):
		return []

	return sorted(
		{
			str(row.get("user_id"))
			for row in rows
			if isinstance(row, dict) and row.get("user_id")
		}
	)


def _notify_security_staff(title: str, body: str, data: Optional[Dict[str, Any]] = None) -> int:
	security_user_ids = _list_security_user_ids_with_tokens()
	if not security_user_ids:
		return 0

	count = 0
	for user_id in security_user_ids:
		notification_id = _insert_notification_event(
			user_id=user_id,
			event_type="cash_payment_needed",
			title=title,
			body=body,
			data=data,
		)
		if notification_id:
			count += 1
	return count


def _list_notification_events_for_user(user_id: str, limit: int) -> List[Dict[str, Any]]:
	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/notification_events",
		params={
			"select": "id,event_type,title,body,data,read_at,created_at",
			"user_id": f"eq.{user_id}",
			"order": "created_at.desc",
			"limit": str(limit),
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		raise HTTPException(status_code=502, detail=f"Failed to load notifications: {response.text[:240]}")

	rows = response.json()
	if not isinstance(rows, list):
		return []
	return [row for row in rows if isinstance(row, dict)]


def _fetch_vehicle_ids_for_owner(owner_id: str) -> List[str]:
	if not owner_id or not _supabase_configured():
		return []

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"select": "id",
			"owner_id": f"eq.{owner_id}",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to list vehicles for owner %s: %s", owner_id, response.text[:240])
		return []

	rows = response.json()
	if not isinstance(rows, list):
		return []

	vehicle_ids: List[str] = []
	for row in rows:
		if not isinstance(row, dict):
			continue
		vehicle_id = row.get("id")
		if vehicle_id:
			vehicle_ids.append(str(vehicle_id))

	return vehicle_ids


def _resolve_or_create_open_parking_session(
	*,
	vehicle_id: Optional[str],
	created_by: Optional[str],
	parking_location: Optional[str],
	plate_arabic: Optional[str] = None,
	plate_english: Optional[str] = None,
) -> Optional[str]:
	if not _supabase_configured():
		return None

	params: Dict[str, str] = {
		"select": "id,parking_location,status,check_out_at",
		"check_out_at": "is.null",
		"status": "not.in.(exited,left_without_payment)",
		"order": "check_in_at.desc",
		"limit": "1",
	}
	
	if vehicle_id:
		params["vehicle_id"] = f"eq.{vehicle_id}"
	elif plate_arabic or plate_english:
		params["vehicle_id"] = "is.null"
		# Try to match by either plate
		plate_filters = []
		if plate_arabic:
			plate_filters.append(f"plate_arabic.eq.{plate_arabic}")
		if plate_english:
			plate_filters.append(f"plate_english.eq.{plate_english}")
		if plate_filters:
			params["or"] = f"({','.join(plate_filters)})"
	else:
		return None

	open_session_response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params=params,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if open_session_response.status_code != 200:
		raise RuntimeError(
			f"Failed to query parking_sessions ({open_session_response.status_code}): "
			f"{open_session_response.text[:280]}"
		)

	rows = open_session_response.json()
	if isinstance(rows, list) and rows:
		session_id = rows[0].get("id")
		if not session_id:
			return None

		current_location = rows[0].get("parking_location")
		if parking_location and parking_location != current_location:
			update_response = requests.patch(
				f"{SUPABASE_URL}/rest/v1/parking_sessions",
				params={"id": f"eq.{session_id}"},
				json={"parking_location": parking_location},
				headers=_supabase_headers(
					api_key=SUPABASE_SERVICE_ROLE_KEY,
					bearer=SUPABASE_SERVICE_ROLE_KEY,
					content_type="application/json",
				),
				timeout=SUPABASE_TIMEOUT_SECONDS,
			)
			if update_response.status_code not in {200, 204}:
				raise RuntimeError(
					f"Failed to update parking_location ({update_response.status_code}): "
					f"{update_response.text[:280]}"
				)

		return str(session_id)

	payload: Dict[str, Any] = {
		"vehicle_id": vehicle_id,
		"status": "entered",
		"plate_arabic": plate_arabic,
		"plate_english": plate_english,
		"check_in_at": datetime.now(timezone.utc).isoformat(),
	}
	if created_by:
		payload["created_by"] = created_by
	if parking_location:
		payload["parking_location"] = parking_location

	create_response = requests.post(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if create_response.status_code not in {200, 201}:
		raise RuntimeError(
			f"Failed to create parking_session ({create_response.status_code}): {create_response.text[:300]}"
		)

	created_rows = create_response.json()
	if isinstance(created_rows, list) and created_rows:
		session_id = created_rows[0].get("id")
		return str(session_id) if session_id else None

	return None


def _insert_payment_transaction_record(
	*,
	created_by: Optional[str],
	session_id: str,
	amount_cents: int,
	currency: str,
	merchant_order_id: str,
	paymob_order_id: Optional[str],
	parking_location: Optional[str],
	request_payload: Dict[str, Any],
	response_payload: Dict[str, Any],
	provider: str = "paymob",
	status: str = "pending",
	payment_reference: Optional[str] = None,
) -> Optional[str]:
	if not _supabase_configured():
		return None

	payload: Dict[str, Any] = {
		"provider": provider,
		"status": status,
		"session_id": session_id,
		"amount_cents": amount_cents,
		"currency": currency,
		"merchant_order_id": merchant_order_id,
		"request_payload": request_payload,
		"response_payload": response_payload,
	}
	if created_by:
		payload["created_by"] = created_by
	if paymob_order_id:
		payload["paymob_order_id"] = paymob_order_id
	if payment_reference:
		payload["payment_reference"] = payment_reference
	if parking_location:
		payload["parking_location"] = parking_location

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/payment_transactions",
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201}:
		raise RuntimeError(f"Failed to insert payment_transactions row ({response.status_code}): {response.text[:400]}")

	rows = response.json()
	if isinstance(rows, list) and rows:
		payment_id = rows[0].get("id")
		return str(payment_id) if payment_id else None
	return None


def _update_payment_transaction_by_merchant_order_id(
	*,
	merchant_order_id: str,
	status: str,
	payment_reference: Optional[str],
	paymob_order_id: Optional[str],
	response_payload: Dict[str, Any],
) -> None:
	if not _supabase_configured() or not merchant_order_id:
		return

	patch_payload: Dict[str, Any] = {
		"status": status,
		"response_payload": response_payload,
	}
	if payment_reference:
		patch_payload["payment_reference"] = payment_reference
	if paymob_order_id:
		patch_payload["paymob_order_id"] = paymob_order_id

	response = requests.patch(
		f"{SUPABASE_URL}/rest/v1/payment_transactions",
		params={"merchant_order_id": f"eq.{merchant_order_id}"},
		json=patch_payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		raise RuntimeError(
			f"Failed to update payment transaction ({response.status_code}): {response.text[:280]}"
		)


def _get_parking_session_owner_id(session_id: str) -> Optional[str]:
	if not _supabase_configured() or not session_id:
		return None

	session_response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={
			"select": "vehicle_id",
			"id": f"eq.{session_id}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if session_response.status_code != 200:
		return None

	session_rows = session_response.json()
	if not isinstance(session_rows, list) or not session_rows:
		return None
	vehicle_id = session_rows[0].get("vehicle_id")
	if not vehicle_id:
		return None

	vehicle_response = requests.get(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"select": "owner_id",
			"id": f"eq.{vehicle_id}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if vehicle_response.status_code != 200:
		return None

	vehicle_rows = vehicle_response.json()
	if not isinstance(vehicle_rows, list) or not vehicle_rows:
		return None

	owner_id = vehicle_rows[0].get("owner_id")
	return str(owner_id) if owner_id else None


def _mark_parking_session_paid(session_id: str, payment_reference: Optional[str] = None) -> None:
	if not _supabase_configured() or not session_id:
		return

	patch_payload: Dict[str, Any] = {
		"payment_confirmed_at": datetime.now(timezone.utc).isoformat(),
		"status": "paid",
	}
	if payment_reference:
		patch_payload["notes"] = f"Paymob ref: {payment_reference}"

	response = requests.patch(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={"id": f"eq.{session_id}"},
		json=patch_payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		raise RuntimeError(f"Failed to update parking session as paid ({response.status_code}): {response.text[:280]}")


def _set_parking_session_location(session_id: str, parking_location: Optional[str]) -> None:
	if not _supabase_configured() or not session_id or not parking_location:
		return

	response = requests.patch(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={"id": f"eq.{session_id}"},
		json={"parking_location": parking_location},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		raise RuntimeError(f"Failed to set parking location ({response.status_code}): {response.text[:280]}")


def _get_payment_session_id_by_merchant_order_id(merchant_order_id: str) -> Optional[str]:
	if not _supabase_configured() or not merchant_order_id:
		return None

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/payment_transactions",
		params={
			"select": "session_id",
			"merchant_order_id": f"eq.{merchant_order_id}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		return None

	rows = response.json()
	if not isinstance(rows, list) or not rows:
		return None
	session_id = rows[0].get("session_id")
	return str(session_id) if session_id else None


def _call_paymob_api(path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
	url = f"{PAYMOB_BASE_URL}{path}"
	try:
		response = requests.post(url, json=payload, timeout=SUPABASE_TIMEOUT_SECONDS)
	except requests.RequestException as exc:
		raise RuntimeError(f"Paymob API call failed: {exc}") from exc

	if response.status_code not in {200, 201}:
		raise RuntimeError(f"Paymob API error ({response.status_code}): {response.text[:400]}")

	try:
		body = response.json()
	except ValueError as exc:
		raise RuntimeError("Paymob returned a non-JSON response.") from exc

	if not isinstance(body, dict):
		raise RuntimeError("Unexpected Paymob response format.")
	return body


def _paymob_create_checkout(
	*,
	amount_cents: int,
	currency: str,
	merchant_order_id: str,
	request_user: Dict[str, Any],
) -> Dict[str, Any]:
	auth_payload = {
		"api_key": PAYMOB_API_KEY,
	}
	auth_response = _call_paymob_api("/auth/tokens", auth_payload)
	auth_token = str(auth_response.get("token") or "").strip()
	if not auth_token:
		raise RuntimeError("Paymob did not return an auth token.")

	order_payload = {
		"auth_token": auth_token,
		"delivery_needed": False,
		"amount_cents": str(amount_cents),
		"currency": currency,
		"merchant_order_id": merchant_order_id,
		"items": [],
	}
	order_response = _call_paymob_api("/ecommerce/orders", order_payload)
	order_id = order_response.get("id")
	if order_id is None:
		raise RuntimeError("Paymob did not return an order id.")

	billing_data = {
		"first_name": "Ain",
		"last_name": "User",
		"email": request_user.get("email") or "user@example.com",
		"phone_number": "NA",
		"apartment": "NA",
		"floor": "NA",
		"street": "NA",
		"building": "NA",
		"shipping_method": "NA",
		"postal_code": "NA",
		"city": "Cairo",
		"country": "EG",
		"state": "Cairo",
	}

	payment_key_payload = {
		"auth_token": auth_token,
		"amount_cents": str(amount_cents),
		"expiration": 3600,
		"order_id": order_id,
		"currency": currency,
		"integration_id": int(PAYMOB_INTEGRATION_ID),
		"billing_data": billing_data,
	}
	payment_key_response = _call_paymob_api("/acceptance/payment_keys", payment_key_payload)
	payment_token = str(payment_key_response.get("token") or "").strip()
	if not payment_token:
		raise RuntimeError("Paymob did not return a payment token.")

	return {
		"paymob_order_id": str(order_id),
		"payment_token": payment_token,
		"iframe_url": f"{PAYMOB_BASE_URL}/acceptance/iframes/{PAYMOB_IFRAME_ID}?payment_token={payment_token}",
		"auth_response": auth_response,
		"order_response": order_response,
		"payment_key_response": payment_key_response,
	}


def _run_weekly_refresh_non_core() -> Dict[str, Any]:
	if not _supabase_configured():
		raise RuntimeError("Supabase is not configured.")

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/rpc/weekly_refresh_non_core",
		json={},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201, 204}:
		raise RuntimeError(
			f"Failed to run weekly refresh RPC ({response.status_code}): {response.text[:360]}"
		)

	if response.status_code == 204 or not response.text:
		return {"truncated_tables": [], "count": 0}

	result = response.json()
	if isinstance(result, dict):
		return result
	return {"result": result}


def _resolve_vehicle_record_by_plate(
	plate_text_ar: str,
	plate_text_en: str,
	user_id: Optional[str],
) -> Optional[Dict[str, Any]]:
	if not _supabase_configured():
		return None

	plate_letters_ar, plate_numbers_ar = _split_arabic_plate(plate_text_ar)
	plate_letters_en, plate_numbers_en = _split_english_plate(plate_text_en)

	def _number_variants(num_str: Optional[str]) -> List[str]:
		"""Build all plausible stored forms of a plate number string."""
		if not num_str:
			return []
		raw = num_str.strip()
		if not raw:
			return []

		no_space = "".join(raw.split())

		western = no_space
		arabic_indic = no_space
		_ARABIC_INDIC = '\u0660\u0661\u0662\u0663\u0664\u0665\u0666\u0667\u0668\u0669'
		_WESTERN      = '0123456789'
		for ar, en in zip(_ARABIC_INDIC, _WESTERN):
			western = western.replace(ar, en)
			arabic_indic = arabic_indic.replace(en, ar)

		# Collect unique non-empty variants
		candidates = set()
		candidates.add(raw)
		candidates.add(no_space)
		candidates.add(western)
		candidates.add(arabic_indic)
		# Also add spaced variants (e.g. "1 2 4 5")
		if len(western) > 1:
			candidates.add(" ".join(list(western)))
		if len(arabic_indic) > 1:
			candidates.add(" ".join(list(arabic_indic)))
		# Add reversed variants ΓÇö OCR can mis-order digit bboxes on some plates
		# (e.g. reads 748 as 847). Query includes reversed forms so the candidate
		# row is fetched; normalize_for_cmp still does exact-order check first,
		# then a sorted-digits fallback.
		rev_western = western[::-1]
		rev_arabic  = arabic_indic[::-1]
		if rev_western != western:
			candidates.add(rev_western)
		if rev_arabic != arabic_indic:
			candidates.add(rev_arabic)
		if len(rev_western) > 1 and rev_western != western:
			candidates.add(" ".join(list(rev_western)))
		if len(rev_arabic) > 1 and rev_arabic != arabic_indic:
			candidates.add(" ".join(list(rev_arabic)))
		return [v for v in candidates if v]

	def normalize_for_cmp(text: Optional[str]) -> str:
		if not text:
			return ""
		t = "".join(text.split()).lower()
		t = t.replace('\u0627', '\u0623').replace('\u0625', '\u0623').replace('\u0622', '\u0623')
		t = t.replace('\u0629', '\u0647')
		t = t.replace('\u064a', '\u0649')
		_ARABIC_INDIC = '\u0660\u0661\u0662\u0663\u0664\u0665\u0666\u0667\u0668\u0669'
		_WESTERN      = '0123456789'
		for ar, en in zip(_ARABIC_INDIC, _WESTERN):
			t = t.replace(ar, en)
		return t

	matched_vehicles: List[Tuple[Dict[str, Any], str]] = []

	def _fetch_and_filter(ocr_numbers: Optional[str], ocr_letters: Optional[str], lang: str):
		if not ocr_numbers or not ocr_letters:
			return

		variants = _number_variants(ocr_numbers)
		if not variants:
			return

		col = "plate_numbers_ar" if lang == "ar" else "plate_numbers_en"

		# Build PostgREST OR filter with exact eq matches for every variant
		# Values with spaces MUST be double-quoted in PostgREST filters.
		conditions = []
		for v in variants:
			safe_v = v.replace('"', '')
			conditions.append(f'{col}.eq."{safe_v}"')
		or_filter = "(" + ",".join(conditions) + ")"

		params = {
			"select": "id,owner_id,plate_letters_ar,plate_numbers_ar,plate_letters_en,plate_numbers_en,car_model,car_color,is_active",
			"is_active": "eq.true",
			"or": or_filter,
		}

		LOGGER.info(
			"plate_query | lang=%s | col=%s | variants=%s | or_filter=%s",
			lang, col, variants, or_filter,
		)

		response = requests.get(
			f"{SUPABASE_URL}/rest/v1/vehicles",
			params=params,
			headers=_supabase_headers(
				api_key=SUPABASE_SERVICE_ROLE_KEY,
				bearer=SUPABASE_SERVICE_ROLE_KEY,
			),
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
		if response.status_code == 200:
			rows = response.json()
			if isinstance(rows, list):
				ocr_num_norm = normalize_for_cmp(ocr_numbers)
				ocr_let_norm = normalize_for_cmp(ocr_letters)

				let_col = "plate_letters_ar" if lang == "ar" else "plate_letters_en"

				LOGGER.info(
					"plate_lookup | lang=%s | ocr_num_raw='%s' ocr_let_raw='%s' | "
					"ocr_num_norm='%s' ocr_let_norm='%s' | candidate_rows=%d",
					lang, ocr_numbers, ocr_letters,
					ocr_num_norm, ocr_let_norm, len(rows),
				)

				for row in rows:
					db_num_norm = normalize_for_cmp(row.get(col, ""))
					db_let_norm = normalize_for_cmp(row.get(let_col, ""))
					# Primary: exact digit order + exact letters
					is_match = (db_num_norm == ocr_num_norm and db_let_norm == ocr_let_norm)
					# Fallback: same digits in any order (handles OCR bbox mis-ordering)
					# Only applies when letters match exactly to avoid false positives.
					if not is_match and db_let_norm == ocr_let_norm:
						is_match = sorted(db_num_norm) == sorted(ocr_num_norm)
						if is_match:
							LOGGER.warning(
								"plate_row_cmp | ANAGRAM_MATCH (OCR digit order wrong) | "
								"lang=%s | vehicle_id=%s | db_num='%s' ocr_num='%s'",
								lang, row.get("id"), db_num_norm, ocr_num_norm,
							)
					LOGGER.info(
						"plate_row_cmp | lang=%s | vehicle_id=%s | "
						"db_num_raw='%s' db_let_raw='%s' | "
						"db_num_norm='%s' db_let_norm='%s' | match=%s",
						lang, row.get("id"),
						row.get(col, ""), row.get(let_col, ""),
						db_num_norm, db_let_norm, is_match,
					)
					if is_match:
						matched_vehicles.append((row, lang))
		else:
			LOGGER.warning("Failed to resolve vehicle by plate %s (%s): %s", lang, response.status_code, response.text[:300])

	_fetch_and_filter(plate_numbers_ar, plate_letters_ar, "ar")
	if not matched_vehicles:
		_fetch_and_filter(plate_numbers_en, plate_letters_en, "en")

	if not matched_vehicles:
		LOGGER.info("plate_resolution | NO_MATCH | ar='%s' en='%s'", plate_text_ar, plate_text_en)
		return None

	# If user_id is provided, prioritize their vehicle
	if user_id:
		for v, lang in matched_vehicles:
			if str(v.get("owner_id")) == str(user_id):
				v["_match_path"] = f"{lang}_local_filter"
				LOGGER.info("plate_resolution | match_path=%s_local_filter | user_id=%s | vehicle_id=%s", lang, user_id, v.get("id"))
				return v

	# Otherwise return the first matched
	v, lang = matched_vehicles[0]
	v["_match_path"] = f"{lang}_local_filter"
	LOGGER.info("plate_resolution | match_path=%s_local_filter | global | vehicle_id=%s", lang, v.get("id"))
	return v


def _parse_iso_datetime(value: Optional[Any]) -> Optional[datetime]:
	if value is None:
		return None
	raw = str(value).strip()
	if not raw:
		return None

	try:
		parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
	except ValueError:
		return None

	if parsed.tzinfo is None:
		return parsed.replace(tzinfo=timezone.utc)
	return parsed.astimezone(timezone.utc)


def _resolve_location_capacity(
	location_key: Optional[str],
	*,
	explicit_capacity: Optional[int] = None,
) -> Tuple[int, str]:
	if explicit_capacity is not None:
		return max(0, int(explicit_capacity)), "location_pricing"

	if location_key and location_key in PARKING_LOCATION_CAPACITIES:
		return max(0, int(PARKING_LOCATION_CAPACITIES[location_key])), "location_capacity_map"

	return GARAGE_TOTAL_CAPACITY, "default"


def _resolve_location_pricing_policy(parking_location: Optional[Any]) -> Dict[str, Any]:
	candidates: List[str] = []
	if parking_location is not None:
		raw = str(parking_location).strip()
		if raw:
			for part in raw.split(","):
				location_key = _normalize_location_key_for_config(part)
				if location_key:
					candidates.append(location_key)

	for location_key in candidates:
		location_policy = PARKING_LOCATION_PRICING.get(location_key)
		if isinstance(location_policy, dict):
			capacity_value, capacity_source = _resolve_location_capacity(
				location_key,
				explicit_capacity=int(location_policy.get("capacity", GARAGE_TOTAL_CAPACITY)),
			)
			return {
				"pricing_location": location_key,
				"billing_mode": _parse_billing_mode(location_policy.get("billing_mode"), default="hourly"),
				"base_rate_egp": max(0.0, float(location_policy.get("base_rate_egp", 0.0))),
				"extra_after_first_hour_egp": max(0.0, float(location_policy.get("extra_after_first_hour_egp", 0.0))),
				"rate_source": "location_pricing",
				"location_capacity": capacity_value,
				"location_capacity_source": capacity_source,
			}

		hourly_rate_override = PARKING_LOCATION_HOURLY_RATES.get(location_key)
		if hourly_rate_override is not None:
			capacity_value, capacity_source = _resolve_location_capacity(location_key)
			return {
				"pricing_location": location_key,
				"billing_mode": "hourly",
				"base_rate_egp": max(0.0, float(hourly_rate_override)),
				"extra_after_first_hour_egp": max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP),
				"rate_source": "location_hourly_legacy",
				"location_capacity": capacity_value,
				"location_capacity_source": capacity_source,
			}

	fallback_location = candidates[0] if candidates else None
	default_capacity, default_capacity_source = _resolve_location_capacity(fallback_location)
	default_base_rate = PARKING_HOURLY_RATE_EGP if PARKING_BILLING_MODE == "hourly" else PARKING_DAILY_RATE_EGP
	return {
		"pricing_location": fallback_location,
		"billing_mode": PARKING_BILLING_MODE,
		"base_rate_egp": max(0.0, default_base_rate),
		"extra_after_first_hour_egp": max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP),
		"rate_source": "default",
		"location_capacity": default_capacity,
		"location_capacity_source": default_capacity_source,
	}


def _compute_parking_fee_cents(
	*,
	duration_seconds: float,
	billing_mode: str,
	base_rate_egp: float,
	extra_after_first_hour_egp: float,
) -> Tuple[int, int, int]:
	if duration_seconds <= 0:
		return 0, 0, 0

	base_rate_cents = max(0, int(round(max(0.0, base_rate_egp) * 100.0)))
	extra_rate_cents = max(0, int(round(max(0.0, extra_after_first_hour_egp) * 100.0)))

	if billing_mode == "daily":
		base_window_seconds = 24 * 3600
	else:
		base_window_seconds = 3600

	extra_hours_billed = 0
	if duration_seconds > base_window_seconds and extra_rate_cents > 0:
		extra_hours_billed = int(math.ceil((duration_seconds - base_window_seconds) / 3600.0))

	fee_cents = base_rate_cents + (extra_hours_billed * extra_rate_cents)
	return max(0, fee_cents), extra_hours_billed, int(base_window_seconds // 3600)


def _compute_session_time_and_fee(
	session_row: Optional[Dict[str, Any]],
	now_utc: Optional[datetime] = None,
	parking_location_override: Optional[str] = None,
) -> Dict[str, Any]:
	reference_now = now_utc or datetime.now(timezone.utc)
	location_for_pricing = parking_location_override
	if isinstance(session_row, dict):
		location_for_pricing = session_row.get("parking_location") or parking_location_override

	pricing_policy = _resolve_location_pricing_policy(location_for_pricing)
	billing_mode = str(pricing_policy.get("billing_mode") or "hourly")
	base_rate_egp = max(0.0, float(pricing_policy.get("base_rate_egp") or 0.0))
	extra_after_first_hour_egp = max(0.0, float(pricing_policy.get("extra_after_first_hour_egp") or 0.0))
	pricing_location = pricing_policy.get("pricing_location")
	rate_source = str(pricing_policy.get("rate_source") or "default")
	location_capacity = max(0, int(pricing_policy.get("location_capacity") or 0))
	location_capacity_source = str(pricing_policy.get("location_capacity_source") or "default")
	hourly_rate_egp = base_rate_egp if billing_mode == "hourly" else round(base_rate_egp / 24.0, 2)
	app_fee_egp = max(0.0, APP_SERVICE_FEE_EGP)
	app_fee_cents = int(round(app_fee_egp * 100.0))
	result = {
		"pricing_location": pricing_location,
		"rate_source": rate_source,
		"billing_mode": billing_mode,
		"base_rate_egp": round(base_rate_egp, 2),
		"extra_after_first_hour_egp": round(extra_after_first_hour_egp, 2),
		"location_capacity": location_capacity,
		"location_capacity_source": location_capacity_source,
		"base_window_hours": 24 if billing_mode == "daily" else 1,
		"hourly_rate_egp": round(hourly_rate_egp, 2),
		"app_fee_egp": round(app_fee_egp, 2),
		"app_fee_cents": app_fee_cents,
		"duration_minutes": 0.0,
		"extra_hours_billed": 0,
		"parking_fee_egp": 0.0,
		"parking_fee_cents": 0,
		"total_fee_egp": 0.0,
		"total_fee_cents": 0,
		"billing_reference_time": reference_now.isoformat(),
	}

	if not isinstance(session_row, dict):
		return result

	check_in_dt = _parse_iso_datetime(session_row.get("check_in_at"))
	if check_in_dt is None:
		return result

	check_out_dt = _parse_iso_datetime(session_row.get("check_out_at"))
	end_dt = check_out_dt or reference_now
	duration_seconds = max(0.0, (end_dt - check_in_dt).total_seconds())
	duration_minutes = round(duration_seconds / 60.0, 2)
	parking_fee_cents, extra_hours_billed, base_window_hours = _compute_parking_fee_cents(
		duration_seconds=duration_seconds,
		billing_mode=billing_mode,
		base_rate_egp=base_rate_egp,
		extra_after_first_hour_egp=extra_after_first_hour_egp,
	)
	total_fee_cents = parking_fee_cents + app_fee_cents

	result["duration_minutes"] = duration_minutes
	result["extra_hours_billed"] = extra_hours_billed
	result["base_window_hours"] = base_window_hours
	result["parking_fee_cents"] = parking_fee_cents
	result["parking_fee_egp"] = round(parking_fee_cents / 100.0, 2)
	result["total_fee_cents"] = total_fee_cents
	result["total_fee_egp"] = round(total_fee_cents / 100.0, 2)
	return result


def _fetch_parking_session_by_id(session_id: Optional[str]) -> Optional[Dict[str, Any]]:
	if not _supabase_configured() or not session_id:
		return None

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={
			"select": "id,vehicle_id,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
			"id": f"eq.{session_id}",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch parking session by id %s: %s", session_id, response.text[:300])
		return None

	rows = response.json()
	if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
		return None
	return rows[0]


def _fetch_open_parking_session_by_vehicle(vehicle_id: Optional[str]) -> Optional[Dict[str, Any]]:
	if not _supabase_configured() or not vehicle_id:
		return None

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={
			"select": "id,vehicle_id,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
			"vehicle_id": f"eq.{vehicle_id}",
			"check_out_at": "is.null",
			"order": "check_in_at.desc",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch open parking session for vehicle %s: %s", vehicle_id, response.text[:300])
		return None

	rows = response.json()
	if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
		return None
	return rows[0]


def _fetch_latest_parking_session_by_vehicle(vehicle_id: Optional[str]) -> Optional[Dict[str, Any]]:
	if not _supabase_configured() or not vehicle_id:
		return None

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={
			"select": "id,vehicle_id,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
			"vehicle_id": f"eq.{vehicle_id}",
			"order": "check_in_at.desc",
			"limit": "1",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch latest parking session for vehicle %s: %s", vehicle_id, response.text[:300])
		return None

	rows = response.json()
	if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
		return None
	return rows[0]


def _insert_parking_session(payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
	if not _supabase_configured():
		return None

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 201}:
		LOGGER.error("Failed to insert parking session (%s): %s", response.status_code, response.text[:300])
		return None

	rows = response.json()
	if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
		return None
	return rows[0]


def _link_unmatched_events_to_session(plate_ar: str, session_id: str) -> None:
	if not _supabase_configured() or not plate_ar or not session_id:
		return

	# Search for all events where session_id is null and plate matches
	# We use plate_arabic or ocr_arabic in metadata
	response = requests.patch(
		f"{SUPABASE_URL}/rest/v1/car_events",
		params={
			"session_id": "is.null",
			"ocr_arabic": f"eq.{plate_ar}"
		},
		json={"session_id": session_id},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		LOGGER.warning("Failed to link unmatched events for plate %s to session %s: %s", plate_ar, session_id, response.text[:300])


def _update_parking_session(session_id: str, patch_payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
	if not _supabase_configured() or not session_id:
		return None

	response = requests.patch(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={"id": f"eq.{session_id}"},
		json=patch_payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		raise RuntimeError(f"Failed to update parking session ({response.status_code}): {response.text[:300]}")

	if response.status_code == 204:
		return _fetch_parking_session_by_id(session_id)

	rows = response.json()
	if isinstance(rows, list) and rows and isinstance(rows[0], dict):
		return rows[0]
	return _fetch_parking_session_by_id(session_id)


def _derive_payment_status_from_session(session_row: Optional[Dict[str, Any]]) -> str:
	if not isinstance(session_row, dict):
		return "no_active_session"

	payment_confirmed_at = session_row.get("payment_confirmed_at")
	check_out_at = session_row.get("check_out_at")
	left_within_5 = bool(session_row.get("left_within_5_minutes"))
	status_value = str(session_row.get("status") or "").strip().lower()

	if payment_confirmed_at is None and check_out_at is None:
		return "unpaid"
	if payment_confirmed_at is not None and check_out_at is None:
		if status_value == "overstayed":
			return "paid_grace_period_expired"
		return "paid_waiting_exit"
	if payment_confirmed_at is None and check_out_at is not None:
		return "left_without_payment"
	if left_within_5:
		return "left_within_5_minutes_after_payment"
	return "left_after_5_minutes_after_payment"


def _build_gate_decision(
	*,
	event_type: str,
	session_row: Optional[Dict[str, Any]],
	plate_detected: bool,
) -> Dict[str, Any]:
	if event_type != "exit":
		return {
			"allowed": True,
			"action": "open_gate",
			"reason": "entry_access_granted",
		}

	if not plate_detected:
		return {
			"allowed": False,
			"action": "manual_open_only",
			"reason": "plate_not_detected",
		}

	payment_status = _derive_payment_status_from_session(session_row)
	if payment_status in {"paid_waiting_exit", "left_within_5_minutes_after_payment"}:
		return {
			"allowed": True,
			"action": "open_gate",
			"reason": "paid_within_grace_period",
			"payment_status": payment_status,
		}

	if payment_status == "paid_grace_period_expired":
		return {
			"allowed": False,
			"action": "deny_and_alert_security",
			"reason": "grace_period_expired",
			"payment_status": payment_status,
		}

	if payment_status == "no_active_session":
		return {
			"allowed": False,
			"action": "manual_review_required",
			"reason": "no_active_session",
			"payment_status": payment_status,
		}

	return {
		"allowed": False,
		"action": "deny_gate",
		"reason": "payment_not_confirmed",
		"payment_status": payment_status,
	}


def _resolve_session_for_event(
	*,
	event_type: str,
	vehicle_id: Optional[str],
	created_by: Optional[str],
	parking_location: Optional[str],
	plate_arabic: Optional[str] = None,
	plate_english: Optional[str] = None,
) -> Tuple[Optional[str], Optional[Dict[str, Any]]]:
	if not _supabase_configured():
		return None, None

	if event_type == "entry":
		session_id = _resolve_or_create_open_parking_session(
			vehicle_id=vehicle_id,
			created_by=created_by,
			parking_location=parking_location,
			plate_arabic=plate_arabic,
			plate_english=plate_english,
		)
		return session_id, _fetch_parking_session_by_id(session_id)

	if event_type == "exit":
		open_session = None
		if vehicle_id:
			open_session = _fetch_open_parking_session_by_vehicle(vehicle_id)
		elif plate_arabic or plate_english:
			# Find open session for guest car by plate
			open_session_response = requests.get(
				f"{SUPABASE_URL}/rest/v1/parking_sessions",
				params={
					"select": "*",
					"vehicle_id": "is.null",
					"check_out_at": "is.null",
					"or": f"(plate_arabic.eq.{plate_arabic},plate_english.eq.{plate_english})",
					"order": "check_in_at.desc",
					"limit": "1",
				},
				headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY),
				timeout=SUPABASE_TIMEOUT_SECONDS,
			)
			if open_session_response.status_code == 200:
				rows = open_session_response.json()
				if isinstance(rows, list) and rows:
					open_session = rows[0]

		if not open_session:
			# Fallback to latest session (even if closed)
			latest_session = None
			if vehicle_id:
				latest_session = _fetch_latest_parking_session_by_vehicle(vehicle_id)
			elif plate_arabic or plate_english:
				latest_session_response = requests.get(
					f"{SUPABASE_URL}/rest/v1/parking_sessions",
					params={
						"select": "*",
						"vehicle_id": "is.null",
						"or": f"(plate_arabic.eq.{plate_arabic},plate_english.eq.{plate_english})",
						"order": "check_in_at.desc",
						"limit": "1",
					},
					headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY),
					timeout=SUPABASE_TIMEOUT_SECONDS,
				)
				if latest_session_response.status_code == 200:
					rows = latest_session_response.json()
					if isinstance(rows, list) and rows:
						latest_session = rows[0]
			
			session_id = str(latest_session.get("id")) if isinstance(latest_session, dict) and latest_session.get("id") else None
			return session_id, latest_session

		session_id = str(open_session.get("id") or "").strip()
		if not session_id:
			return None, None

		if parking_location and parking_location != open_session.get("parking_location"):
			_set_parking_session_location(session_id, parking_location)
			open_session["parking_location"] = parking_location

		now_utc = datetime.now(timezone.utc)
		payment_dt = _parse_iso_datetime(open_session.get("payment_confirmed_at"))
		if payment_dt is None:
			# Keep session open when payment is missing; this powers gate deny logic.
			return session_id, open_session

		left_within_5 = now_utc <= (payment_dt + timedelta(minutes=5))

		patch_payload: Dict[str, Any]
		if left_within_5:
			patch_payload = {
				"check_out_at": now_utc.isoformat(),
				"status": "exited",
			}
		else:
			# Grace window expired after payment; RESET session to start a new billing cycle (penalty).
			patch_payload = {
				"status": "entered",
				"payment_confirmed_at": None,
				"check_in_at": now_utc.isoformat(),
				"notes": f"{open_session.get('notes') or ''} | Grace expired. Session reset. Old payment: {open_session.get('payment_confirmed_at')}".strip()
			}

		if parking_location:
			patch_payload["parking_location"] = parking_location

		updated_session = _update_parking_session(session_id, patch_payload)
		return session_id, updated_session

	# ocr_scan: keep latest context but do not modify session state.
	open_session = _fetch_open_parking_session_by_vehicle(vehicle_id)
	if open_session:
		session_id = str(open_session.get("id") or "").strip() or None
		if session_id and parking_location and parking_location != open_session.get("parking_location"):
			_set_parking_session_location(session_id, parking_location)
			open_session["parking_location"] = parking_location
		return session_id, open_session

	latest_session = _fetch_latest_parking_session_by_vehicle(vehicle_id)
	session_id = str(latest_session.get("id")) if isinstance(latest_session, dict) and latest_session.get("id") else None
	return session_id, latest_session


def _insert_car_event_record(
	*,
	event_type: str,
	created_by: Optional[str],
	vehicle_id: Optional[str],
	session_id: Optional[str],
	raw_image_path: str,
	user_split_image_path: str,
	admin_annotated_image_path: str,
	ocr_arabic: str,
	ocr_english: str,
	ocr_raw_labels: List[str],
	ocr_confidence: Optional[float],
	parking_location: Optional[str],
	event_metadata: Dict[str, Any],
) -> Optional[str]:
	if not _supabase_configured():
		return None

	payload: Dict[str, Any] = {
		"event_type": event_type,
		"raw_image_path": raw_image_path,
		"user_split_image_path": user_split_image_path,
		"admin_annotated_image_path": admin_annotated_image_path,
		"ocr_arabic": ocr_arabic,
		"ocr_english": ocr_english,
		"ocr_raw_labels": ocr_raw_labels,
		"event_metadata": event_metadata,
	}

	if created_by:
		payload["created_by"] = created_by
	if vehicle_id:
		payload["vehicle_id"] = vehicle_id
	if session_id:
		payload["session_id"] = session_id
	if ocr_confidence is not None:
		payload["ocr_confidence"] = ocr_confidence
	if parking_location:
		payload["parking_location"] = parking_location

	response = requests.post(
		f"{SUPABASE_URL}/rest/v1/car_events",
		json=payload,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)

	if response.status_code not in {200, 201}:
		raise RuntimeError(f"Failed to insert car_events row ({response.status_code}): {response.text[:400]}")

	rows = response.json()
	if isinstance(rows, list) and rows:
		return rows[0].get("id")
	return None


def _build_new_car_alert_payload(
	*,
	plate_info: Dict[str, Any],
	vehicle_id: Optional[str],
	owner_id: Optional[str],
) -> Dict[str, Any]:
	plate_detected = _is_plate_detected(plate_info)
	plate_arabic_raw = _normalize_spaces(str(plate_info.get("arabic") or ""))
	plate_english_raw = _normalize_spaces(str(plate_info.get("english") or ""))
	plate_arabic = plate_arabic_raw if plate_arabic_raw and plate_arabic_raw.upper() != "N/A" else None
	plate_english = plate_english_raw if plate_english_raw and plate_english_raw.upper() != "N/A" else None

	payload: Dict[str, Any] = {
		"is_new_car": False,
		"notify_roles": [],
		"action": "none",
		"reason": "",
		"message": "",
		"vehicle_id": vehicle_id,
		"owner_id": owner_id,
		"plate_arabic": plate_arabic,
		"plate_english": plate_english,
	}

	if not plate_detected:
		payload.update(
			{
				"reason": "plate_not_detected",
				"message": "Could not detect a plate, so registration status is unknown.",
			}
		)
		return payload

	if vehicle_id:
		payload.update(
			{
				"reason": "already_registered",
				"message": "Car is already registered by a user.",
			}
		)
		return payload

	payload.update(
		{
			"is_new_car": True,
			"notify_roles": ["admin", "security"],
			"action": "vendor_outreach_recommended",
			"reason": "car_not_registered_by_any_user",
			"message": "New car detected. Admin/security should contact a vendor to call the driver and introduce the app.",
		}
	)
	return payload


def _persist_prediction_to_supabase(
	*,
	original_image_bytes: bytes,
	response_payload: Dict[str, Any],
	request_user: Optional[Dict[str, Any]],
	parking_location: Optional[str],
	event_type: str,
	camera_source: Optional[str],
) -> Dict[str, Any]:
	default_new_car_alert = {
		"is_new_car": None,
		"notify_roles": ["admin", "security"],
		"action": "lookup_unavailable",
		"reason": "supabase_unavailable",
		"message": "Could not verify registration status because Supabase is unavailable.",
		"vehicle_id": None,
		"owner_id": None,
		"plate_arabic": None,
		"plate_english": None,
	}

	if not _supabase_configured():
		return {
			"enabled": False,
			"saved": False,
			"reason": "Supabase env vars are not configured.",
			"payment_status": "unknown",
			"pricing": _compute_session_time_and_fee(None, parking_location_override=parking_location),
			"gate_decision": _build_gate_decision(event_type=event_type, session_row=None, plate_detected=False),
			"app_registration": {
				"is_registered": False,
				"vehicle_id": None,
				"owner_id": None,
			},
			"new_car_alert": default_new_car_alert,
		}

	user_folder = request_user["id"] if request_user else "anonymous"
	now = datetime.now(timezone.utc)
	object_base = f"{user_folder}/{now:%Y/%m/%d}/{uuid4().hex}"
	raw_path = f"{object_base}_raw.jpg"
	user_split_path = f"{object_base}_user.jpg"
	admin_annotated_path = f"{object_base}_admin.jpg"
	new_car_alert_payload: Dict[str, Any] = dict(default_new_car_alert)
	uploaded_objects: List[Tuple[str, str]] = []
	event_persisted = False

	try:
		user_page_payload = response_payload.get("user_page") if isinstance(response_payload.get("user_page"), dict) else {}
		admin_page_payload = response_payload.get("admin_page") if isinstance(response_payload.get("admin_page"), dict) else {}

		split_base64 = user_page_payload.get("split_image_base64")
		car_focus_base64 = user_page_payload.get("car_focus_base64")
		admin_annotated_base64 = admin_page_payload.get("annotated_image_base64")
		plates_detected = int(response_payload.get("plates_detected_in_selected_car") or 0)
		processed_image_variant = "split"

		if plates_detected <= 0 and isinstance(car_focus_base64, str) and car_focus_base64.strip():
			user_split_bytes = base64.b64decode(car_focus_base64)
			processed_image_variant = "car_focus_fallback_no_plate"
		elif isinstance(split_base64, str) and split_base64.strip():
			user_split_bytes = base64.b64decode(split_base64)
		elif isinstance(car_focus_base64, str) and car_focus_base64.strip():
			user_split_bytes = base64.b64decode(car_focus_base64)
			processed_image_variant = "car_focus_fallback_missing_split"
		else:
			user_split_bytes = original_image_bytes
			processed_image_variant = "raw_fallback"

		if isinstance(admin_annotated_base64, str) and admin_annotated_base64.strip():
			admin_annotated_bytes = base64.b64decode(admin_annotated_base64)
		else:
			admin_annotated_bytes = user_split_bytes

		plate_info = response_payload.get("plate_info", {})
		plate_detected = _is_plate_detected(plate_info)
		request_user_actor_id = str((request_user or {}).get("id") or "").strip() or None
		request_user_id = _extract_created_by_user_id(request_user)
		vehicle_record = _resolve_vehicle_record_by_plate(
			plate_info.get("arabic", ""),
			plate_info.get("english", ""),
			request_user_id,
		)
		vehicle_id = str(vehicle_record.get("id")) if isinstance(vehicle_record, dict) and vehicle_record.get("id") else None
		owner_id = str(vehicle_record.get("owner_id")) if isinstance(vehicle_record, dict) and vehicle_record.get("owner_id") else None
		match_path = str(vehicle_record.get("_match_path") or "") if isinstance(vehicle_record, dict) else ""
		if match_path.startswith("arabic"):
			registration_match_source = "arabic_ocr"
		elif match_path.startswith("english"):
			registration_match_source = "english_ocr"
		else:
			registration_match_source = None
		LOGGER.info(
			"plate_resolution | ocr_ar='%s' ocr_en='%s' | vehicle_found=%s | vehicle_id=%s | owner_id=%s",
			plate_info.get("arabic", ""), plate_info.get("english", ""),
			vehicle_id is not None, vehicle_id, owner_id,
		)
		new_car_alert_payload = _build_new_car_alert_payload(
			plate_info=plate_info,
			vehicle_id=vehicle_id,
			owner_id=owner_id,
		)

		session_id, session_row = _resolve_session_for_event(
			event_type=event_type,
			vehicle_id=vehicle_id,
			created_by=request_user_id,
			parking_location=parking_location,
			plate_arabic=plate_info.get("arabic"),
			plate_english=plate_info.get("english"),
		)
		if event_type == "entry" and vehicle_id and session_id is None:
			LOGGER.error("SESSION_NOT_CREATED for registered vehicle %s on entry", vehicle_id)
		LOGGER.info(
			"session_resolution | event_type=%s | vehicle_id=%s | session_id=%s",
			event_type, vehicle_id, session_id,
		)
		payment_status = _derive_payment_status_from_session(session_row)
		left_within_5 = bool(session_row.get("left_within_5_minutes")) if isinstance(session_row, dict) else False
		pricing = _compute_session_time_and_fee(session_row, parking_location_override=parking_location)
		gate_decision = _build_gate_decision(
			event_type=event_type,
			session_row=session_row,
			plate_detected=plate_detected,
		)

		_upload_to_supabase_storage(SUPABASE_RAW_BUCKET, raw_path, original_image_bytes)
		uploaded_objects.append((SUPABASE_RAW_BUCKET, raw_path))
		_upload_to_supabase_storage(SUPABASE_PROCESSED_BUCKET, user_split_path, user_split_bytes)
		uploaded_objects.append((SUPABASE_PROCESSED_BUCKET, user_split_path))
		_upload_to_supabase_storage(SUPABASE_PROCESSED_BUCKET, admin_annotated_path, admin_annotated_bytes)
		uploaded_objects.append((SUPABASE_PROCESSED_BUCKET, admin_annotated_path))

		chars = plate_info.get("characters", [])
		ocr_confidence = None
		if chars:
			avg_conf = sum(float(item.get("confidence", 0.0)) for item in chars) / len(chars)
			ocr_confidence = round(avg_conf, 4)

		event_id = _insert_car_event_record(
			event_type=event_type,
			created_by=request_user_id,
			vehicle_id=vehicle_id,
			session_id=session_id,
			raw_image_path=f"{SUPABASE_RAW_BUCKET}/{raw_path}",
			user_split_image_path=f"{SUPABASE_PROCESSED_BUCKET}/{user_split_path}",
			admin_annotated_image_path=f"{SUPABASE_PROCESSED_BUCKET}/{admin_annotated_path}",
			ocr_arabic=plate_info.get("arabic", "N/A"),
			ocr_english=plate_info.get("english", "N/A"),
			ocr_raw_labels=plate_info.get("raw_ordered_labels", []),
			ocr_confidence=ocr_confidence,
			parking_location=parking_location,
			event_metadata={
				"event_type": event_type,
				"cars_detected": response_payload.get("cars_detected", 0),
				"plates_detected_in_selected_car": response_payload.get("plates_detected_in_selected_car", 0),
				"processed_image_variant": processed_image_variant,
				"selected_car": response_payload.get("selected_car"),
				"selected_plate": response_payload.get("selected_plate"),
				"filename": response_payload.get("filename"),
				"parking_location": parking_location,
				"camera_source": camera_source,
				"payment_status": payment_status,
				"left_within_5_minutes": left_within_5,
				"gate_decision": gate_decision,
				"pricing": pricing,
				"app_registration": {
					"is_registered": vehicle_id is not None,
					"vehicle_id": vehicle_id,
					"owner_id": owner_id,
					"match_source": registration_match_source,
				},
				"new_car_alert": new_car_alert_payload,
			},
		)
		if not event_id:
			raise RuntimeError("Failed to persist car event id.")
		event_persisted = True

		notification_event_ids: List[str] = []
		if owner_id and event_type in {"entry", "exit"}:
			plate_label = str(plate_info.get("arabic") or plate_info.get("english") or "your vehicle")
			timestamp_label = datetime.now(timezone.utc).isoformat()
			if event_type == "entry":
				notification_title = "Vehicle Entry Detected"
				notification_body = f"{plate_label} entered at {timestamp_label}."
			else:
				notification_title = "Vehicle Exit Detected"
				notification_body = f"{plate_label} exited at {timestamp_label}."

			notification_id = _insert_notification_event(
				user_id=owner_id,
				event_type=f"vehicle_{event_type}",
				title=notification_title,
				body=notification_body,
				data={
					"event_id": event_id,
					"session_id": session_id,
					"vehicle_id": vehicle_id,
					"parking_location": parking_location,
					"payment_status": payment_status,
				},
			)
			if notification_id:
				notification_event_ids.append(notification_id)

		if owner_id and event_type == "exit" and not bool(gate_decision.get("allowed")):
			amount_due = 0.0
			try:
				amount_due = float((pricing or {}).get("total_fee_egp") or 0.0)
			except (TypeError, ValueError):
				amount_due = 0.0

			payment_reminder_id = _insert_notification_event(
				user_id=owner_id,
				event_type="payment_reminder",
				title="Payment Required",
				body=f"You have a balance of {amount_due:.2f} EGP.",
				data={
					"event_id": event_id,
					"session_id": session_id,
					"vehicle_id": vehicle_id,
					"amount_due_egp": round(max(0.0, amount_due), 2),
					"payment_status": payment_status,
					"gate_decision": gate_decision,
				},
			)
			if payment_reminder_id:
				notification_event_ids.append(payment_reminder_id)

		return {
			"enabled": True,
			"saved": True,
			"event_type": event_type,
			"event_id": event_id,
			"session_id": session_id,
			"payment_status": payment_status,
			"left_within_5_minutes": left_within_5,
			"pricing": pricing,
			"gate_decision": gate_decision,
			"processed_image_variant": processed_image_variant,
			"app_registration": {
				"is_registered": vehicle_id is not None,
				"vehicle_id": vehicle_id,
				"owner_id": owner_id,
				"matched_to_request_user": bool(request_user_actor_id and owner_id and request_user_actor_id == owner_id),
				"match_source": registration_match_source,
			},
			"new_car_alert": new_car_alert_payload,
			"notification_event_ids": notification_event_ids,
			"parking_location": parking_location,
			"raw_image_path": f"{SUPABASE_RAW_BUCKET}/{raw_path}",
			"user_split_image_path": f"{SUPABASE_PROCESSED_BUCKET}/{user_split_path}",
			"processed_image_path": f"{SUPABASE_PROCESSED_BUCKET}/{user_split_path}",
			"admin_annotated_image_path": f"{SUPABASE_PROCESSED_BUCKET}/{admin_annotated_path}",
		}
	except Exception as exc:  # pragma: no cover
		LOGGER.exception("Supabase persistence failed")
		if uploaded_objects and not event_persisted:
			for bucket, object_path in reversed(uploaded_objects):
				try:
					_delete_from_supabase_storage(bucket, object_path)
				except Exception as cleanup_exc:  # pragma: no cover
					LOGGER.warning(
						"Supabase cleanup failed for %s/%s: %s",
						bucket,
						object_path,
						cleanup_exc,
					)
		return {
			"enabled": True,
			"saved": False,
			"error": str(exc),
			"payment_status": "unknown",
			"pricing": _compute_session_time_and_fee(None, parking_location_override=parking_location),
			"gate_decision": _build_gate_decision(event_type=event_type, session_row=None, plate_detected=False),
			"app_registration": {
				"is_registered": False,
				"vehicle_id": None,
				"owner_id": None,
			},
			"new_car_alert": new_car_alert_payload,
			"notification_event_ids": [],
		}


app = FastAPI(title="Ain El Aql Local Plate Recognition API", version="2.0.0")
app.add_middleware(
    CORSMiddleware,
	allow_origins=CORS_ALLOW_ORIGINS or ["*"],
	allow_credentials=False,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def rewrite_supabase_auth_header(request: Request, call_next):
    x_supa = request.headers.get("x-supabase-auth")
    if x_supa:
        new_headers = []
        for name, value in request.scope["headers"]:
            if name == b"authorization":
                continue
            new_headers.append((name, value))
        new_headers.append((b"authorization", x_supa.encode("latin-1")))
        request.scope["headers"] = new_headers
    return await call_next(request)



@app.get("/")
def root() -> Dict[str, Any]:
	return {
		"status": "ok",
		"service": "ain-el-aql-backend",
		"model_inference_provider": MODEL_INFERENCE_PROVIDER,
		"docs_url": "/docs",
		"health_url": "/health",
		"supabase_health_url": "/supabase/health",
	}


_model_lock = Lock()


# ---------------------------------------------------------------------------
# Character map for OCR label → (Arabic, English) translation
# Derived from the arabic_mapping in app.py ΓÇö used by _decode_ocr_result()
# ---------------------------------------------------------------------------
CHAR_MAP = {
	# Letters
	"aain": ("╪╣", "E"),
	"alef": ("أ", "A"),
	"a":    ("أ", "A"),
	"alf":  ("أ", "A"),
	"baa":  ("ب", "B"),
	"daal": ("د", "D"),
	"dal":  ("د", "D"),
	"e":    ("╪╣", "E"),
	"faa":  ("ف", "F"),
	"geem": ("ج", "G"),
	"haa":  ("هـ", "H"),
	"kaaf": ("ك", "K"),
	"laam": ("ل", "L"),
	"meem": ("م", "M"),
	"noon": ("ن", "N"),
	"qaf":  ("ق", "Q"),
	"raa":  ("╪▒", "R"),
	"sad":  ("╪╡", "S"),
	"seen": ("╪│", "C"),
	"taa":  ("╪╖", "T"),
	"waaw": ("و", "W"),
	"waw":  ("و", "W"),
	"yaa":  ("ى", "Y"),
	"zay":  ("╪▓", "Z"),
	"dad":  ("╪╢", "DD"),
	# Digits
	"0": ("0", "0"), "1": ("1", "1"), "2": ("2", "2"), "3": ("3", "3"), "4": ("4", "4"),
	"5": ("5", "5"), "6": ("6", "6"), "7": ("7", "7"), "8": ("8", "8"), "9": ("9", "9"),
	"٠": ("0", "0"), "١": ("1", "1"), "٢": ("2", "2"), "٣": ("3", "3"), "٤": ("4", "4"),
	"٥": ("5", "5"), "٦": ("6", "6"), "٧": ("7", "7"), "٨": ("8", "8"), "٩": ("9", "9"),
}


def _clamp_bbox(x1: float, y1: float, x2: float, y2: float, width: int, height: int) -> Tuple[int, int, int, int]:
    left = max(0, min(int(x1), width - 1))
    top = max(0, min(int(y1), height - 1))
    right = max(1, min(int(x2), width))
    bottom = max(1, min(int(y2), height))

    if right <= left:
        right = min(width, left + 1)
    if bottom <= top:
        bottom = min(height, top + 1)

    return left, top, right, bottom


def _encode_image_base64(image_bgr: np.ndarray) -> str:
    ok, encoded = cv2.imencode(".jpg", image_bgr)
    if not ok:
        raise HTTPException(status_code=500, detail="Failed to encode image.")
    return base64.b64encode(encoded.tobytes()).decode("utf-8")


def _decode_image_bytes(image_bytes: bytes) -> np.ndarray:
	image_np = np.frombuffer(image_bytes, dtype=np.uint8)
	image_bgr = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
	if image_bgr is None:
		raise HTTPException(status_code=400, detail="Unable to decode image.")
	return image_bgr


def _fit_into_canvas(image_bgr: np.ndarray, target_w: int, target_h: int) -> np.ndarray:
    canvas = np.full((target_h, target_w, 3), 18, dtype=np.uint8)
    if image_bgr.size == 0:
        return canvas

    src_h, src_w = image_bgr.shape[:2]
    scale = min(target_w / max(1, src_w), target_h / max(1, src_h))
    new_w = max(1, int(src_w * scale))
    new_h = max(1, int(src_h * scale))

    resized = cv2.resize(image_bgr, (new_w, new_h), interpolation=cv2.INTER_AREA)
    x_off = (target_w - new_w) // 2
    y_off = (target_h - new_h) // 2
    canvas[y_off : y_off + new_h, x_off : x_off + new_w] = resized
    return canvas


def _compose_user_split_image(plate_focus_bgr: np.ndarray, car_focus_bgr: np.ndarray) -> np.ndarray:
    half_h = max(plate_focus_bgr.shape[0], car_focus_bgr.shape[0], 220)
    half_w = max(plate_focus_bgr.shape[1], car_focus_bgr.shape[1], 320)

    left_half = _fit_into_canvas(plate_focus_bgr, half_w, half_h)
    right_half = _fit_into_canvas(car_focus_bgr, half_w, half_h)
    split_image = np.concatenate([left_half, right_half], axis=1)
    return split_image


def _build_plate_placeholder(reference_bgr: np.ndarray) -> np.ndarray:
	placeholder = np.full(reference_bgr.shape, 18, dtype=np.uint8)
	h, w = placeholder.shape[:2]
	text = "NO PLATE DETECTED"
	font_scale = 0.8 if w >= 500 else 0.6
	thickness = 2
	text_size, _ = cv2.getTextSize(text, cv2.FONT_HERSHEY_SIMPLEX, font_scale, thickness)
	text_x = max(10, (w - text_size[0]) // 2)
	text_y = max(28, h // 2)
	cv2.putText(
		placeholder,
		text,
		(text_x, text_y),
		cv2.FONT_HERSHEY_SIMPLEX,
		font_scale,
		(0, 220, 220),
		thickness,
		cv2.LINE_AA,
	)
	return placeholder


def _decode_ocr_result(result: Any) -> Dict[str, Any]:
    if result.boxes is None or len(result.boxes) == 0:
        return {
            "raw_ordered_labels": [],
            "characters": [],
            "arabic": "N/A",
            "english": "N/A",
        }

    names = result.names if hasattr(result, "names") else {}
    detections: List[Dict[str, Any]] = []

    for box in result.boxes:
        cls_idx = int(box.cls[0].item()) if box.cls is not None else -1
        raw_label = names.get(cls_idx, str(cls_idx)) if isinstance(names, dict) else str(cls_idx)
        norm_label = str(raw_label).strip().lower()
        xyxy = box.xyxy[0].tolist()
        confidence = float(box.conf[0].item()) if box.conf is not None else 0.0
        center_x = (xyxy[0] + xyxy[2]) / 2.0
        ar_char, en_char = CHAR_MAP.get(norm_label, (str(raw_label), str(raw_label)))

        detections.append(
            {
                "label": str(raw_label),
                "normalized_label": norm_label,
                "arabic": ar_char,
                "english": en_char,
                "is_digit": norm_label.isdigit(),
                "confidence": round(confidence, 4),
                "bbox": [int(xyxy[0]), int(xyxy[1]), int(xyxy[2]), int(xyxy[3])],
                "center_x": center_x,
            }
        )

    # Egyptian plate layout:
    # - Letters are read right-to-left (Arabic reading direction) → sort RTL
    # - Numbers are always written left-to-right                  → sort LTR
    # Splitting before sorting ensures each group gets the correct direction.

    letter_detections = [d for d in detections if not d["is_digit"]]
    number_detections = [d for d in detections if d["is_digit"]]

    letter_detections.sort(key=lambda item: item["center_x"], reverse=True)  # RTL
    number_detections.sort(key=lambda item: item["center_x"], reverse=False)  # LTR

    ar_letters: List[str] = [d["arabic"] for d in letter_detections]
    ar_numbers: List[str] = [d["arabic"] for d in number_detections]
    en_letters: List[str] = [d["english"] for d in letter_detections]
    en_numbers: List[str] = [d["english"] for d in number_detections]

    if ar_letters or ar_numbers:
        arabic_text = f"{' '.join(ar_letters)} | {' '.join(ar_numbers)}"
    else:
        arabic_text = "N/A"

    if en_letters or en_numbers:
        english_text = f"{' '.join(en_letters)} | {''.join(en_numbers)}"
    else:
        english_text = "N/A"

    clean_chars = [
        {
            "label": item["label"],
            "arabic": item["arabic"],
            "english": item["english"],
            "confidence": item["confidence"],
            "bbox": item["bbox"],
        }
        for item in detections
    ]

    return {
        "raw_ordered_labels": [item["label"] for item in detections],
        "characters": clean_chars,
        "arabic": arabic_text,
        "english": english_text,
    }











def _run_pipeline_remote(*, image_bytes: bytes, filename: Optional[str], content_type: Optional[str]) -> Dict[str, Any]:
	try:
		from gradio_client import Client, handle_file
	except ImportError as exc:
		raise HTTPException(status_code=500, detail="gradio_client is not installed on the backend. Please run 'pip install gradio_client'.") from exc

	import tempfile
	import base64
	import os

	# Resolve which HF Space to call from env (default to the V2 space with new models)
	hf_space_id = os.getenv("HF_SPACE_ID", "Pant0x/EALPR_OCR_V2")

	# Save the uploaded bytes to a temp file for gradio_client
	with tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") as tmp:
		tmp.write(image_bytes)
		tmp_path = tmp.name

	try:
		client = Client(hf_space_id)
		result = client.predict(
			img=handle_file(tmp_path),
			api_name="/predict_plate",
		)

		# The HF Gradio app returns a tuple: (annotated_filepath, numbers_ar, letters_ar)
		if not isinstance(result, tuple) or len(result) < 3:
			raise HTTPException(status_code=502, detail=f"Unexpected response from Hugging Face: {result}")

		annotated_img_path = result[0]
		numbers_ar = str(result[1] or "").strip()
		letters_ar = str(result[2] or "").strip()

		arabic_text = f"{letters_ar} | {numbers_ar}" if (letters_ar or numbers_ar) else "N/A"

		# Build english text from the arabic chars using CHAR_MAP reverse lookup
		_ar_to_en = {v[0]: v[1] for v in CHAR_MAP.values()}
		letters_en_parts = [_ar_to_en.get(ch, ch) for ch in letters_ar.split(" ") if ch]
		numbers_en_parts = [_ar_to_en.get(ch, ch) for ch in numbers_ar.split(" ") if ch]
		letters_en = " ".join(letters_en_parts)
		numbers_en = "".join(numbers_en_parts)
		english_text = f"{letters_en} | {numbers_en}" if (letters_en or numbers_en) else "N/A"

		# Read annotated image back to base64 so Flutter/Admin can display it
		annotated_b64 = ""
		if annotated_img_path and os.path.exists(str(annotated_img_path)):
			with open(str(annotated_img_path), "rb") as f:
				annotated_b64 = base64.b64encode(f.read()).decode("utf-8")

		return {
			"plate_info": {
				"arabic": arabic_text,
				"english": english_text,
				"characters": [],
				"raw_ordered_labels": [],
			},
			"user_page": {},
			"admin_page": {
				"annotated_image_base64": annotated_b64,
			},
		}
	except HTTPException:
		raise
	except Exception as exc:
		LOGGER.exception("Remote model service failed")
		raise HTTPException(status_code=502, detail=f"Remote model service failed: {exc}") from exc
	finally:
		if os.path.exists(tmp_path):
			try:
				os.remove(tmp_path)
			except OSError:
				pass


@app.post("/auth/register")
def auth_register(
	payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	_ensure_auth_profile_schema_ready()

	username = _normalize_spaces(str(payload.get("username") or ""))
	phone_number = _normalize_spaces(str(payload.get("phone") or payload.get("phone_number") or ""))
	email = _normalize_spaces(str(payload.get("email") or "")).lower()
	national_id = _normalize_spaces(str(payload.get("nid") or payload.get("national_id") or ""))
	password = str(payload.get("password") or "").strip()
	first_name, last_name, full_name = _resolve_profile_name_fields(payload, fallback_first_name=username)
	if not first_name:
		first_name = username
	full_name = _compose_full_name(first_name, last_name) or username

	if not username:
		raise HTTPException(status_code=400, detail="username is required.")
	if not USERNAME_PATTERN.fullmatch(username):
		raise HTTPException(
			status_code=400,
			detail="username must be 3-32 chars and only contain letters, numbers, underscore, dot, or dash.",
		)

	if not phone_number:
		raise HTTPException(status_code=400, detail="phone is required.")
	if not PHONE_PATTERN.fullmatch(phone_number):
		raise HTTPException(status_code=400, detail="phone must be 7-15 digits and may start with +.")

	if email and "@" not in email:
		raise HTTPException(status_code=400, detail="email format is invalid.")

	if national_id and not NATIONAL_ID_PATTERN.fullmatch(national_id):
		raise HTTPException(
			status_code=400,
			detail="nid format is invalid. Use 6-32 chars with letters, numbers, or dash.",
		)

	if len(password) < 6:
		raise HTTPException(status_code=400, detail="Password must be at least 6 characters.")

	registration_email = email or _build_internal_login_email(username, phone_number)
	existing_auth_user_id: Optional[str] = None
	if email:
		existing_auth_user = _find_supabase_auth_user_by_email(registration_email)
		existing_auth_user_id = str((existing_auth_user or {}).get("id") or "").strip() or None

	def _has_profile_conflict(profile_row: Optional[Dict[str, Any]]) -> bool:
		if not isinstance(profile_row, dict):
			return False

		profile_user_id = str(profile_row.get("id") or "").strip()
		if not profile_user_id:
			return True

		if existing_auth_user_id and profile_user_id == existing_auth_user_id:
			return False

		return True

	if _has_profile_conflict(_supabase_get_profile_by_field("username", username)):
		raise HTTPException(status_code=409, detail="username is already in use.")
	if _has_profile_conflict(_supabase_get_profile_by_field("phone_number", phone_number)):
		raise HTTPException(status_code=409, detail="phone is already in use.")
	if national_id and _has_profile_conflict(_supabase_get_profile_by_field("national_id", national_id)):
		raise HTTPException(status_code=409, detail="nid is already in use.")
	if email and _has_profile_conflict(_supabase_get_profile_by_field("email", email, case_insensitive=True)):
		raise HTTPException(status_code=409, detail="email is already in use.")

	vehicle_payload = _extract_registration_vehicle_payload(payload)
	auth_user_metadata = {
		"first_name": first_name,
		"last_name": last_name,
		"full_name": full_name,
		"username": username,
		"phone_number": phone_number,
		"national_id": national_id,
	}
	auth_payload = {
		"email": registration_email,
		"password": password,
		"email_confirm": True,
		"user_metadata": auth_user_metadata,
	}

	user_id: Optional[str] = None
	if existing_auth_user_id:
		try:
			_update_supabase_auth_user(
				existing_auth_user_id,
				{
					"email": registration_email,
					"password": password,
					"email_confirm": True,
					"user_metadata": auth_user_metadata,
				},
			)
		except RuntimeError as exc:
			raise HTTPException(
				status_code=502,
				detail=f"Register failed while finalizing existing auth user: {exc}",
			) from exc

		user_id = existing_auth_user_id
	else:
		try:
			response = requests.post(
				f"{SUPABASE_URL}/auth/v1/admin/users",
				json=auth_payload,
				headers={
					"apikey": SUPABASE_SERVICE_ROLE_KEY,
					"Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}",
					"Content-Type": "application/json",
				},
				timeout=SUPABASE_TIMEOUT_SECONDS,
			)
		except requests.RequestException as exc:
			raise HTTPException(status_code=503, detail=f"Supabase register failed: {exc}") from exc

		if response.status_code in {200, 201}:
			try:
				auth_user = response.json() if response.text else {}
			except ValueError:
				auth_user = {}
			if not isinstance(auth_user, dict):
				auth_user = {}
			user_id = str(auth_user.get("id") or auth_user.get("user", {}).get("id") or "").strip()
		else:
			error_msg = _extract_response_error_message(response)
			if registration_email and _is_supabase_duplicate_auth_user_error(error_msg):
				existing_auth_user = _find_supabase_auth_user_by_email(registration_email)
				existing_user_id = str((existing_auth_user or {}).get("id") or "").strip()
				if not existing_user_id:
					raise HTTPException(status_code=409, detail="Register failed: email is already in use.")

				try:
					_update_supabase_auth_user(
						existing_user_id,
						{
							"email": registration_email,
							"password": password,
							"email_confirm": True,
							"user_metadata": auth_user_metadata,
						},
					)
				except RuntimeError as exc:
					raise HTTPException(
						status_code=502,
						detail=f"Register failed while finalizing existing auth user: {exc}",
					) from exc

				user_id = existing_user_id
			else:
				raise HTTPException(status_code=response.status_code, detail=f"Register failed: {error_msg}")

	if not user_id:
		raise HTTPException(status_code=502, detail="Supabase did not return a user id.")

	try:
		profile = _upsert_profile_after_register(
			user_id=user_id,
			first_name=first_name,
			last_name=last_name,
			username=username,
			phone_number=phone_number,
			email=registration_email,
			national_id=national_id or None,
			role=str(payload.get("role") or "user"),
		)
		created_vehicle = _create_vehicle_for_owner(owner_id=user_id, vehicle_payload=vehicle_payload) if vehicle_payload else None
	except HTTPException:
		raise
	except Exception as exc:
		raise HTTPException(status_code=502, detail=f"Failed to finalize registration profile: {exc}") from exc

	session_data, login_error = _supabase_password_login(password=password, email=registration_email)
	if not isinstance(session_data, dict):
		raise HTTPException(status_code=502, detail=f"Registration succeeded but login failed: {login_error}")

	resolved_profile = profile if isinstance(profile, dict) else _supabase_get_profile_by_user_id(user_id)
	role = str((resolved_profile or {}).get("role") or "user")
	resolved_first_name, resolved_last_name, resolved_full_name = _resolve_profile_name_fields(
		resolved_profile if isinstance(resolved_profile, dict) else None,
		fallback_full_name=full_name,
		fallback_first_name=first_name,
		fallback_last_name=last_name,
	)

	return {
		"status": "ok",
		"user": {
			"id": user_id,
			"first_name": resolved_first_name,
			"last_name": resolved_last_name,
			"full_name": resolved_full_name,
			"username": (resolved_profile or {}).get("username") or username,
			"phone_number": (resolved_profile or {}).get("phone_number") or phone_number,
			"email": (resolved_profile or {}).get("email") or registration_email,
			"national_id": (resolved_profile or {}).get("national_id"),
			"role": role,
			"is_staff": _is_staff_role(role),
		},
		"session": {
			"access_token": session_data.get("access_token"),
			"refresh_token": session_data.get("refresh_token"),
			"expires_in": session_data.get("expires_in"),
		},
		"vehicle": created_vehicle,
	}


@app.post("/auth/login")
def auth_login(
	payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	_ensure_auth_profile_schema_ready()

	identifier = _normalize_spaces(
		str(
			payload.get("identifier")
			or payload.get("email")
			or payload.get("phone")
			or payload.get("phone_number")
			or payload.get("username")
			or payload.get("nid")
			or payload.get("national_id")
			or ""
		)
	)
	password = str(payload.get("password") or "").strip()

	if not identifier or not password:
		raise HTTPException(status_code=400, detail="identifier and password are required.")

	profile = _supabase_find_profile_by_identifier(identifier)

	login_candidates: List[Tuple[Optional[str], Optional[str]]] = []
	if isinstance(profile, dict):
		profile_email = _normalize_spaces(str(profile.get("email") or "")).lower()
		profile_phone = _normalize_spaces(str(profile.get("phone_number") or ""))
		if profile_email:
			login_candidates.append((profile_email, None))
		if profile_phone:
			login_candidates.append((None, profile_phone))

	if "@" in identifier:
		login_candidates.append((_normalize_spaces(identifier).lower(), None))
	if PHONE_PATTERN.fullmatch(identifier):
		login_candidates.append((None, identifier))

	if not login_candidates:
		raise HTTPException(status_code=401, detail="Login failed: account was not found.")

	data: Optional[Dict[str, Any]] = None
	last_error = "Invalid login credentials"
	seen_candidates: set = set()
	for candidate_email, candidate_phone in login_candidates:
		candidate_key = (candidate_email or "", candidate_phone or "")
		if candidate_key in seen_candidates:
			continue
		seen_candidates.add(candidate_key)

		login_result, login_error = _supabase_password_login(
			password=password,
			email=candidate_email,
			phone_number=candidate_phone,
		)
		if isinstance(login_result, dict):
			data = login_result
			break
		if login_error:
			last_error = login_error

	if not isinstance(data, dict):
		raise HTTPException(status_code=401, detail=f"Login failed: {last_error}")

	user_id = str(data.get("user", {}).get("id") or "").strip()
	resolved_profile = _supabase_get_profile_by_user_id(user_id) if user_id else profile
	role = str((resolved_profile or {}).get("role") or "user")
	resolved_first_name, resolved_last_name, resolved_full_name = _resolve_profile_name_fields(
		resolved_profile if isinstance(resolved_profile, dict) else None,
		fallback_full_name=(resolved_profile or {}).get("full_name") if isinstance(resolved_profile, dict) else None,
	)
	return {
		"status": "ok",
		"user": {
			"id": user_id,
			"first_name": resolved_first_name,
			"last_name": resolved_last_name,
			"full_name": resolved_full_name,
			"username": (resolved_profile or {}).get("username"),
			"phone_number": (resolved_profile or {}).get("phone_number"),
			"email": (resolved_profile or {}).get("email") or data.get("user", {}).get("email"),
			"national_id": (resolved_profile or {}).get("national_id"),
			"role": role,
			"is_staff": _is_staff_role(role),
		},
		"session": {
			"access_token": data.get("access_token"),
			"refresh_token": data.get("refresh_token"),
			"expires_in": data.get("expires_in"),
		},
	}


@app.get("/auth/reset-password-page", response_class=HTMLResponse)
def auth_reset_password_page() -> HTMLResponse:
	html = """<!doctype html>
<html lang=\"en\">
<head>
	<meta charset=\"utf-8\" />
	<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\" />
	<title>Reset Password</title>
	<style>
		body { font-family: Arial, sans-serif; margin: 0; background: #f5f7fb; color: #0f172a; }
		.wrap { max-width: 420px; margin: 40px auto; background: #ffffff; border-radius: 12px; padding: 20px; box-shadow: 0 10px 30px rgba(0,0,0,0.08); }
		h1 { font-size: 20px; margin: 0 0 14px; }
		label { display: block; margin: 12px 0 6px; font-size: 14px; }
		input { width: 100%; padding: 10px 12px; border: 1px solid #cbd5e1; border-radius: 8px; font-size: 14px; box-sizing: border-box; }
		button { width: 100%; margin-top: 16px; background: #111827; color: #fff; border: 0; border-radius: 8px; padding: 10px 14px; font-size: 14px; cursor: pointer; }
		button:disabled { opacity: 0.6; cursor: not-allowed; }
		.hidden { display: none; }
		.msg { margin-top: 12px; font-size: 14px; }
		.ok { color: #166534; }
		.err { color: #991b1b; }
	</style>
</head>
<body>
	<div class=\"wrap\">
		<div id=\"resetFormView\">
			<h1>Reset your password</h1>
			<p id=\"tokenState\" class=\"msg\"></p>
			<label for=\"password\">New password</label>
			<input id=\"password\" type=\"password\" minlength=\"6\" placeholder=\"At least 6 characters\" />
			<label for=\"confirm\">Confirm password</label>
			<input id=\"confirm\" type=\"password\" minlength=\"6\" placeholder=\"Repeat password\" />
			<button id=\"submitBtn\">Reset password</button>
			<p id=\"result\" class=\"msg\"></p>
		</div>
		<div id=\"successView\" class=\"hidden\">
			<h1>Password updated</h1>
			<p class=\"msg ok\">Your password has been changed.</p>
		</div>
	</div>
	<script>
		function getTokenMap() {
			const hash = new URLSearchParams((window.location.hash || '').replace(/^#/, ''));
			const query = new URLSearchParams(window.location.search || '');
			return {
				accessToken: hash.get('access_token') || query.get('access_token') || '',
				tokenHash: hash.get('token_hash') || query.get('token_hash') || '',
				code: hash.get('code') || query.get('code') || '',
				type: hash.get('type') || query.get('type') || 'recovery',
				email: hash.get('email') || query.get('email') || ''
			};
		}

		let token = '';
		const tokenState = document.getElementById('tokenState');
		const result = document.getElementById('result');
		const submitBtn = document.getElementById('submitBtn');
		const resetFormView = document.getElementById('resetFormView');
		const successView = document.getElementById('successView');
		submitBtn.disabled = true;

		function showSuccessRedirectScreen() {
			resetFormView.classList.add('hidden');
			successView.classList.remove('hidden');
		}

		async function resolveAccessToken() {
			const tokenMap = getTokenMap();
			if (tokenMap.accessToken) {
				token = tokenMap.accessToken;
				tokenState.textContent = 'Reset token detected. Enter your new password.';
				tokenState.className = 'msg ok';
				submitBtn.disabled = false;
				return;
			}

			if (!tokenMap.tokenHash && !tokenMap.code) {
				tokenState.textContent = 'This reset link is invalid or expired.';
				tokenState.className = 'msg err';
				return;
			}

			tokenState.textContent = 'Verifying reset link...';
			tokenState.className = 'msg';

			try {
				const response = await fetch('/auth/reset-password/resolve-token', {
					method: 'POST',
					headers: { 'Content-Type': 'application/json' },
					body: JSON.stringify({
						token_hash: tokenMap.tokenHash,
						code: tokenMap.code,
						type: tokenMap.type,
						email: tokenMap.email
					})
				});

				const data = await response.json().catch(function () { return {}; });
				if (!response.ok) {
					throw new Error((data && data.detail) ? data.detail : 'Reset link could not be verified.');
				}

				token = (data && data.session && data.session.access_token) ? data.session.access_token : '';
				if (!token) {
					throw new Error('Reset link could not be verified.');
				}

				tokenState.textContent = 'Reset token detected. Enter your new password.';
				tokenState.className = 'msg ok';
				submitBtn.disabled = false;
			} catch (error) {
				tokenState.textContent = (error && error.message) ? error.message : 'This reset link is invalid or expired.';
				tokenState.className = 'msg err';
			}
		}

		resolveAccessToken();

		submitBtn.addEventListener('click', async function () {
			if (!token) {
				result.textContent = 'This reset link is invalid or expired.';
				result.className = 'msg err';
				return;
			}

			const password = document.getElementById('password').value || '';
			const confirm = document.getElementById('confirm').value || '';

			if (password.length < 6) {
				result.textContent = 'Password must be at least 6 characters.';
				result.className = 'msg err';
				return;
			}
			if (password !== confirm) {
				result.textContent = 'Passwords do not match.';
				result.className = 'msg err';
				return;
			}

			submitBtn.disabled = true;
			result.textContent = 'Updating password...';
			result.className = 'msg';

			try {
				const response = await fetch('/auth/reset-password', {
					method: 'POST',
					headers: { 'Content-Type': 'application/json' },
					body: JSON.stringify({ access_token: token, new_password: password })
				});

				const data = await response.json().catch(function () { return {}; });
				if (!response.ok) {
					throw new Error((data && data.detail) ? data.detail : 'Reset failed.');
				}

				result.textContent = '';
				result.className = 'msg';
				showSuccessRedirectScreen();
			} catch (error) {
				result.textContent = (error && error.message) ? error.message : 'Reset failed.';
				result.className = 'msg err';
			} finally {
				submitBtn.disabled = false;
			}
		});
	</script>
</body>
</html>
"""
	return HTMLResponse(content=html)


@app.get("/auth/reset-password-bridge", response_class=HTMLResponse)
def auth_reset_password_bridge() -> HTMLResponse:
	deep_link_base = _resolve_password_reset_deep_link_url()
	html = f"""<!doctype html>
<html lang=\"en\">
<head>
	<meta charset=\"utf-8\" />
	<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\" />
	<title>Open Ain El Aql App</title>
	<style>
		body {{ font-family: Arial, sans-serif; margin: 0; background: #f4f7fb; color: #0f172a; }}
		.wrap {{ max-width: 480px; margin: 40px auto; background: #fff; border-radius: 12px; padding: 22px; box-shadow: 0 10px 24px rgba(0,0,0,0.08); }}
		h1 {{ font-size: 22px; margin: 0 0 12px; }}
		p {{ font-size: 14px; line-height: 1.5; }}
		.actions {{ display: flex; flex-direction: column; gap: 10px; margin-top: 16px; }}
		a.btn {{ display: inline-block; text-decoration: none; text-align: center; border-radius: 8px; padding: 11px 14px; font-size: 14px; }}
		a.primary {{ background: #111827; color: #fff; }}
		a.secondary {{ background: #e2e8f0; color: #0f172a; }}
		.hint {{ margin-top: 14px; color: #475569; }}
	</style>
</head>
<body>
	<div class=\"wrap\">
		<h1>Open the app to reset password</h1>
		<p id=\"status\">Preparing your secure reset link...</p>
		<div class=\"actions\">
			<a id=\"openApp\" class=\"btn primary\" href=\"#\">Open Ain El Aql App</a>
			<a id=\"openWeb\" class=\"btn secondary\" href=\"/auth/reset-password-page\">Continue in browser</a>
		</div>
		<p class=\"hint\">If the app does not open automatically, tap \"Open Ain El Aql App\".</p>
	</div>
	<script>
		const deepLinkBase = {json.dumps(deep_link_base)};

		function getTokenMap() {{
			const hash = new URLSearchParams((window.location.hash || '').replace(/^#/, ''));
			const query = new URLSearchParams(window.location.search || '');
			const accessToken = hash.get('access_token') || query.get('access_token') || '';
			const refreshToken = hash.get('refresh_token') || query.get('refresh_token') || '';
			const tokenType = hash.get('token_type') || query.get('token_type') || '';
			const tokenHash = hash.get('token_hash') || query.get('token_hash') || '';
			const code = hash.get('code') || query.get('code') || '';
			const type = hash.get('type') || query.get('type') || '';
			const email = hash.get('email') || query.get('email') || '';
			return {{ accessToken, refreshToken, tokenType, tokenHash, code, type, email }};
		}}

		function appendTokens(baseUrl, tokenMap) {{
			const params = new URLSearchParams();
			if (tokenMap.accessToken) params.set('access_token', tokenMap.accessToken);
			if (tokenMap.refreshToken) params.set('refresh_token', tokenMap.refreshToken);
			if (tokenMap.tokenType) params.set('token_type', tokenMap.tokenType);
			if (tokenMap.tokenHash) params.set('token_hash', tokenMap.tokenHash);
			if (tokenMap.code) params.set('code', tokenMap.code);
			if (tokenMap.type) params.set('type', tokenMap.type);
			if (tokenMap.email) params.set('email', tokenMap.email);

			const query = params.toString();
			if (!query) return baseUrl;

			const separator = baseUrl.includes('?') ? '&' : '?';
			return `${{baseUrl}}${{separator}}${{query}}`;
		}}

		const statusEl = document.getElementById('status');
		const openAppEl = document.getElementById('openApp');
		const openWebEl = document.getElementById('openWeb');
		const tokenMap = getTokenMap();

		const deepLinkUrl = appendTokens(deepLinkBase, tokenMap);
		const fallbackUrl = appendTokens(`${{window.location.origin}}/auth/reset-password-page`, tokenMap);

		openAppEl.href = deepLinkUrl;
		openWebEl.href = fallbackUrl;

		if (!tokenMap.accessToken && !tokenMap.tokenHash && !tokenMap.code) {{
			statusEl.textContent = 'No reset token found in this link. Please request a new reset email.';
		}} else {{
			statusEl.textContent = 'Attempting to open the app now...';
			setTimeout(function() {{
				window.location.href = deepLinkUrl;
			}}, 400);
		}}
	</script>
</body>
</html>
"""
	return HTMLResponse(content=html)


@app.post("/auth/forgot-password")
def auth_forgot_password(
	request: Request,
	payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	identifier = _normalize_spaces(
		str(
			payload.get("identifier")
			or payload.get("email")
			or payload.get("phone")
			or payload.get("phone_number")
			or payload.get("username")
			or payload.get("nid")
			or payload.get("national_id")
			or ""
		)
	)
	if not identifier:
		raise HTTPException(status_code=400, detail="identifier is required.")

	profile = _supabase_find_profile_by_identifier(identifier)
	resolved_email = _normalize_spaces(str((profile or {}).get("email") or "")).lower()
	resolved_phone = _normalize_spaces(str((profile or {}).get("phone_number") or ""))
	if not resolved_email and "@" in identifier:
		resolved_email = _normalize_spaces(identifier).lower()

	has_real_email = bool(resolved_email and "@" in resolved_email and not _is_internal_login_email(resolved_email))

	# Do not leak whether account exists.
	if not has_real_email and not resolved_phone:
		return {
			"status": "ok",
			"message": "If the account exists, reset instructions were sent.",
		}

	if has_real_email:
		request_payload: Dict[str, Any] = {"email": resolved_email}
		payload_redirect_to = _normalize_spaces(str(payload.get("redirect_to") or ""))
		redirect_to = payload_redirect_to or _resolve_default_password_reset_redirect_url(request)

		request_host = _normalize_spaces(urlparse(_resolve_public_base_url(request)).hostname or "").lower()
		allowed_hosts = [request_host] if request_host else None
		if not _is_valid_redirect_url(redirect_to, allowed_http_hosts=allowed_hosts):
			raise HTTPException(status_code=400, detail="redirect_to must be an allow-listed URL or deep-link.")
		request_payload["redirect_to"] = redirect_to

		try:
			response = requests.post(
				f"{SUPABASE_URL}/auth/v1/recover",
				json=request_payload,
				headers={
					"apikey": SUPABASE_ANON_KEY,
					"Content-Type": "application/json",
				},
				timeout=SUPABASE_TIMEOUT_SECONDS,
			)
		except requests.RequestException as exc:
			raise HTTPException(status_code=503, detail=f"Supabase forgot-password failed: {exc}") from exc

		if response.status_code not in {200, 201, 204}:
			error_data = response.json() if response.text else {}
			error_msg = ""
			if isinstance(error_data, dict):
				error_msg = str(error_data.get("message") or error_data.get("error_description") or "").strip()
			if not error_msg:
				error_msg = response.text[:220]
			LOGGER.warning("Forgot-password provider response %s: %s", response.status_code, error_msg)
	else:
		try:
			_request_supabase_otp(
				channel="phone",
				identifier=resolved_phone,
				create_user=False,
				redirect_to=None,
				purpose="password_reset",
			)
		except HTTPException as exc:
			LOGGER.warning("Forgot-password phone OTP response %s: %s", exc.status_code, exc.detail)

	return {
		"status": "ok",
		"message": "If the account exists, reset instructions were sent.",
	}


@app.post("/auth/reset-password/resolve-token")
def auth_reset_password_resolve_token(
	payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	direct_access_token = _normalize_spaces(str(payload.get("access_token") or payload.get("token") or ""))
	if direct_access_token:
		return {
			"status": "ok",
			"session": {
				"access_token": direct_access_token,
				"refresh_token": _normalize_spaces(str(payload.get("refresh_token") or "")),
				"expires_in": payload.get("expires_in"),
			},
		}

	verify_type = _normalize_spaces(str(payload.get("type") or "recovery")).lower() or "recovery"
	resolved_email = _normalize_spaces(str(payload.get("email") or "")).lower()

	token_candidates: List[str] = []
	token_hash = _normalize_spaces(str(payload.get("token_hash") or payload.get("tokenHash") or ""))
	if token_hash:
		token_candidates.append(token_hash)

	code = _normalize_spaces(str(payload.get("code") or ""))
	if code and code not in token_candidates:
		token_candidates.append(code)

	if not token_candidates:
		raise HTTPException(status_code=400, detail="token_hash or code is required when access_token is missing.")

	last_error = ""
	for candidate in token_candidates:
		verify_payload: Dict[str, Any] = {
			"type": verify_type,
			"token_hash": candidate,
		}
		if resolved_email:
			verify_payload["email"] = resolved_email

		try:
			response = requests.post(
				f"{SUPABASE_URL}/auth/v1/verify",
				json=verify_payload,
				headers={
					"apikey": SUPABASE_ANON_KEY,
					"Content-Type": "application/json",
				},
				timeout=SUPABASE_TIMEOUT_SECONDS,
			)
		except requests.RequestException as exc:
			raise HTTPException(status_code=503, detail=f"Supabase verify failed: {exc}") from exc

		if response.status_code == 200:
			try:
				data = response.json() if response.text else {}
			except ValueError:
				data = {}
			if not isinstance(data, dict):
				last_error = "Supabase verify returned an unexpected payload."
				continue

			session_payload = _extract_supabase_session_tokens(data)
			if session_payload.get("access_token"):
				return {
					"status": "ok",
					"session": session_payload,
				}

			last_error = "Supabase verify did not include an access token."
			continue

		try:
			error_data = response.json() if response.text else {}
		except ValueError:
			error_data = {}
		error_msg = ""
		if isinstance(error_data, dict):
			error_msg = str(error_data.get("message") or error_data.get("error_description") or "").strip()
		if not error_msg:
			error_msg = response.text[:220]
		last_error = error_msg or f"Supabase verify failed with status {response.status_code}."

	raise HTTPException(status_code=401, detail=f"Could not resolve reset token: {last_error or 'Invalid or expired link.'}")


@app.post("/auth/reset-password")
def auth_reset_password(
	payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	access_token = _normalize_spaces(str(payload.get("access_token") or payload.get("token") or ""))
	new_password = str(payload.get("new_password") or payload.get("password") or "").strip()

	if not access_token:
		raise HTTPException(status_code=400, detail="access_token is required.")
	if len(new_password) < 6:
		raise HTTPException(status_code=400, detail="new_password must be at least 6 characters.")

	try:
		response = requests.put(
			f"{SUPABASE_URL}/auth/v1/user",
			json={"password": new_password},
			headers={
				"apikey": SUPABASE_ANON_KEY,
				"Authorization": f"Bearer {access_token}",
				"Content-Type": "application/json",
			},
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
	except requests.RequestException as exc:
		raise HTTPException(status_code=503, detail=f"Supabase reset-password failed: {exc}") from exc

	if response.status_code != 200:
		error_data = response.json() if response.text else {}
		error_msg = ""
		if isinstance(error_data, dict):
			error_msg = str(error_data.get("message") or error_data.get("error_description") or "").strip()
		if not error_msg:
			error_msg = response.text[:220]
		raise HTTPException(status_code=response.status_code, detail=f"Reset-password failed: {error_msg}")

	return {
		"status": "ok",
		"message": "Password reset successful.",
	}


@app.post("/auth/otp/request")
def auth_otp_request(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	purpose = _normalize_spaces(str(payload.get("purpose") or "register")).lower() or "register"
	raw_channel = payload.get("channel") or payload.get("otp_channel")
	if raw_channel is None:
		has_email = bool(_normalize_spaces(str(payload.get("email") or "")))
		has_phone = bool(_normalize_spaces(str(payload.get("phone") or payload.get("phone_number") or "")))
		raw_channel = "email" if has_email else ("phone" if has_phone else "email")
	channel = _normalize_otp_channel(str(raw_channel))

	request_user: Optional[Dict[str, Any]] = None
	requester_id: Optional[str] = None
	profile: Optional[Dict[str, Any]] = None
	if purpose in {"vehicle_verify", "profile_update"}:
		request_user = _require_authenticated_user(authorization)
		if request_user is None:
			raise HTTPException(status_code=401, detail="Authentication is required for this OTP purpose.")
		requester_id = str(request_user.get("id") or "").strip()
		if not requester_id:
			raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
		profile = _supabase_get_profile_by_user_id(requester_id)

	identifier = _resolve_otp_identifier(channel=channel, payload=payload, profile=profile)

	raw_create_user = payload.get("create_user")
	if isinstance(raw_create_user, bool):
		create_user = raw_create_user
	elif raw_create_user is None:
		create_user = purpose in {"register", "signup"}
	else:
		create_user = str(raw_create_user).strip().lower() in {"1", "true", "yes", "on"}

	# Registration OTP must allow creating missing auth users, even if client sends create_user=false.
	if purpose in {"register", "signup"}:
		create_user = True

	raw_redirect_to = _normalize_spaces(str(payload.get("redirect_to") or ""))
	redirect_to: Optional[str] = raw_redirect_to or None
	if channel == "email" and not redirect_to:
		default_redirect = _normalize_spaces(PASSWORD_RESET_REDIRECT_URL)
		if default_redirect:
			redirect_to = default_redirect

	_request_supabase_otp(
		channel=channel,
		identifier=identifier,
		create_user=create_user,
		redirect_to=redirect_to,
		purpose=purpose,
	)

	return {
		"status": "ok",
		"purpose": purpose,
		"channel": channel,
		"identifier": identifier,
		"message": "OTP was sent successfully.",
	}


@app.post("/auth/otp/verify")
def auth_otp_verify(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	purpose = _normalize_spaces(str(payload.get("purpose") or "register")).lower() or "register"
	channel = _normalize_otp_channel(str(payload.get("channel") or payload.get("otp_channel") or "email"))

	request_user: Optional[Dict[str, Any]] = None
	token = _extract_bearer_token(authorization)
	if token:
		barrier_user = _verify_barrier_token(token)
		if barrier_user is not None:
			request_user = barrier_user
		else:
			request_user = _verify_supabase_user_token(token)

	if purpose in {"vehicle_verify", "profile_update"} and request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required for this OTP purpose.")

	requester_profile = None
	if request_user is not None:
		requester_id = str(request_user.get("id") or "").strip()
		if requester_id:
			requester_profile = _supabase_get_profile_by_user_id(requester_id)

	identifier = _resolve_otp_identifier(
		channel=channel,
		payload=payload,
		profile=requester_profile,
	)
	verify_payload = _verify_supabase_otp(
		channel=channel,
		identifier=identifier,
		token=str(payload.get("otp_token") or payload.get("token") or ""),
		verify_type=str(payload.get("type") or payload.get("verify_type") or "").strip() or None,
	)

	user_payload = verify_payload.get("user") if isinstance(verify_payload.get("user"), dict) else {}

	return {
		"status": "ok",
		"purpose": purpose,
		"channel": channel,
		"user": {
			"id": user_payload.get("id"),
			"email": user_payload.get("email"),
			"phone": user_payload.get("phone"),
		},
		"session": {
			"access_token": verify_payload.get("access_token"),
			"refresh_token": verify_payload.get("refresh_token"),
			"expires_in": verify_payload.get("expires_in"),
			"token_type": verify_payload.get("token_type"),
		},
		"message": "OTP verified successfully.",
	}


@app.put("/profile/update")
def profile_update(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	current_profile = _supabase_get_profile_by_user_id(requester_id) or {}
	current_first_name, current_last_name, _ = _resolve_profile_name_fields(
		current_profile if isinstance(current_profile, dict) else None,
		fallback_full_name=(current_profile or {}).get("full_name") if isinstance(current_profile, dict) else None,
	)
	name_fields_provided = any(
		key in payload
		for key in ["first_name", "firstName", "last_name", "lastName", "full_name", "name"]
	)
	requested_first_name, requested_last_name, _ = _resolve_profile_name_fields(payload)

	email = _normalize_spaces(str(payload.get("email") or "")).lower()
	phone_number = _normalize_spaces(str(payload.get("phone") or payload.get("phone_number") or ""))

	profile_patch: Dict[str, Any] = {}
	if name_fields_provided:
		next_first_name = requested_first_name or current_first_name
		next_last_name = requested_last_name or current_last_name
		next_full_name = _compose_full_name(next_first_name, next_last_name)
		if not next_full_name:
			raise HTTPException(status_code=400, detail="Name update requires at least first_name or full_name.")
		profile_patch["first_name"] = next_first_name
		profile_patch["last_name"] = next_last_name
		profile_patch["full_name"] = next_full_name
	if email:
		if "@" not in email:
			raise HTTPException(status_code=400, detail="email format is invalid.")
		other_with_email = _supabase_get_profile_by_field("email", email, case_insensitive=True)
		if isinstance(other_with_email, dict) and str(other_with_email.get("id") or "") != requester_id:
			raise HTTPException(status_code=409, detail="email is already in use.")
		profile_patch["email"] = email
	if phone_number:
		if not PHONE_PATTERN.fullmatch(phone_number):
			raise HTTPException(status_code=400, detail="phone must be 7-15 digits and may start with +.")
		other_with_phone = _supabase_get_profile_by_field("phone_number", phone_number)
		if isinstance(other_with_phone, dict) and str(other_with_phone.get("id") or "") != requester_id:
			raise HTTPException(status_code=409, detail="phone is already in use.")
		profile_patch["phone_number"] = phone_number

	if not profile_patch:
		raise HTTPException(status_code=400, detail="No profile fields were provided.")

	response = requests.patch(
		f"{SUPABASE_URL}/rest/v1/profiles",
		params={
			"id": f"eq.{requester_id}",
			"select": PROFILE_SELECT_FIELDS,
		},
		json=profile_patch,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
			content_type="application/json",
			prefer="return=representation",
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code not in {200, 204}:
		raise HTTPException(status_code=400, detail=f"Failed to update profile: {_missing_schema_hint(response.text)}")

	updated_rows = response.json() if response.text else []
	updated_profile = (
		updated_rows[0]
		if isinstance(updated_rows, list) and updated_rows and isinstance(updated_rows[0], dict)
		else (_supabase_get_profile_by_user_id(requester_id) or current_profile)
	)

	auth_patch: Dict[str, Any] = {}
	user_metadata_patch: Dict[str, Any] = {}
	if "email" in profile_patch:
		auth_patch["email"] = profile_patch["email"]
	if "first_name" in profile_patch:
		user_metadata_patch["first_name"] = profile_patch["first_name"]
	if "last_name" in profile_patch:
		user_metadata_patch["last_name"] = profile_patch["last_name"]
	if "full_name" in profile_patch:
		user_metadata_patch["full_name"] = profile_patch["full_name"]
	if "phone_number" in profile_patch:
		user_metadata_patch["phone_number"] = profile_patch["phone_number"]
		user_metadata_patch["phone"] = profile_patch["phone_number"]
	if user_metadata_patch:
		auth_patch["user_metadata"] = user_metadata_patch

	if auth_patch:
		try:
			auth_response = requests.put(
				f"{SUPABASE_URL}/auth/v1/admin/users/{requester_id}",
				json=auth_patch,
				headers={
					"apikey": SUPABASE_SERVICE_ROLE_KEY,
					"Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}",
					"Content-Type": "application/json",
				},
				timeout=SUPABASE_TIMEOUT_SECONDS,
			)
			if auth_response.status_code not in {200, 201}:
				LOGGER.warning("Auth metadata update failed for %s: %s", requester_id, auth_response.text[:220])
		except requests.RequestException as exc:
			LOGGER.warning("Auth metadata update request failed for %s: %s", requester_id, exc)

	return {
		"status": "ok",
		"profile": updated_profile,
		"message": "Profile updated successfully.",
	}


@app.get("/users")
def get_all_users(
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_role = _resolve_requester_role(request_user)
	if not _is_staff_role(requester_role):
		raise HTTPException(status_code=403, detail="Admin/Security role required to view users.")

	try:
		response = requests.get(
			f"{SUPABASE_URL}/rest/v1/profiles",
			params={"select": PROFILE_SELECT_FIELDS},
			headers=_supabase_headers(
				api_key=SUPABASE_SERVICE_ROLE_KEY,
				bearer=SUPABASE_SERVICE_ROLE_KEY,
			),
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
		if response.status_code != 200:
			raise HTTPException(status_code=502, detail="Failed to fetch users from database.")
		
		users = response.json()
		if isinstance(users, dict) and "error" in users:
			users = []
		
		return {"status": "ok", "users": users}
	except Exception as e:
		raise HTTPException(status_code=500, detail=str(e))



@app.post("/vehicles/add")
def vehicles_add(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	vehicle_payload = _extract_vehicle_payload(payload, require_plate=True)
	created_vehicle = _create_vehicle_for_owner(owner_id=requester_id, vehicle_payload=vehicle_payload)
	if not isinstance(created_vehicle, dict):
		raise HTTPException(status_code=502, detail="Vehicle was not created.")

	return {
		"status": "ok",
		"vehicle": created_vehicle,
	}


@app.get("/vehicles/my")
def vehicles_my(
	include_inactive: bool = False,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	rows = _list_vehicles_for_owner(requester_id)
	if not include_inactive:
		rows = [row for row in rows if bool(row.get("is_active", True))]

	return {
		"status": "ok",
		"count": len(rows),
		"vehicles": rows,
	}


@app.put("/vehicles/{vehicle_id}")
def vehicles_update(
	vehicle_id: str,
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	update_payload: Dict[str, Any] = {}
	try:
		update_payload.update(_extract_vehicle_payload(payload, require_plate=False))
	except HTTPException as exc:
		if str(exc.detail) != "No vehicle fields were provided.":
			raise

	if "is_active" in payload:
		raw_is_active = payload.get("is_active")
		if isinstance(raw_is_active, bool):
			update_payload["is_active"] = raw_is_active
		elif raw_is_active is not None:
			update_payload["is_active"] = str(raw_is_active).strip().lower() in {"1", "true", "yes", "on"}

	if not update_payload:
		raise HTTPException(status_code=400, detail="No updatable vehicle fields were provided.")

	updated_vehicle = _update_vehicle_for_owner(requester_id, vehicle_id, update_payload)
	if not isinstance(updated_vehicle, dict):
		raise HTTPException(status_code=404, detail="Vehicle was not found.")

	return {
		"status": "ok",
		"vehicle": updated_vehicle,
	}


@app.delete("/vehicles/{vehicle_id}")
def vehicles_delete(
	vehicle_id: str,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	vehicle_record = _fetch_vehicle_for_owner(requester_id, vehicle_id)
	if not isinstance(vehicle_record, dict):
		raise HTTPException(status_code=404, detail="Vehicle was not found.")

	deleted = _delete_vehicle_for_owner(requester_id, vehicle_id)
	if not deleted:
		raise HTTPException(status_code=404, detail="Vehicle was not found.")

	return {
		"status": "ok",
		"deleted_vehicle_id": vehicle_id,
	}


@app.post("/vehicles/verify")
def vehicles_verify(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	profile = _supabase_get_profile_by_user_id(requester_id)
	if not isinstance(profile, dict):
		raise HTTPException(status_code=404, detail="Profile was not found.")

	vehicle_id = _normalize_spaces(str(payload.get("vehicle_id") or ""))
	if vehicle_id:
		vehicle = _fetch_vehicle_for_owner(requester_id, vehicle_id)
	else:
		vehicle_plate_payload = _extract_vehicle_payload(payload, require_plate=True)
		vehicle = _fetch_vehicle_by_plate_for_owner(
			requester_id,
			vehicle_plate_payload["plate_letters_ar"],
			vehicle_plate_payload["plate_numbers_ar"],
		)

	if not isinstance(vehicle, dict):
		raise HTTPException(status_code=404, detail="Vehicle was not found.")

	channel = _normalize_otp_channel(str(payload.get("channel") or payload.get("otp_channel") or "email"))
	identifier = _resolve_otp_identifier(channel=channel, payload=payload, profile=profile)
	_verify_supabase_otp(
		channel=channel,
		identifier=identifier,
		token=str(payload.get("otp_token") or payload.get("token") or ""),
		verify_type=str(payload.get("type") or payload.get("verify_type") or "").strip() or None,
	)

	resolved_vehicle_id = str(vehicle.get("id") or "").strip()
	updated_vehicle = _update_vehicle_for_owner(
		requester_id,
		resolved_vehicle_id,
		{
			"is_verified": True,
			"verified_at": datetime.now(timezone.utc).isoformat(),
			"verified_by": requester_id,
		},
	)

	return {
		"status": "ok",
		"channel": channel,
		"vehicle": updated_vehicle or vehicle,
		"message": "Vehicle ownership verified.",
	}


@app.post("/notifications/fcm-token")
def notifications_register_fcm_token(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if not _supabase_configured():
		raise HTTPException(status_code=503, detail="Supabase is not configured.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	fcm_token = _normalize_spaces(str(payload.get("fcm_token") or payload.get("token") or ""))
	if not fcm_token:
		raise HTTPException(status_code=400, detail="fcm_token is required.")

	platform = _normalize_spaces(str(payload.get("platform") or "")) or None
	device_id = _normalize_spaces(str(payload.get("device_id") or "")) or None

	row = _upsert_notification_device_token(
		user_id=requester_id,
		fcm_token=fcm_token,
		platform=platform,
		device_id=device_id,
	)

	return {
		"status": "ok",
		"device": row,
		"message": "FCM token registered.",
	}


@app.get("/notifications/my")
def notifications_my(
	limit: int = 50,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	resolved_limit = max(1, min(200, int(limit)))
	rows = _list_notification_events_for_user(requester_id, resolved_limit)

	return {
		"status": "ok",
		"count": len(rows),
		"notifications": rows,
	}


@app.get("/health")
def health() -> Dict[str, Any]:
	local_models_required = _should_load_local_models()
	local_models_loaded = False
	local_model_error: Optional[str] = None

	if local_models_required:
		try:
			_ensure_models_loaded()
			local_models_loaded = True
		except Exception as exc:  # pragma: no cover
			local_model_error = str(exc)

	status_value = "ok" if (not local_models_required or local_models_loaded) else "degraded"
	model_inference_payload: Dict[str, Any] = {
		"provider": MODEL_INFERENCE_PROVIDER,
		"local_models_required": local_models_required,
		"local_models_loaded": local_models_loaded,
		"local_endpoint_enabled": MODEL_SERVICE_ENABLE_LOCAL_ENDPOINT,
		"remote_service_configured": _model_service_configured(),
		"remote_infer_url": _resolve_model_service_infer_url(),
		"header_name": MODEL_SERVICE_HEADER_NAME,
	}
	if local_model_error:
		model_inference_payload["local_model_error"] = local_model_error

	return {
		"status": status_value,
		"models_dir": str(MODELS_DIR),
		"ocr_model_file": MODEL_PATHS["ocr"].name,
		"model_inference": model_inference_payload,
		"supabase_configured": _supabase_configured(),
		"supabase_auth_enforced": SUPABASE_ENFORCE_AUTH,
		"pricing": {
			"default_billing_mode": PARKING_BILLING_MODE,
			"default_hourly_rate_egp": round(max(0.0, PARKING_HOURLY_RATE_EGP), 2),
			"default_daily_rate_egp": round(max(0.0, PARKING_DAILY_RATE_EGP), 2),
			"default_extra_after_first_hour_egp": round(max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP), 2),
			"location_hourly_rates": PARKING_LOCATION_HOURLY_RATES,
			"location_pricing": PARKING_LOCATION_PRICING,
			"location_capacities": PARKING_LOCATION_CAPACITIES,
			"garage_total_capacity": GARAGE_TOTAL_CAPACITY,
			"app_service_fee_egp": round(max(0.0, APP_SERVICE_FEE_EGP), 2),
		},
	}


@app.get("/supabase/health")
def supabase_health() -> Dict[str, Any]:
	if not _supabase_configured():
		return {
			"status": "disabled",
			"reason": "Set SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_SERVICE_ROLE_KEY",
		}

	try:
		response = requests.get(
			f"{SUPABASE_URL}/auth/v1/settings",
			headers={"apikey": SUPABASE_ANON_KEY},
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
		if response.status_code != 200:
			return {
				"status": "error",
				"http_status": response.status_code,
				"detail": response.text[:400],
			}

		existing_buckets = set(_supabase_list_buckets())
		required_buckets = {SUPABASE_RAW_BUCKET, SUPABASE_PROCESSED_BUCKET}
		missing_buckets = sorted(required_buckets - existing_buckets)

		table_checks: Dict[str, Dict[str, Any]] = {}
		all_tables_ready = True
		for table_name in sorted(CORE_TABLES):
			exists, detail = _supabase_table_exists(table_name)
			table_checks[table_name] = {
				"exists": exists,
				"detail": detail,
			}
			if not exists:
				all_tables_ready = False

		optional_checks: Dict[str, Dict[str, Any]] = {}
		for table_name in ["payment_transactions", "notification_device_tokens", "notification_events", "app_links"]:
			exists, detail = _supabase_table_exists(table_name)
			optional_checks[table_name] = {
				"exists": exists,
				"detail": detail,
			}

		ready_for_persistence = all_tables_ready and not missing_buckets
	except requests.RequestException as exc:
		return {
			"status": "error",
			"detail": str(exc),
		}
	except Exception as exc:
		return {
			"status": "error",
			"detail": str(exc),
		}

	return {
		"status": "ok" if ready_for_persistence else "partial",
		"url": SUPABASE_URL,
		"raw_bucket": SUPABASE_RAW_BUCKET,
		"processed_bucket": SUPABASE_PROCESSED_BUCKET,
		"auth_enforced": SUPABASE_ENFORCE_AUTH,
		"readiness": {
			"ready_for_persistence": ready_for_persistence,
			"missing_buckets": missing_buckets,
			"tables": table_checks,
			"optional_tables": optional_checks,
		},
	}


@app.get("/models")
def models() -> Dict[str, Any]:
	local_models_required = _should_load_local_models()
	if local_models_required:
		_ensure_models_loaded()

	return {
		"inference_provider": MODEL_INFERENCE_PROVIDER,
		"local_models_required": local_models_required,
		"local_endpoint_enabled": MODEL_SERVICE_ENABLE_LOCAL_ENDPOINT,
		"remote_service_configured": _model_service_configured(),
		"remote_infer_url": _resolve_model_service_infer_url(),
		"plate_detector": str(MODEL_PATHS["plate"]) if local_models_required else None,
		"ocr_model": str(MODEL_PATHS["ocr"]) if local_models_required else None,
	}


@app.get("/developer/models")
def developer_models(
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user, requester_role = _require_developer_user(authorization)
	payload = models()
	payload["request_user"] = {
		"id": request_user.get("id"),
		"email": request_user.get("email"),
		"role": requester_role,
	}
	return payload


@app.post("/model/infer")
async def model_infer(
	image: UploadFile = File(...),
	model_service_secret: Optional[str] = Header(default=None, alias=MODEL_SERVICE_HEADER_NAME),
) -> Dict[str, Any]:
	# Always use remote HuggingFace Space for inference
	if image.content_type and not image.content_type.startswith("image/") and image.content_type != "application/octet-stream":
		raise HTTPException(status_code=400, detail="Only image uploads are supported.")

	image_bytes = await image.read()
	if not image_bytes:
		raise HTTPException(status_code=400, detail="Uploaded file is empty.")

	response = _run_pipeline_remote(
		image_bytes=image_bytes,
		filename=image.filename,
		content_type=image.content_type,
	)
	response["filename"] = image.filename
	response["inference_provider"] = "remote"
	return response


def _resolve_history_scope(requester_id: str, requester_has_global_scope: bool, for_user_id: Optional[str]) -> Optional[str]:
	target_user_id = (for_user_id or "").strip() or None
	if target_user_id and not requester_has_global_scope and target_user_id != requester_id:
		raise HTTPException(status_code=403, detail="You can only view your own records.")
	if not requester_has_global_scope:
		target_user_id = requester_id
	return target_user_id


def _fetch_active_inside_counts_by_location() -> Tuple[int, Dict[str, int]]:
	if not _supabase_configured():
		return 0, {}

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={
			"select": "parking_location",
			"check_out_at": "is.null",
			"limit": "10000",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch inside occupancy counts: %s", response.text[:280])
		return 0, {}

	rows = response.json()
	if not isinstance(rows, list):
		return 0, {}

	inside_by_location: Dict[str, int] = {}
	total_inside = 0
	for row in rows:
		if not isinstance(row, dict):
			continue

		total_inside += 1
		raw_location = str(row.get("parking_location") or "").strip()
		if not raw_location:
			continue

		location_key = _normalize_location_key_for_config(raw_location.split(",", 1)[0])
		if not location_key:
			continue

		inside_by_location[location_key] = inside_by_location.get(location_key, 0) + 1

	return total_inside, inside_by_location


def _build_parking_locations_snapshot() -> Dict[str, Any]:
	total_inside_from_sessions, inside_by_location_from_sessions = _fetch_active_inside_counts_by_location()
	total_inside_inferred, inside_by_location_inferred = _count_inferred_inside_by_location()

	total_inside = total_inside_from_sessions + total_inside_inferred
	inside_by_location: Dict[str, int] = dict(inside_by_location_from_sessions)
	for location_key, inferred_count in inside_by_location_inferred.items():
		inside_by_location[location_key] = inside_by_location.get(location_key, 0) + inferred_count

	location_keys = set(PARKING_LOCATION_PRICING.keys())
	location_keys.update(PARKING_LOCATION_HOURLY_RATES.keys())
	location_keys.update(PARKING_LOCATION_CAPACITIES.keys())
	location_keys.update(inside_by_location.keys())
	if not location_keys:
		location_keys.add("DEFAULT")

	locations: List[Dict[str, Any]] = []
	for location_key in sorted(location_keys):
		policy = _resolve_location_pricing_policy(location_key)
		location_capacity = max(0, int(policy.get("location_capacity") or 0))
		inside_count = inside_by_location.get(location_key, 0)
		left_count = max(0, location_capacity - inside_count)
		base_window_hours = 24 if str(policy.get("billing_mode") or "hourly") == "daily" else 1

		locations.append(
			{
				"parking_location": location_key,
				"pricing": {
					"billing_mode": str(policy.get("billing_mode") or "hourly"),
					"base_rate_egp": round(max(0.0, float(policy.get("base_rate_egp") or 0.0)), 2),
					"extra_after_first_hour_egp": round(max(0.0, float(policy.get("extra_after_first_hour_egp") or 0.0)), 2),
					"app_service_fee_egp": round(max(0.0, APP_SERVICE_FEE_EGP), 2),
					"base_window_hours": base_window_hours,
					"rate_source": str(policy.get("rate_source") or "default"),
				},
				"occupancy": {
					"inside": inside_count,
					"inside_from_registered_sessions": inside_by_location_from_sessions.get(location_key, 0),
					"inside_from_inferred_unmatched": inside_by_location_inferred.get(location_key, 0),
					"total_capacity": location_capacity,
					"left": left_count,
					"display": f"{inside_count}/{location_capacity}",
					"capacity_source": str(policy.get("location_capacity_source") or "default"),
				},
			}
		)

	unknown_location_inside = max(0, total_inside - sum(inside_by_location.values()))
	garage_left = max(0, GARAGE_TOTAL_CAPACITY - total_inside)

	return {
		"garage": {
			"inside": total_inside,
			"inside_from_registered_sessions": total_inside_from_sessions,
			"inside_from_inferred_unmatched": total_inside_inferred,
			"total_capacity": GARAGE_TOTAL_CAPACITY,
			"left": garage_left,
			"display": f"{total_inside}/{GARAGE_TOTAL_CAPACITY}",
			"unknown_location_inside": unknown_location_inside,
		},
		"locations": locations,
	}


@app.get("/parking/locations")
def parking_locations(
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	snapshot = _build_parking_locations_snapshot()

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
		"pricing_defaults": {
			"billing_mode": PARKING_BILLING_MODE,
			"hourly_rate_egp": round(max(0.0, PARKING_HOURLY_RATE_EGP), 2),
			"daily_rate_egp": round(max(0.0, PARKING_DAILY_RATE_EGP), 2),
			"extra_after_first_hour_egp": round(max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP), 2),
			"app_service_fee_egp": round(max(0.0, APP_SERVICE_FEE_EGP), 2),
		},
		"garage": snapshot["garage"],
		"locations": snapshot["locations"],
	}


@app.get("/parking/occupancy")
def parking_occupancy(
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	snapshot = _build_parking_locations_snapshot()

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
		"garage": snapshot["garage"],
		"locations": [
			{
				"parking_location": row.get("parking_location"),
				"occupancy": row.get("occupancy"),
			}
			for row in snapshot["locations"]
		],
	}


def _fetch_vehicle_map_by_ids(vehicle_ids: List[str]) -> Dict[str, Dict[str, Any]]:
	if not vehicle_ids or not _supabase_configured():
		return {}

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/vehicles",
		params={
			"select": "id,owner_id,plate_letters_ar,plate_numbers_ar,plate_letters_en,plate_numbers_en,car_name,car_model,car_color,is_active,is_verified,verified_at,verified_by,created_at,updated_at",
			"id": f"in.({','.join(sorted(set(vehicle_ids)))})",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch vehicle map: %s", response.text[:280])
		return {}

	rows = response.json()
	if not isinstance(rows, list):
		return {}

	result: Dict[str, Dict[str, Any]] = {}
	for row in rows:
		if isinstance(row, dict) and row.get("id"):
			result[str(row["id"])] = row
	return result


def _fetch_session_map_by_ids(session_ids: List[str]) -> Dict[str, Dict[str, Any]]:
	if not session_ids or not _supabase_configured():
		return {}

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params={
			"select": "id,vehicle_id,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
			"id": f"in.({','.join(sorted(set(session_ids)))})",
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch session map: %s", response.text[:280])
		return {}

	rows = response.json()
	if not isinstance(rows, list):
		return {}

	result: Dict[str, Dict[str, Any]] = {}
	for row in rows:
		if isinstance(row, dict) and row.get("id"):
			result[str(row["id"])] = row
	return result


def _fetch_latest_event_by_session_ids(session_ids: List[str]) -> Dict[str, Dict[str, Any]]:
	if not session_ids or not _supabase_configured():
		return {}

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/car_events",
		params={
			"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
			"session_id": f"in.({','.join(sorted(set(session_ids)))})",
			"order": "captured_at.desc",
			"limit": str(max(len(session_ids) * 6, 200)),
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch latest events by session ids: %s", response.text[:280])
		return {}

	rows = response.json()
	if not isinstance(rows, list):
		return {}

	result: Dict[str, Dict[str, Any]] = {}
	for row in rows:
		if not isinstance(row, dict):
			continue
		session_id = str(row.get("session_id") or "").strip()
		if not session_id or session_id in result:
			continue
		result[session_id] = row

	return result


def _event_plate_value(value: Any) -> Optional[str]:
	raw = _normalize_spaces(str(value or ""))
	if not raw or raw.upper() == "N/A":
		return None
	return raw


def _build_unmatched_plate_key(event_row: Dict[str, Any]) -> Tuple[Optional[str], Optional[str], Optional[str]]:
	plate_arabic = _event_plate_value(event_row.get("ocr_arabic"))
	plate_english = _event_plate_value(event_row.get("ocr_english"))

	if plate_arabic:
		return f"AR::{plate_arabic}", plate_arabic, plate_english
	if plate_english:
		plate_english_upper = plate_english.upper()
		return f"EN::{plate_english_upper}", None, plate_english_upper
	return None, None, None


def _resolve_event_timestamp(event_row: Dict[str, Any]) -> Optional[datetime]:
	for field_name in ["captured_at", "created_at"]:
		parsed = _parse_iso_datetime(event_row.get(field_name))
		if parsed is not None:
			return parsed
	return None


def _resolve_event_location_key(event_row: Dict[str, Any]) -> Optional[str]:
	raw_location = _normalize_spaces(str(event_row.get("parking_location") or ""))
	if not raw_location:
		event_metadata = event_row.get("event_metadata") if isinstance(event_row.get("event_metadata"), dict) else {}
		raw_location = _normalize_spaces(str(event_metadata.get("parking_location") or ""))

	if not raw_location:
		return None

	return _normalize_location_key_for_config(raw_location.split(",", 1)[0])


def _fetch_unmatched_car_events(limit: int = 4000) -> List[Dict[str, Any]]:
	if not _supabase_configured():
		return []

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/car_events",
		params={
			"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,event_metadata,captured_at,created_at",
			"session_id": "is.null",
			"order": "captured_at.desc",
			"limit": str(max(limit, 1)),
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		LOGGER.warning("Failed to fetch unmatched car events: %s", response.text[:280])
		return []

	rows = response.json()
	if not isinstance(rows, list):
		return []

	return [row for row in rows if isinstance(row, dict)]


def _build_inferred_unmatched_session_rows(limit: int = 300, *, inside_only: bool) -> List[Dict[str, Any]]:
	events = _fetch_unmatched_car_events(limit=max(limit * 8, 400))
	if not events:
		return []

	epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
	sorted_events = sorted(events, key=lambda row: _resolve_event_timestamp(row) or epoch)

	state_by_plate: Dict[str, Dict[str, Any]] = {}
	for event in sorted_events:
		event_type = str(event.get("event_type") or "").strip().lower()
		if event_type not in SUPPORTED_EVENT_TYPES:
			continue

		event_timestamp = _resolve_event_timestamp(event)
		if event_timestamp is None:
			continue

		plate_key, plate_arabic, plate_english = _build_unmatched_plate_key(event)
		if not plate_key:
			continue

		state = state_by_plate.setdefault(
			plate_key,
			{
				"plate_key": plate_key,
				"plate_arabic": plate_arabic,
				"plate_english": plate_english,
				"inside": False,
				"has_transition": False,
				"check_in_at": None,
				"check_out_at": None,
				"parking_location": None,
				"latest_event": None,
				"latest_event_at": None,
				"event_count": 0,
			},
		)

		if plate_arabic and not state.get("plate_arabic"):
			state["plate_arabic"] = plate_arabic
		if plate_english and not state.get("plate_english"):
			state["plate_english"] = plate_english

		state["event_count"] = int(state.get("event_count") or 0) + 1
		state["latest_event"] = event
		state["latest_event_at"] = event_timestamp

		resolved_location = _resolve_event_location_key(event)
		if resolved_location:
			state["parking_location"] = resolved_location

		if event_type == "entry":
			state["inside"] = True
			state["has_transition"] = True
			state["check_in_at"] = event_timestamp
			state["check_out_at"] = None
			if resolved_location:
				state["parking_location"] = resolved_location
		elif event_type == "exit":
			state["inside"] = False
			state["has_transition"] = True
			state["check_out_at"] = event_timestamp
			if state.get("check_in_at") is None:
				state["check_in_at"] = event_timestamp

	inferred_rows: List[Dict[str, Any]] = []
	for state in state_by_plate.values():
		if not bool(state.get("has_transition")):
			continue

		is_inside = bool(state.get("inside"))
		if inside_only and not is_inside:
			continue

		latest_event = state.get("latest_event") if isinstance(state.get("latest_event"), dict) else None
		latest_event_at = state.get("latest_event_at") if isinstance(state.get("latest_event_at"), datetime) else None
		if latest_event is None or latest_event_at is None:
			continue

		check_in_at = state.get("check_in_at") if isinstance(state.get("check_in_at"), datetime) else latest_event_at
		check_out_at = state.get("check_out_at") if isinstance(state.get("check_out_at"), datetime) else None
		if not is_inside and check_out_at is None:
			check_out_at = latest_event_at

		parking_location = state.get("parking_location")
		if not parking_location:
			parking_location = _resolve_event_location_key(latest_event)

		plate_key = str(state.get("plate_key") or "")
		session_id = f"inferred:{hashlib.sha1(plate_key.encode('utf-8')).hexdigest()[:20]}"
		session_row: Dict[str, Any] = {
			"id": session_id,
			"vehicle_id": None,
			"check_in_at": check_in_at.isoformat(),
			"payment_confirmed_at": None,
			"check_out_at": check_out_at.isoformat() if check_out_at else None,
			"left_within_5_minutes": False,
			"status": "inferred_inside_unmatched" if is_inside else "inferred_exited_unmatched",
			"parking_location": parking_location,
			"created_by": None,
			"notes": "Inferred from unmatched OCR events.",
			"created_at": check_in_at.isoformat(),
			"updated_at": latest_event_at.isoformat(),
			"inferred": True,
		}

		pricing = _compute_session_time_and_fee(session_row, parking_location_override=parking_location)
		inferred_rows.append(
			{
				"session": session_row,
				"vehicle": None,
				"latest_event": latest_event,
				"events": [latest_event],
				"plate": {
					"key": plate_key,
					"arabic": state.get("plate_arabic"),
					"english": state.get("plate_english"),
				},
				"event_count": int(state.get("event_count") or 0),
				"inferred_unmatched": True,
				"payment_status": _derive_payment_status_from_session(session_row),
				"left_within_5_minutes": False,
				"pricing": pricing,
				"gate_decision_if_exit_attempted": _build_gate_decision(
					event_type="exit",
					session_row=session_row,
					plate_detected=True,
				),
			}
		)

	inferred_rows.sort(
		key=lambda row: _resolve_event_timestamp(row.get("latest_event") if isinstance(row.get("latest_event"), dict) else {}) or epoch,
		reverse=True,
	)
	return inferred_rows[:limit]


def _count_inferred_inside_by_location() -> Tuple[int, Dict[str, int]]:
	inferred_inside_rows = _build_inferred_unmatched_session_rows(limit=1200, inside_only=True)
	inside_by_location: Dict[str, int] = {}
	total_inside = 0

	for row in inferred_inside_rows:
		session_row = row.get("session") if isinstance(row.get("session"), dict) else {}
		total_inside += 1

		location_key = _normalize_location_key_for_config(str(session_row.get("parking_location") or ""))
		if not location_key:
			continue

		inside_by_location[location_key] = inside_by_location.get(location_key, 0) + 1

	return total_inside, inside_by_location


def _list_event_feed(
	*,
	limit: int,
	target_user_id: Optional[str],
	event_type: str,
	within_5_only: bool,
) -> List[Dict[str, Any]]:
	fetch_limit = max(limit * 4, 300) if within_5_only else limit
	event_params: Dict[str, str] = {
		"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
		"event_type": f"eq.{event_type}",
		"order": "captured_at.desc",
		"limit": str(fetch_limit),
	}

	if target_user_id:
		vehicle_ids = _fetch_vehicle_ids_for_owner(target_user_id)
		if not vehicle_ids:
			return []
		event_params["vehicle_id"] = f"in.({','.join(vehicle_ids)})"

	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/car_events",
		params=event_params,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		raise HTTPException(status_code=502, detail=f"Failed to load event feed: {response.text[:320]}")

	events_rows = response.json()
	events = events_rows if isinstance(events_rows, list) else []

	session_ids: List[str] = []
	vehicle_ids: List[str] = []
	for event in events:
		if not isinstance(event, dict):
			continue
		sid = str(event.get("session_id") or "").strip()
		vid = str(event.get("vehicle_id") or "").strip()
		if sid:
			session_ids.append(sid)
		if vid:
			vehicle_ids.append(vid)

	session_map = _fetch_session_map_by_ids(session_ids)
	vehicle_map = _fetch_vehicle_map_by_ids(vehicle_ids)

	filtered_events: List[Dict[str, Any]] = []
	for event in events:
		if not isinstance(event, dict):
			continue

		event_metadata = event.get("event_metadata") if isinstance(event.get("event_metadata"), dict) else {}
		session_id = str(event.get("session_id") or "").strip()
		session_row = session_map.get(session_id)
		left_within_5 = bool(
			(session_row or {}).get("left_within_5_minutes")
			or event_metadata.get("left_within_5_minutes")
		)

		if within_5_only and not left_within_5:
			continue

		filtered_events.append(event)
		if len(filtered_events) >= limit:
			break

	items: List[Dict[str, Any]] = []
	for event in filtered_events:
		session_id = str(event.get("session_id") or "").strip()
		vehicle_id = str(event.get("vehicle_id") or "").strip()
		session_row = session_map.get(session_id)
		event_metadata = event.get("event_metadata") if isinstance(event.get("event_metadata"), dict) else {}
		payment_status = event_metadata.get("payment_status") or _derive_payment_status_from_session(session_row)
		pricing = (
			event_metadata.get("pricing")
			if isinstance(event_metadata.get("pricing"), dict)
			else _compute_session_time_and_fee(
				session_row,
				parking_location_override=(
					str(event.get("parking_location") or "").strip()
					or str(event_metadata.get("parking_location") or "").strip()
					or None
				),
			)
		)
		plate_detected = str(event.get("ocr_arabic") or "").strip().upper() not in {"", "N/A"} or str(event.get("ocr_english") or "").strip().upper() not in {"", "N/A"}
		gate_decision = event_metadata.get("gate_decision") if isinstance(event_metadata.get("gate_decision"), dict) else _build_gate_decision(
			event_type=event_type,
			session_row=session_row,
			plate_detected=plate_detected,
		)
		items.append(
			{
				"event": event,
				"vehicle": vehicle_map.get(vehicle_id),
				"session": session_row,
				"payment_status": payment_status,
				"left_within_5_minutes": bool(
					(session_row or {}).get("left_within_5_minutes")
					or event_metadata.get("left_within_5_minutes")
				),
				"pricing": pricing,
				"gate_decision": gate_decision,
			}
		)

	return items


@app.get("/events/entered-cars")
def entered_cars_feed(
	limit: int = 100,
	for_user_id: Optional[str] = None,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if limit < 1 or limit > 500:
		raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	requester_has_global_scope = _can_view_global_records(requester_role)
	target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)

	items = _list_event_feed(
		limit=limit,
		target_user_id=target_user_id,
		event_type="entry",
		within_5_only=False,
	)

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
		"scope": {
			"for_user_id": target_user_id,
			"limit": limit,
			"event_type": "entry",
		},
		"items": items,
	}


@app.get("/events/leaving-cars-within-5-minutes")
def leaving_within_5_minutes_feed(
	limit: int = 100,
	for_user_id: Optional[str] = None,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if limit < 1 or limit > 500:
		raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	requester_has_global_scope = _can_view_global_records(requester_role)
	target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)

	items = _list_event_feed(
		limit=limit,
		target_user_id=target_user_id,
		event_type="exit",
		within_5_only=True,
	)

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
		"scope": {
			"for_user_id": target_user_id,
			"limit": limit,
			"event_type": "exit",
			"within_5_minutes_after_payment": True,
		},
		"items": items,
	}


@app.get("/events/new-cars")
def new_cars_feed(
	limit: int = 100,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if limit < 1 or limit > 500:
		raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	if not _is_staff_role(requester_role):
		raise HTTPException(status_code=403, detail="Only admin/security can view new-car alerts.")

	fetch_limit = max(limit * 8, 400)
	response = requests.get(
		f"{SUPABASE_URL}/rest/v1/car_events",
		params={
			"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
			"order": "captured_at.desc",
			"limit": str(fetch_limit),
		},
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if response.status_code != 200:
		raise HTTPException(status_code=502, detail=f"Failed to fetch new-car alerts: {response.text[:320]}")

	events_rows = response.json()
	events = events_rows if isinstance(events_rows, list) else []

	session_ids: List[str] = []
	vehicle_ids: List[str] = []
	for event in events:
		if not isinstance(event, dict):
			continue
		sid = str(event.get("session_id") or "").strip()
		vid = str(event.get("vehicle_id") or "").strip()
		if sid:
			session_ids.append(sid)
		if vid:
			vehicle_ids.append(vid)

	session_map = _fetch_session_map_by_ids(session_ids)
	vehicle_map = _fetch_vehicle_map_by_ids(vehicle_ids)

	items: List[Dict[str, Any]] = []
	for event in events:
		if not isinstance(event, dict):
			continue

		event_metadata = event.get("event_metadata") if isinstance(event.get("event_metadata"), dict) else {}
		new_car_alert = event_metadata.get("new_car_alert") if isinstance(event_metadata.get("new_car_alert"), dict) else {}

		if new_car_alert.get("is_new_car") is not True:
			app_registration = event_metadata.get("app_registration") if isinstance(event_metadata.get("app_registration"), dict) else {}
			if app_registration.get("is_registered") is False:
				new_car_alert = {
					"is_new_car": True,
					"notify_roles": ["admin", "security"],
					"action": "vendor_outreach_recommended",
					"reason": "car_not_registered_by_any_user",
					"message": "New car detected. Admin/security should contact a vendor to call the driver and introduce the app.",
				}
			else:
				continue

		session_id = str(event.get("session_id") or "").strip()
		vehicle_id = str(event.get("vehicle_id") or "").strip()

		items.append(
			{
				"event": event,
				"vehicle": vehicle_map.get(vehicle_id),
				"session": session_map.get(session_id),
				"new_car_alert": new_car_alert,
			}
		)

		if len(items) >= limit:
			break

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": True,
		},
		"scope": {
			"limit": limit,
			"event_type": "new_car_alert",
			"notify_roles": ["admin", "security"],
			"action": "vendor_outreach_recommended",
		},
		"items": items,
	}


@app.get("/events/left-cars")
def left_cars_feed(
	limit: int = 100,
	for_user_id: Optional[str] = None,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if limit < 1 or limit > 500:
		raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	requester_has_global_scope = _can_view_global_records(requester_role)
	target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)

	items = _list_event_feed(
		limit=limit,
		target_user_id=target_user_id,
		event_type="exit",
		within_5_only=False,
	)

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
		"scope": {
			"for_user_id": target_user_id,
			"limit": limit,
			"event_type": "exit",
			"within_5_minutes_after_payment": False,
		},
		"items": items,
	}


@app.get("/events/inside-cars")
def inside_cars_feed(
	limit: int = 100,
	for_user_id: Optional[str] = None,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if limit < 1 or limit > 500:
		raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	requester_has_global_scope = _can_view_global_records(requester_role)
	target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)

	session_params: Dict[str, str] = {
		"select": "id,vehicle_id,plate_arabic,plate_english,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
		"check_out_at": "is.null",
		"status": "not.in.(exited,left_without_payment)",
		"order": "check_in_at.desc",
		"limit": str(limit),
	}

	if target_user_id:
		vehicle_ids = _fetch_vehicle_ids_for_owner(target_user_id)
		if not vehicle_ids:
			return {
				"status": "ok",
				"requester": {
					"id": requester_id,
					"role": requester_role or "user",
					"is_staff": requester_is_staff,
				},
				"scope": {
					"for_user_id": target_user_id,
					"limit": limit,
					"event_type": "inside",
				},
				"items": [],
			}
		session_params["vehicle_id"] = f"in.({','.join(vehicle_ids)})"

	sessions_response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params=session_params,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if sessions_response.status_code != 200:
		raise HTTPException(status_code=502, detail=f"Failed to fetch inside sessions: {sessions_response.text[:320]}")

	sessions_rows = sessions_response.json()
	sessions = sessions_rows if isinstance(sessions_rows, list) else []

	vehicle_ids_in_sessions = [
		str(row.get("vehicle_id"))
		for row in sessions
		if isinstance(row, dict) and row.get("vehicle_id")
	]
	vehicle_map = _fetch_vehicle_map_by_ids(vehicle_ids_in_sessions)

	session_ids = [
		str(row.get("id"))
		for row in sessions
		if isinstance(row, dict) and row.get("id")
	]
	latest_event_by_session = _fetch_latest_event_by_session_ids(session_ids)

	items: List[Dict[str, Any]] = []
	for session in sessions:
		if not isinstance(session, dict):
			continue

		session_id = str(session.get("id") or "").strip()
		vehicle_id = str(session.get("vehicle_id") or "").strip()
		
		# Self-healing: Hide and close old ghost sessions created before the fix
		v_id = vehicle_id.upper()
		plate_ar = str(session.get("plate_arabic") or "").strip().upper()
		plate_en = str(session.get("plate_english") or "").strip().upper()
		if v_id in {"", "NONE", "NULL"} and plate_ar in {"", "N/A", "NONE", "NULL", "-"} and plate_en in {"", "N/A", "NONE", "NULL", "-"}:
			LOGGER.warning("Self-healing: Closing invalid ghost session %s", session_id)
			_update_parking_session(session_id, {
				"status": "exited",
				"check_out_at": datetime.now(timezone.utc).isoformat(),
				"notes": "Closed invalid ghost session automatically."
			})
			continue
			
		latest_event = latest_event_by_session.get(session_id)
		# Pricing and Status calculation
		pricing = _compute_session_time_and_fee(session)
		payment_status = _derive_payment_status_from_session(session)
		plate_detected_in_latest_event = bool(
			isinstance(latest_event, dict)
			and (
				str(latest_event.get("ocr_arabic") or "").strip().upper() not in {"", "N/A"}
				or str(latest_event.get("ocr_english") or "").strip().upper() not in {"", "N/A"}
			)
		)

		items.append(
			{
				"session": session,
				"vehicle": vehicle_map.get(vehicle_id),
				"latest_event": latest_event,
				"payment_status": payment_status,
				"pricing": pricing,
				"gate_decision_if_exit_attempted": _build_gate_decision(
					event_type="exit",
					session_row=session,
					plate_detected=True,
				),
				"plate_detected_in_latest_event": plate_detected_in_latest_event,
				"inferred_unmatched": False,
			}
		)

	if target_user_id is None:
		real_session_plates = set()
		for item in items:
			s = item.get("session") or {}
			plate_ar = _normalize_plate_text(s.get("plate_arabic"))
			plate_en = _normalize_plate_text(s.get("plate_english"))
			if plate_ar: real_session_plates.add(plate_ar)
			if plate_en: real_session_plates.add(plate_en)

		inferred_items = _build_inferred_unmatched_session_rows(limit=max(limit * 3, 300), inside_only=True)
		for inferred in inferred_items:
			inf_plate = inferred.get("plate")
			inf_plate_norm = _normalize_plate_text(inf_plate)
			if inf_plate_norm in real_session_plates:
				continue

			items.append(
				{
					"session": inferred.get("session"),
					"vehicle": None,
					"latest_event": inferred.get("latest_event"),
					"payment_status": inferred.get("payment_status"),
					"pricing": inferred.get("pricing"),
					"gate_decision_if_exit_attempted": inferred.get("gate_decision_if_exit_attempted"),
					"plate_detected_in_latest_event": True,
					"inferred_unmatched": True,
					"plate": inf_plate,
					"event_count": inferred.get("event_count"),
				}
			)

	epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
	items.sort(
		key=lambda row: (
			_parse_iso_datetime(((row.get("session") if isinstance(row.get("session"), dict) else {}).get("check_in_at")))
			or _resolve_event_timestamp((row.get("latest_event") if isinstance(row.get("latest_event"), dict) else {}))
			or epoch
		),
		reverse=True,
	)
	items = items[:limit]

	parking_snapshot = _build_parking_locations_snapshot()

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
		"scope": {
			"for_user_id": target_user_id,
			"limit": limit,
			"event_type": "inside",
		},
		"garage_occupancy": parking_snapshot["garage"],
		"location_pricing": parking_snapshot["locations"],
		"items": items,
	}


@app.get("/parking/history")
def parking_history(
	limit: int = 100,
	for_user_id: Optional[str] = None,
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if limit < 1 or limit > 500:
		raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")

	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	requester_has_global_scope = _can_view_global_records(requester_role)

	target_user_id = (for_user_id or "").strip() or None
	if target_user_id and not requester_has_global_scope and target_user_id != requester_id:
		raise HTTPException(status_code=403, detail="You can only view your own parking history.")

	if not requester_has_global_scope:
		target_user_id = requester_id

	session_params: Dict[str, str] = {
		"select": "id,vehicle_id,plate_arabic,plate_english,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
		"order": "check_in_at.desc",
		"limit": str(limit),
	}

	if target_user_id:
		vehicle_ids = _fetch_vehicle_ids_for_owner(target_user_id)
		if not vehicle_ids:
			return {
				"status": "ok",
				"requester": {
					"id": requester_id,
					"role": requester_role or "user",
					"is_staff": requester_is_staff,
				},
				"scope": {
					"for_user_id": target_user_id,
					"limit": limit,
				},
				"history": [],
			}

		session_params["vehicle_id"] = f"in.({','.join(vehicle_ids)})"

	sessions_response = requests.get(
		f"{SUPABASE_URL}/rest/v1/parking_sessions",
		params=session_params,
		headers=_supabase_headers(
			api_key=SUPABASE_SERVICE_ROLE_KEY,
			bearer=SUPABASE_SERVICE_ROLE_KEY,
		),
		timeout=SUPABASE_TIMEOUT_SECONDS,
	)
	if sessions_response.status_code != 200:
		raise HTTPException(
			status_code=502,
			detail=f"Failed to fetch parking history: {sessions_response.text[:320]}",
		)

	sessions_rows = sessions_response.json()
	sessions = sessions_rows if isinstance(sessions_rows, list) else []

	vehicle_ids_in_history: List[str] = []
	for row in sessions:
		if isinstance(row, dict) and row.get("vehicle_id"):
			vehicle_ids_in_history.append(str(row["vehicle_id"]))

	vehicle_map: Dict[str, Dict[str, Any]] = {}
	if vehicle_ids_in_history:
		vehicle_response = requests.get(
			f"{SUPABASE_URL}/rest/v1/vehicles",
			params={
				"select": "id,owner_id,plate_letters_ar,plate_numbers_ar,plate_letters_en,plate_numbers_en,car_model,car_color,is_active",
				"id": f"in.({','.join(sorted(set(vehicle_ids_in_history)))})",
			},
			headers=_supabase_headers(
				api_key=SUPABASE_SERVICE_ROLE_KEY,
				bearer=SUPABASE_SERVICE_ROLE_KEY,
			),
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
		if vehicle_response.status_code == 200:
			vehicle_rows = vehicle_response.json()
			if isinstance(vehicle_rows, list):
				for vehicle in vehicle_rows:
					if isinstance(vehicle, dict) and vehicle.get("id"):
						vehicle_map[str(vehicle["id"])] = vehicle

	session_ids: List[str] = []
	for row in sessions:
		if isinstance(row, dict) and row.get("id"):
			session_ids.append(str(row["id"]))

	events_by_session: Dict[str, List[Dict[str, Any]]] = {}
	if session_ids:
		events_response = requests.get(
			f"{SUPABASE_URL}/rest/v1/car_events",
			params={
				"select": "id,session_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
				"session_id": f"in.({','.join(sorted(set(session_ids)))})",
				"order": "captured_at.desc",
				"limit": str(max(limit * 6, 200)),
			},
			headers=_supabase_headers(
				api_key=SUPABASE_SERVICE_ROLE_KEY,
				bearer=SUPABASE_SERVICE_ROLE_KEY,
			),
			timeout=SUPABASE_TIMEOUT_SECONDS,
		)
		if events_response.status_code == 200:
			events_rows = events_response.json()
			if isinstance(events_rows, list):
				for event in events_rows:
					if not isinstance(event, dict):
						continue
					sid = str(event.get("session_id") or "").strip()
					if not sid:
						continue
					events_by_session.setdefault(sid, []).append(event)

	history_items: List[Dict[str, Any]] = []
	for session in sessions:
		if not isinstance(session, dict):
			continue
		session_id = str(session.get("id") or "").strip()
		vehicle_id = str(session.get("vehicle_id") or "").strip()
		pricing = _compute_session_time_and_fee(session)
		payment_status = _derive_payment_status_from_session(session)
		history_items.append(
			{
				"session": session,
				"payment_status": payment_status,
				"left_within_5_minutes": bool(session.get("left_within_5_minutes")),
				"pricing": pricing,
				"gate_decision_if_exit_attempted": _build_gate_decision(
					event_type="exit",
					session_row=session,
					plate_detected=True,
				),
				"vehicle": vehicle_map.get(vehicle_id),
				"events": events_by_session.get(session_id, []),
				"inferred_unmatched": False,
			}
		)

	if target_user_id is None:
		inferred_history = _build_inferred_unmatched_session_rows(limit=max(limit * 3, 300), inside_only=False)
		for inferred in inferred_history:
			history_items.append(
				{
					"session": inferred.get("session"),
					"payment_status": inferred.get("payment_status"),
					"left_within_5_minutes": False,
					"pricing": inferred.get("pricing"),
					"gate_decision_if_exit_attempted": inferred.get("gate_decision_if_exit_attempted"),
					"vehicle": None,
					"events": inferred.get("events", []),
					"inferred_unmatched": True,
					"plate": inferred.get("plate"),
					"event_count": inferred.get("event_count"),
				}
			)

	epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
	history_items.sort(
		key=lambda row: (
			_parse_iso_datetime(((row.get("session") if isinstance(row.get("session"), dict) else {}).get("check_in_at")))
			or _resolve_event_timestamp(((row.get("events") or [None])[0] if isinstance((row.get("events") or [None])[0], dict) else {}))
			or epoch
		),
		reverse=True,
	)
	history_items = history_items[:limit]

	parking_snapshot = _build_parking_locations_snapshot()

	return {
		"status": "ok",
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
		"scope": {
			"for_user_id": target_user_id,
			"limit": limit,
		},
		"garage_occupancy": parking_snapshot["garage"],
		"location_pricing": parking_snapshot["locations"],
		"history": history_items,
	}


@app.post("/payments/manual/cash-confirm")
def manual_cash_confirm_payment(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	if not _is_staff_role(requester_role):
		raise HTTPException(status_code=403, detail="Only admin/security can confirm manual cash payments.")

	session_id = str(payload.get("session_id") or "").strip()
	if not session_id:
		raise HTTPException(status_code=400, detail="session_id is required.")

	parking_location = _normalize_parking_location(payload.get("parking_location"))
	
	# === GHOST FIX: Handle Inferred Sessions (Unregistered Cars) ===
	if session_id.startswith("inferred:"):
		LOGGER.info("Handling inferred session checkout for: %s", session_id)
		# Extract plate info from inferred session (if possible) or use provided plate
		# For simplicity, we search for the latest unmatched event for this 'inferred' car
		# and create a real session for it.
		plate_ar = str(payload.get("plate_arabic") or payload.get("plate") or "").strip()
		if not plate_ar:
			# Fallback: try to find it in the inside feed cache or inferred logic
			# But usually the dashboard will send the plate now.
			raise HTTPException(status_code=400, detail="plate is required for inferred session checkout.")
		
		# 1. Create a REAL session
		new_session = _insert_parking_session({
			"plate_arabic": plate_ar,
			"status": "inside",
			"check_in_at": datetime.now(timezone.utc).isoformat(),
			"parking_location": parking_location or "OPERA",
			"notes": f"Inferred promote from {session_id}"
		})
		if not new_session:
			raise HTTPException(status_code=500, detail="Failed to promote inferred session to real session.")
		
		real_sid = str(new_session.get("id"))
		LOGGER.info("Promoted inferred session to real ID: %s", real_sid)
		
		# 2. Link any unmatched events for this plate to the new session
		_link_unmatched_events_to_session(plate_ar, real_sid)
		
		# 3. Use the REAL session ID for the rest of the flow
		session_id = real_sid
		session_row = new_session
	else:
		session_row = _fetch_parking_session_by_id(session_id)
		if not isinstance(session_row, dict):
			raise HTTPException(status_code=404, detail="parking session was not found.")

	if parking_location:
		_set_parking_session_location(session_id, parking_location)
		session_row = _fetch_parking_session_by_id(session_id) or session_row

	pricing = _compute_session_time_and_fee(session_row)

	amount_raw = payload.get("amount_cents")
	if amount_raw is None:
		amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)
	else:
		try:
			amount_cents = int(amount_raw)
		except Exception:
			raise HTTPException(status_code=400, detail="amount_cents must be a positive integer.")
		if amount_cents <= 0:
			raise HTTPException(status_code=400, detail="amount_cents must be greater than zero.")

	currency = str(payload.get("currency") or PAYMOB_CURRENCY).strip().upper() or PAYMOB_CURRENCY
	manual_reference = str(payload.get("manual_reference") or f"cash-{uuid4().hex[:18]}").strip()
	manual_reference = manual_reference or f"cash-{uuid4().hex[:18]}"

	already_paid = _parse_iso_datetime(session_row.get("payment_confirmed_at")) is not None
	force_exit = bool(payload.get("force_exit", False))
	
	if not already_paid:
		_mark_parking_session_paid(session_id, payment_reference=manual_reference)
	
	if force_exit:
		now_str = datetime.now(timezone.utc).isoformat()
		_update_parking_session(session_id, {
			"status": "exited",
			"check_out_at": now_str,
			"left_within_5_minutes": True,
			"notes": f"Staff cash-confirm checkout ({requester_id})"
		})

		plate_label = _resolve_plate_label_for_session(session_row)
		if not plate_label:
			plate_label = _normalize_spaces(str(payload.get("plate_arabic") or payload.get("plate") or ""))
		if plate_label:
			_notification_title = "✅ Cash Payment Received"
			_notification_body = f"تم استلام الدفع كاش للعربية {plate_label} — تم فتح البوابة"
			_notify_security_staff(
				title=_notification_title,
				body=_notification_body,
				data={
					"session_id": session_id,
					"plate": plate_label,
					"payment_method": "cash",
				},
			)

	# Force checkout for manual cash confirmations (Guard Hub workflow)
	_update_parking_session(session_id, {
		"check_out_at": datetime.now(timezone.utc).isoformat(),
		"status": "exited",
		"left_within_5_minutes": True
	})

	payment_transaction_id: Optional[str] = None
	if not already_paid:
		payment_transaction_id = _insert_payment_transaction_record(
			created_by=requester_id,
			session_id=session_id,
			amount_cents=amount_cents,
			currency=currency,
			merchant_order_id=f"cash-{session_id[:8]}-{uuid4().hex[:12]}",
			paymob_order_id=None,
			parking_location=parking_location,
			request_payload={
				"session_id": session_id,
				"amount_cents": amount_cents,
				"currency": currency,
				"manual_reference": manual_reference,
				"parking_location": parking_location,
			},
			response_payload={
				"manual_reference": manual_reference,
				"confirmed_by": requester_id,
				"confirmed_by_role": requester_role,
			},
			provider="cash",
			status="paid",
			payment_reference=manual_reference,
		)

	updated_session = _fetch_parking_session_by_id(session_id) or session_row
	updated_pricing = _compute_session_time_and_fee(updated_session)

	return {
		"status": "ok",
		"provider": "cash",
		"already_paid": already_paid,
		"session_id": session_id,
		"payment_transaction_id": payment_transaction_id,
		"amount_cents": amount_cents,
		"currency": currency,
		"payment_reference": manual_reference,
		"pricing": updated_pricing,
		"payment_status": _derive_payment_status_from_session(updated_session),
		"gate_decision": _build_gate_decision(
			event_type="exit",
			session_row=updated_session,
			plate_detected=True,
		),
	}


@app.post("/payments/demo/confirm")
def demo_confirm_payment(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)

	session_id = str(payload.get("session_id") or "").strip()
	if not session_id:
		raise HTTPException(status_code=400, detail="session_id is required.")

	session_row = _fetch_parking_session_by_id(session_id)
	if not isinstance(session_row, dict):
		raise HTTPException(status_code=404, detail="parking session was not found.")

	owner_id = _get_parking_session_owner_id(session_id)
	if owner_id and not requester_is_staff and owner_id != requester_id:
		raise HTTPException(status_code=403, detail="You can only confirm payments for your own sessions.")

	already_paid = _parse_iso_datetime(session_row.get("payment_confirmed_at")) is not None
	if not already_paid:
		_mark_parking_session_paid(session_id)
		pricing = _compute_session_time_and_fee(session_row)
		amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)
		_insert_payment_transaction_record(
			created_by=requester_id,
			session_id=session_id,
			amount_cents=amount_cents,
			currency=PAYMOB_CURRENCY,
			merchant_order_id=f"demo-{session_id[:8]}-{uuid4().hex[:10]}",
			paymob_order_id=None,
			parking_location=_normalize_parking_location(session_row.get("parking_location")),
			request_payload={
				"session_id": session_id,
				"provider": "demo_online",
			},
			response_payload={
				"status": "paid",
			},
			provider="demo_online",
			status="paid",
		)

		if owner_id:
			_insert_notification_event(
				user_id=owner_id,
				event_type="payment_confirmed",
				title="Payment Confirmed :white_check_mark:",
				body="تم تأكيد دفعك — عندك 5 دقايق للخروج",
				data={
					"session_id": session_id,
					"payment_method": "demo_online",
				},
			)

	updated_session = _fetch_parking_session_by_id(session_id) or session_row
	return {
		"session_id": session_id,
		"payment_status": _derive_payment_status_from_session(updated_session),
		"gate_decision": _build_gate_decision(
			event_type="exit",
			session_row=updated_session,
			plate_detected=True,
		),
		"left_within_5_minutes": bool(updated_session.get("left_within_5_minutes")),
	}


@app.post("/payments/cash/request")
def request_cash_payment(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)

	session_id = str(payload.get("session_id") or "").strip()
	if not session_id:
		raise HTTPException(status_code=400, detail="session_id is required.")

	session_row = _fetch_parking_session_by_id(session_id)
	if not isinstance(session_row, dict):
		raise HTTPException(status_code=404, detail="parking session was not found.")

	owner_id = _get_parking_session_owner_id(session_id)
	if owner_id and not requester_is_staff and owner_id != requester_id:
		raise HTTPException(status_code=403, detail="You can only request cash for your own sessions.")

	profile = _supabase_get_profile_by_user_id(requester_id) or {}
	requester_name = _normalize_spaces(
		str(profile.get("full_name") or profile.get("username") or profile.get("first_name") or "")
	)
	if not requester_name:
		requester_name = "اليوزر"

	plate_label = _resolve_plate_label_for_session(session_row)
	if not plate_label:
		plate_label = _normalize_spaces(str(payload.get("plate_arabic") or payload.get("plate") or ""))
	if not plate_label:
		plate_label = "غير معروف"

	_notification_title = ":dollar: Cash Payment Needed"
	_notification_body = f"اليوزر {requester_name} عايز يدفع كاش — {plate_label}"
	_notify_security_staff(
		title=_notification_title,
		body=_notification_body,
		data={
			"session_id": session_id,
			"plate": plate_label,
			"requester_id": requester_id,
			"payment_method": "cash",
		},
	)

	old_notes = session_row.get("notes") or ""
	new_notes = old_notes
	if "intent_cash" not in old_notes:
		new_notes = f"{old_notes} | intent_cash".strip(" |")
		
	update_payload = {"notes": new_notes}
	if session_row.get("status") == "paid":
		update_payload["status"] = "entered"
		update_payload["payment_confirmed_at"] = None
		update_payload["notes"] = f"{new_notes} | switched_to_cash".strip(" |")
		
	_update_parking_session(session_id, update_payload)

	return {
		"status": "notified",
		"message": "تم إبلاغ الأمن",
	}


@app.post("/gate/decision")
def gate_decision(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		raise HTTPException(status_code=401, detail="Authenticated user id is missing.")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)
	requester_has_global_scope = _can_view_global_records(requester_role)

	session_id = str(payload.get("session_id") or "").strip()
	if not session_id:
		raise HTTPException(status_code=400, detail="session_id is required.")

	plate_detected = bool(payload.get("plate_detected", True))
	session_row = _fetch_parking_session_by_id(session_id)
	if not isinstance(session_row, dict):
		raise HTTPException(status_code=404, detail="parking session was not found.")

	owner_id = _get_parking_session_owner_id(session_id)
	if not requester_has_global_scope and owner_id and requester_id != owner_id:
		raise HTTPException(status_code=403, detail="You can only access your own session gate decision.")

	decision = _build_gate_decision(
		event_type="exit",
		session_row=session_row,
		plate_detected=plate_detected,
	)

	return {
		"status": "ok",
		"session_id": session_id,
		"payment_status": _derive_payment_status_from_session(session_row),
		"pricing": _compute_session_time_and_fee(session_row),
		"gate_decision": decision,
		"requester": {
			"id": requester_id,
			"role": requester_role or "user",
			"is_staff": requester_is_staff,
		},
	}


@app.post("/payments/paymob/create")
def create_paymob_payment(
	payload: Dict[str, Any] = Body(...),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	if not _paymob_configured():
		return _error_response(
			503,
			"Payment gateway not configured",
			"PAYMOB_NOT_CONFIGURED",
		)

	try:
		request_user = _require_authenticated_user(authorization)
	except HTTPException as exc:
		return _error_response(exc.status_code, str(exc.detail), "AUTH_REQUIRED")

	if request_user is None:
		return _error_response(401, "Authentication is required.", "AUTH_REQUIRED")

	requester_id = str(request_user.get("id") or "").strip()
	if not requester_id:
		return _error_response(401, "Authenticated user id is missing.", "AUTH_REQUIRED")

	requester_role = _resolve_requester_role(request_user)
	requester_is_staff = _is_staff_role(requester_role)

	session_id = str(payload.get("session_id") or "").strip()
	if not session_id:
		return _error_response(400, "session_id is required.", "SESSION_ID_REQUIRED")

	session_row = _fetch_parking_session_by_id(session_id)
	if not isinstance(session_row, dict):
		return _error_response(404, "parking session was not found.", "SESSION_NOT_FOUND")

	parking_location = _normalize_parking_location(payload.get("parking_location"))
	pricing = _compute_session_time_and_fee(session_row, parking_location_override=parking_location)
	amount_raw = payload.get("amount_cents")
	if amount_raw is None:
		amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)
	else:
		try:
			amount_cents = int(amount_raw)
		except Exception:
			return _error_response(400, "amount_cents must be a positive integer.", "INVALID_AMOUNT")
		if amount_cents <= 0:
			amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)

	currency = str(payload.get("currency") or PAYMOB_CURRENCY).strip().upper() or PAYMOB_CURRENCY

	owner_id = _get_parking_session_owner_id(session_id)
	if not owner_id:
		return _error_response(404, "parking session was not found.", "SESSION_NOT_FOUND")

	if not requester_is_staff and requester_id != owner_id:
		return _error_response(
			403,
			"You can only create payments for your own sessions.",
			"FORBIDDEN_SESSION",
		)

	if _parse_iso_datetime(session_row.get("payment_confirmed_at")) is not None:
		return _error_response(409, "Session is already paid.", "ALREADY_PAID")

	try:
		if parking_location:
			_set_parking_session_location(session_id, parking_location)

		merchant_order_id = f"ain-{session_id[:8]}-{uuid4().hex[:12]}"
		checkout = _paymob_create_checkout(
			amount_cents=amount_cents,
			currency=currency,
			merchant_order_id=merchant_order_id,
			request_user=request_user,
		)

		payment_id = _insert_payment_transaction_record(
			created_by=requester_id,
			session_id=session_id,
			amount_cents=amount_cents,
			currency=currency,
			merchant_order_id=merchant_order_id,
			paymob_order_id=checkout.get("paymob_order_id"),
			parking_location=parking_location,
			request_payload={
				"session_id": session_id,
				"amount_cents": amount_cents,
				"currency": currency,
				"merchant_order_id": merchant_order_id,
				"parking_location": parking_location,
			},
			response_payload={
				"paymob_order_id": checkout.get("paymob_order_id"),
				"iframe_url": checkout.get("iframe_url"),
			},
		)
		return {
			"status": "ok",
			"provider": "paymob",
			"payment_transaction_id": payment_id,
			"merchant_order_id": merchant_order_id,
			"paymob_order_id": checkout.get("paymob_order_id"),
			"session_id": session_id,
			"amount_cents": amount_cents,
			"currency": currency,
			"pricing": pricing,
			"parking_location": parking_location,
			"payment_url": checkout.get("iframe_url"),
		}
	except HTTPException as exc:
		return _error_response(exc.status_code, str(exc.detail), "PAYMOB_CREATE_FAILED")
	except Exception as exc:
		return _error_response(502, f"Failed to create Paymob payment: {exc}", "PAYMOB_CREATE_FAILED")


@app.post("/payments/paymob/webhook")
def paymob_webhook(payload: Dict[str, Any] = Body(...)) -> Dict[str, Any]:
	obj = payload.get("obj") if isinstance(payload, dict) else None
	if not isinstance(obj, dict):
		obj = payload if isinstance(payload, dict) else {}

	order_obj = obj.get("order") if isinstance(obj.get("order"), dict) else {}
	merchant_order_id = str(order_obj.get("merchant_order_id") or payload.get("merchant_order_id") or "").strip()
	if not merchant_order_id:
		raise HTTPException(status_code=400, detail="merchant_order_id was not found in webhook payload.")

	paymob_order_id = str(order_obj.get("id") or obj.get("order_id") or payload.get("order_id") or "").strip() or None
	payment_reference = str(obj.get("id") or payload.get("id") or "").strip() or None
	is_pending = bool(obj.get("pending"))
	is_success = bool(obj.get("success")) and not is_pending
	status = "pending"
	if is_success:
		status = "paid"
	elif not is_pending:
		status = "failed"

	try:
		_update_payment_transaction_by_merchant_order_id(
			merchant_order_id=merchant_order_id,
			status=status,
			payment_reference=payment_reference,
			paymob_order_id=paymob_order_id,
			response_payload=payload if isinstance(payload, dict) else {},
		)

		session_id = _get_payment_session_id_by_merchant_order_id(merchant_order_id)
		if is_success and session_id:
			_mark_parking_session_paid(session_id, payment_reference=payment_reference)
		return {
			"status": "ok",
			"payment_status": status,
			"merchant_order_id": merchant_order_id,
			"paymob_order_id": paymob_order_id,
			"session_id": session_id,
		}
	except HTTPException:
		raise
	except Exception as exc:
		raise HTTPException(status_code=502, detail=f"Failed to process Paymob webhook: {exc}") from exc


@app.post("/admin/maintenance/weekly-refresh")
def admin_weekly_refresh(
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	if request_user is None:
		raise HTTPException(status_code=401, detail="Authentication is required.")

	requester_id = str(request_user.get("id") or "").strip()
	requester_role = _resolve_requester_role(request_user)
	if not _is_staff_role(requester_role):
		raise HTTPException(status_code=403, detail="Only admin/security can run weekly refresh.")

	try:
		refresh_result = _run_weekly_refresh_non_core()
		return {
			"status": "ok",
			"requested_by": requester_id,
			"requested_by_role": requester_role,
			"core_tables_protected": sorted(CORE_TABLES),
			"refresh": refresh_result,
		}
	except Exception as exc:
		raise HTTPException(status_code=502, detail=f"Weekly refresh failed: {exc}") from exc


@app.post("/predict")
async def predict(
	image: UploadFile = File(...),
	event_type: Optional[str] = Form(default="entry"),
	parking_location: Optional[str] = Form(default=None),
	camera_source: Optional[str] = Form(default=None),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user = _require_authenticated_user(authorization)
	normalized_parking_location = _normalize_parking_location(parking_location)
	normalized_camera_source = _normalize_camera_source(camera_source)
	normalized_event_type = _normalize_event_type(event_type)

	if image.content_type and not image.content_type.startswith("image/") and image.content_type != "application/octet-stream":
		raise HTTPException(status_code=400, detail="Only image uploads are supported.")

	image_bytes = await image.read()
	if not image_bytes:
		raise HTTPException(status_code=400, detail="Uploaded file is empty.")

	# Always use remote HuggingFace Space for inference
	response = _run_pipeline_remote(
		image_bytes=image_bytes,
		filename=image.filename,
		content_type=image.content_type,
	)
	if not isinstance(response, dict):
		raise HTTPException(status_code=502, detail="Inference pipeline returned an invalid payload.")
	response["inference_provider"] = "remote"
	response["filename"] = image.filename
	response["event_type"] = normalized_event_type
	response["parking_location"] = normalized_parking_location
	response["camera_source"] = normalized_camera_source
	plate_payload = _build_plate_payload(response.get("plate_info", {}))
	response["plate_payload"] = plate_payload
	response["plate"] = plate_payload.get("arabic_text") or "N/A"
	response["plate_text_ar"] = plate_payload.get("arabic_text") or "N/A"
	response["plate_text_en"] = plate_payload.get("english_text") or "N/A"
	# Flat plate fields for Flutter
	_ar_parts = plate_payload.get("arabic") if isinstance(plate_payload.get("arabic"), dict) else {}
	_en_parts = plate_payload.get("english") if isinstance(plate_payload.get("english"), dict) else {}
	response["plate_letters_ar"] = _ar_parts.get("letters") or ""
	response["plate_numbers_ar"] = _ar_parts.get("numbers") or ""
	response["plate_letters_en"] = _en_parts.get("letters") or ""
	response["plate_numbers_en"] = _en_parts.get("numbers") or ""
	original_b64 = base64.b64encode(image_bytes).decode("utf-8")
	response["processed_image_base64"] = str(
		(response.get("user_page", {}) or {}).get("split_image_base64") or original_b64
	)
	response["annotated_image_base64"] = str(
		(response.get("admin_page", {}) or {}).get("annotated_image_base64") or original_b64
	)
	if request_user:
		response["request_user"] = {
			"id": request_user.get("id"),
			"email": request_user.get("email"),
		}

	response["supabase"] = _persist_prediction_to_supabase(
		original_image_bytes=image_bytes,
		response_payload=response,
		request_user=request_user,
		parking_location=normalized_parking_location,
		event_type=normalized_event_type,
		camera_source=normalized_camera_source,
	)

	supabase_payload = response.get("supabase") if isinstance(response.get("supabase"), dict) else {}
	response["app_registration"] = supabase_payload.get(
		"app_registration",
		{
			"is_registered": False,
			"vehicle_id": None,
			"owner_id": None,
		},
	)
	response["new_car_alert"] = supabase_payload.get(
		"new_car_alert",
		{
			"is_new_car": None,
			"notify_roles": ["admin", "security"],
			"action": "lookup_unavailable",
			"reason": "supabase_unavailable",
			"message": "Could not verify registration status.",
			"vehicle_id": None,
			"owner_id": None,
			"plate_arabic": None,
			"plate_english": None,
		},
	)
	response["payment_status"] = supabase_payload.get("payment_status", "unknown")
	response["left_within_5_minutes"] = bool(supabase_payload.get("left_within_5_minutes", False))
	response["pricing"] = supabase_payload.get(
		"pricing",
		_compute_session_time_and_fee(None, parking_location_override=normalized_parking_location),
	)
	response["gate_decision"] = supabase_payload.get(
		"gate_decision",
		_build_gate_decision(
			event_type=normalized_event_type,
			session_row=None,
			plate_detected=_is_plate_detected(response.get("plate_info", {})),
		),
	)
	response["session_id"] = supabase_payload.get("session_id")
	response["event_id"] = supabase_payload.get("event_id")

	# === DEBUG LOGS (remove after bugs confirmed fixed) ===
	LOGGER.info("=== PREDICT START ===")
	LOGGER.info("event_type: %s", normalized_event_type)
	LOGGER.info("plate_ar: %s", response.get("plate_text_ar"))
	LOGGER.info("plate_en: %s", response.get("plate_text_en"))
	LOGGER.info("vehicle_found: %s", response.get("app_registration", {}).get("is_registered"))
	LOGGER.info("vehicle_id: %s", response.get("app_registration", {}).get("vehicle_id"))
	LOGGER.info("session_id: %s", response.get("session_id"))
	LOGGER.info("is_registered: %s", response.get("app_registration", {}).get("is_registered"))
	LOGGER.info("=== PREDICT END ===")

	return response


@app.post("/developer/predict")
async def developer_predict(
	image: UploadFile = File(...),
	event_type: Optional[str] = Form(default="entry"),
	parking_location: Optional[str] = Form(default=None),
	camera_source: Optional[str] = Form(default=None),
	authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
	request_user, requester_role = _require_developer_user(authorization)
	response = await predict(
		image=image,
		event_type=event_type,
		parking_location=parking_location,
		camera_source=camera_source,
		authorization=authorization,
	)
	response["developer_test"] = {
		"enabled": True,
		"request_user": {
			"id": request_user.get("id"),
			"email": request_user.get("email"),
			"role": requester_role,
		},
	}
	return response


# ─── ESP32 Gate Command Queue ────────────────────────────────────
# In-memory command queue for ESP32 gate controllers.
# POST /gate/trigger  → queues an "open" command (called by security dashboard)
# GET  /gate/poll     → ESP32 polls this; returns & clears the pending command

_gate_command_lock = Lock()
_gate_pending_commands: List[Dict[str, Any]] = []


@app.post("/gate/trigger")
def gate_trigger(
	payload: Dict[str, Any] = Body(default={}),
) -> Dict[str, Any]:
	"""Queue a gate-open command for ESP32 devices to pick up."""
	location = str(payload.get("parking_location") or "OPERA").strip().upper()
	duration_sec = int(payload.get("duration_seconds") or 10)
	plate = str(payload.get("plate") or "").strip()

	cmd = {
		"action": "open",
		"parking_location": location,
		"duration_seconds": duration_sec,
		"plate": plate,
		"queued_at": datetime.now(timezone.utc).isoformat(),
	}

	with _gate_command_lock:
		_gate_pending_commands.append(cmd)
		# Keep only last 20 commands to prevent memory leak
		if len(_gate_pending_commands) > 20:
			_gate_pending_commands.pop(0)

	LOGGER.info("Gate trigger queued: %s", cmd)
	return {"status": "queued", "command": cmd}


@app.get("/gate/poll")
def gate_poll(
	location: str = "OPERA",
) -> Dict[str, Any]:
	"""ESP32 polls this endpoint. Returns the oldest pending command and removes it."""
	loc = location.strip().upper()

	with _gate_command_lock:
		for i, cmd in enumerate(_gate_pending_commands):
			if cmd.get("parking_location") == loc:
				_gate_pending_commands.pop(i)
				return {"status": "command", "command": cmd}

	return {"status": "idle"}



if __name__ == "__main__":
	import uvicorn

	uvicorn.run(
		"API:app",
		host=os.getenv("API_HOST", "0.0.0.0"),
		port=int(os.getenv("PORT", os.getenv("API_PORT", "8000"))),
		reload=False,
	)