import asyncio import time as _time from typing import Optional, Any import firebase_admin from firebase_admin import credentials, firestore as fs from google.cloud.firestore_v1.base_query import FieldFilter from app.config import settings from app.internal.logger import logger # Re-exported so callers never need their own `firebase_admin.firestore` import # just to delete a field. server-26#96/#114 review: `doc_set(..., merge=True)` # merges nested maps by key but can never REMOVE one — writing `{"scenes": {}}` # to clear a map is a no-op, not a delete. Use `doc_update(coll, id, {"field": # fstore.DELETE_FIELD})` (or doc_set + merge, DELETE_FIELD works under both) # whenever a re-extraction/reprocess path needs a stale nested field gone # rather than merged over. DELETE_FIELD = fs.DELETE_FIELD # --------------------------------------------------------------------------- # In-memory TTL cache for rarely-changing documents (systems, nodes config) # --------------------------------------------------------------------------- # Key: "collection/doc_id" → (expires_at_monotonic, data_or_None) _doc_cache: dict[str, tuple[float, Optional[dict]]] = {} def _init_firebase(): if firebase_admin._apps: return firestore.client() if settings.gcp_credentials_path: cred = credentials.Certificate(settings.gcp_credentials_path) else: cred = credentials.ApplicationDefault() firebase_admin.initialize_app(cred) logger.info("Firebase initialised.") _init_firebase() db = fs.client(database_id=settings.firestore_database) # --------------------------------------------------------------------------- # Thin async wrappers — firebase-admin is synchronous, run in thread executor # --------------------------------------------------------------------------- async def doc_set(collection: str, doc_id: str, data: dict, merge: bool = True) -> None: ref = db.collection(collection).document(doc_id) await asyncio.to_thread(ref.set, data, merge=merge) async def doc_get(collection: str, doc_id: str) -> Optional[dict]: ref = db.collection(collection).document(doc_id) snap = await asyncio.to_thread(ref.get) return snap.to_dict() if snap.exists else None async def doc_update(collection: str, doc_id: str, data: dict) -> None: ref = db.collection(collection).document(doc_id) await asyncio.to_thread(ref.update, data) async def collection_list(collection: str, **filters) -> list[dict]: """ List all documents in a collection. Optional keyword filters: field=value pairs passed as equality where-clauses. """ def _query(): ref = db.collection(collection) for field, value in filters.items(): ref = ref.where(filter=FieldFilter(field, "==", value)) return [doc.to_dict() for doc in ref.stream()] return await asyncio.to_thread(_query) async def collection_where( collection: str, conditions: list[tuple[str, str, Any]], order_by: Optional[list[tuple[str, str]]] = None, limit_to: Optional[int] = None, start_after: Optional[dict] = None, ) -> list[dict]: """ Query a collection with arbitrary where-clauses. conditions: list of (field, op, value) — e.g. [("ended_at", ">=", cutoff_dt)] Supports any Firestore operator, including "array_contains" — it's just forwarded straight to FieldFilter, so a condition like ("incident_ids", "array_contains", incident_id) already worked before this function grew explicit order_by/limit/cursor params below. order_by: list of (field, direction) — direction is "ASCENDING" or "DESCENDING" (Firestore's own constants; passed straight through as strings so this module doesn't need a google.cloud.firestore_v1.Query import). Applied in list order, so multi-field sorts work. limit_to: cap the number of documents returned. start_after: cursor — a dict of the same field values as the *last* document from a previous page's order_by fields (Firestore's `Query.start_after()` takes a field-value mapping, not a document snapshot, when you're not holding one). Added for org_id-scoped queries that also need to be ordered/paginated — unscoped equality-only lookups can keep using collection_list(). """ def _query(): ref = db.collection(collection) for field, op, value in conditions: ref = ref.where(filter=FieldFilter(field, op, value)) for field, direction in (order_by or []): ref = ref.order_by(field, direction=direction) if start_after is not None: ref = ref.start_after(start_after) if limit_to is not None: ref = ref.limit(limit_to) return [doc.to_dict() for doc in ref.stream()] return await asyncio.to_thread(_query) async def doc_delete(collection: str, doc_id: str) -> None: ref = db.collection(collection).document(doc_id) await asyncio.to_thread(ref.delete) async def doc_get_cached(collection: str, doc_id: str, ttl: float = 300.0) -> Optional[dict]: """ Like doc_get but backed by a short-lived in-memory TTL cache. Use for documents that change rarely (systems config, node assignments). Default TTL is 5 minutes — a write will be visible within that window. """ key = f"{collection}/{doc_id}" now = _time.monotonic() entry = _doc_cache.get(key) if entry and now < entry[0]: return entry[1] data = await doc_get(collection, doc_id) _doc_cache[key] = (now + ttl, data) return data