import json import logging from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional, Any, Tuple from uuid import uuid4 import requests from fastapi import HTTPException from app.core.config import ( SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_SERVICE_ROLE_KEY, SUPABASE_TIMEOUT_SECONDS, SUPABASE_RAW_BUCKET, SUPABASE_PROCESSED_BUCKET, supabase_configured, PROFILE_SELECT_FIELDS, VEHICLE_SELECT_FIELDS, CORE_TABLES, STAFF_ROLES, PHONE_PATTERN, USERNAME_PATTERN, NATIONAL_ID_PATTERN, INTERNAL_LOGIN_EMAIL_SUFFIX, LOGGER, ) from app.security.auth import ( _supabase_headers, _extract_response_error_message, _normalize_spaces, is_valid_redirect_url, MOBILE_REDIRECT_SCHEME_PATTERN, ) TZ_UTC = timezone.utc 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: 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 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: 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 supabase_find_profile_by_identifier(identifier: Optional[str]) -> Optional[Dict[str, Any]]: if not supabase_configured(): return None normalized = _normalize_spaces(str(identifier or "")) if not normalized: return None lookup_chain: List[Tuple[str, str, bool]] = [] if "@" in normalized: lookup_chain.append(("email", normalized.lower(), True)) lookup_chain.extend([ ("username", normalized, False), ("phone_number", normalized, False), ("national_id", normalized, False), ]) if "@" not in normalized: lookup_chain.append(("email", normalized.lower(), True)) seen = set() for field_name, field_value, ci in lookup_chain: key = (field_name, field_value, ci) if key in seen: continue seen.add(key) profile = supabase_get_profile_by_field(field_name, field_value, case_insensitive=ci) 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." try: 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, ) except requests.RequestException as exc: return None, str(exc) if response.status_code == 200: payload = response.json() if isinstance(payload, dict): return payload, None return None, "Unexpected response payload." error_msg = _extract_response_error_message(response) or f"Login failed with status {response.status_code}." return None, error_msg 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 failed ({response.status_code})") 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" try: 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, ) except requests.RequestException as exc: return False, str(exc) if response.status_code == 200: return True, None return False, f"HTTP {response.status_code}: {response.text[:220]}" 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 ensure_profile_row_exists(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 if supabase_get_profile_by_user_id(user_id): 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, ) except requests.RequestException: pass def upsert_profile(*, user_id: str, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: if not user_id: return None allowed_keys = {"id", "first_name", "last_name", "full_name", "username", "phone_number", "email", "national_id", "role", "staff_id"} upsert_payload: Dict[str, Any] = {"id": user_id} for key in allowed_keys: if key == "id": continue value = payload.get(key) if value is None: continue upsert_payload[key] = _normalize_spaces(str(value)) if isinstance(value, str) else value if len(upsert_payload) == 1: return supabase_get_profile_by_user_id(user_id) response = requests.post( f"{SUPABASE_URL}/rest/v1/profiles", json=upsert_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 upsert profile ({response.status_code})") rows = response.json() if response.text else [] if isinstance(rows, list) and rows and isinstance(rows[0], dict): return rows[0] return supabase_get_profile_by_user_id(user_id)