Blame for backend/core/database.py

import time

from pymongo import MongoClient
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance
from .config import settings
from ..core.logger import logger


# MongoDB
class MongoDB:
    client: MongoClient = None
    db = None
    users_col = None
    prompts_col = None
    saved_prompts_col = None
    feedback_col = None
    analytics_col = None
    
    @classmethod
    def connect(cls):
        try:
            cls.client = MongoClient(
                settings.MONGO_URI or "mongodb://localhost:27017",
                serverSelectionTimeoutMS=3000,
            )
            cls.client.admin.command("ping")
            cls.db = cls.client["prompt_engine_db"]
            cls.users_col = cls.db["users"]
            cls.prompts_col = cls.db["prompt_logs"]
            cls.saved_prompts_col = cls.db["saved_prompts"]
            cls.feedback_col = cls.db["user_feedback"]
            cls.analytics_col = cls.db["analytics_events"]

            # Indexes
            cls.users_col.create_index("user_id", unique=True)
            cls.prompts_col.create_index([("user_id", 1), ("timestamp", -1)])
            # Acceptance resolves an opaque log ID inside one user's history.
            # Keep this non-unique for compatibility with older log documents
            # that have no log_id at all.
            cls.prompts_col.create_index([("user_id", 1), ("log_id", 1)])
            # The private dashboard queries all users inside a time window, so
            # the existing user_id-first index cannot serve it efficiently.
            cls.prompts_col.create_index([("timestamp", -1)])
            cls.saved_prompts_col.create_index("user_id")
            cls.feedback_col.create_index([("user_id", 1), ("timestamp", -1)])
            cls.analytics_col.create_index([("timestamp", -1)])
            cls.analytics_col.create_index([("event", 1), ("timestamp", -1)])

            # get_user_feedback_summary() reads db["prompt_feedback"], which is a
            # DIFFERENT collection from feedback_col (db["user_feedback"]) above.
            # It was never indexed, so every /enhance ran an unindexed scan plus
            # an in-memory sort against it.
            cls.db["prompt_feedback"].create_index([("user_id", 1), ("timestamp", -1)])
            cls.db["prompt_feedback"].create_index([("timestamp", -1)])

            # Retention. Atlas M0 is 512 MB and nothing ever deleted a prompt
            # log, so the cluster filled and then failed writes silently.
            # expireAfterSeconds on a date field lets Mongo do the pruning.
            if settings.PROMPT_LOG_TTL_DAYS > 0:
                _ttl = settings.PROMPT_LOG_TTL_DAYS * 86400
                cls._ensure_ttl(cls.prompts_col, "timestamp", _ttl)
                cls._ensure_ttl(cls.db["prompt_feedback"], "timestamp", _ttl)

            if settings.ANALYTICS_EVENT_TTL_DAYS > 0:
                cls._ensure_ttl(
                    cls.analytics_col,
                    "timestamp",
                    settings.ANALYTICS_EVENT_TTL_DAYS * 86400,
                )

            logger.info("✅ MongoDB Indexes Verified")
            logger.info("✅ MongoDB Connected")
        except Exception as e:
            logger.warning(f"⚠️ MongoDB not available ({e}) — using in-memory fallback.")
            # db and client must be cleared too. Leaving them set produced a
            # split brain: feedback writes (which test `MongoDB.db is not None`)
            # kept targeting a dead client while everything else fell back to
            # RAM, so the two halves of a request disagreed about reality.
            cls.client = None
            cls.db = None
            cls.users_col = None
            cls.prompts_col = None
            cls.saved_prompts_col = None
            cls.feedback_col = None
            cls.analytics_col = None

    @staticmethod
    def _ensure_ttl(col, field: str, seconds: int):
        """Create a TTL index, tolerating one that already exists with a different span."""
        try:
            col.create_index(field, expireAfterSeconds=seconds)
        except Exception as e:
            logger.warning(f"⚠️ TTL index on {col.name}.{field} skipped: {e}")

# Qdrant
class QdrantDB:
    """
    Vector store access, with the failure modes made survivable.

    Every one of the behaviours below caused a real, months-long outage that was
    invisible from outside: the saved-prompt library appeared to save correctly
    and silently never matched anything, because a dead cluster and an empty
    result set were indistinguishable at every layer.
    """

    client: QdrantClient = None
    _collections_ready = False
    _last_error: str = None
    _last_attempt: float = 0.0
    _connected_at: float = None

    SAVED_COLLECTION = "saved_prompt_vectors"

    # Must match the embedding model's output width. get_embedding() uses
    # paraphrase-multilingual-MiniLM-L12-v2, which is 384-dimensional; a
    # collection built for a different width rejects every upsert.
    VECTOR_SIZE = 384

    @classmethod
    def get_client(cls):
        """
        A client that has actually talked to the cluster, or None.

        Previously this returned any successfully *constructed* client.
        QdrantClient does not dial on construction, so a cluster that was
        deleted, suspended or unreachable still produced a client object and a
        "Connected" log line, and every subsequent operation failed into an
        exception handler that returned an empty list.
        """
        if cls.client is not None and cls._collections_ready:
            return cls.client

        # Back off between attempts: retrying on every request hammers a
        # struggling cluster, never retrying leaves a transient blip permanent.
        now = time.monotonic()
        if cls.client is None and cls._last_attempt and \
                (now - cls._last_attempt) < settings.QDRANT_RETRY_SECONDS:
            return None
        cls._last_attempt = now

        if cls.client is None:
            try:
                candidate = QdrantClient(
                    url=settings.QDRANT_URL,
                    api_key=settings.QDRANT_API_KEY,
                    timeout=settings.QDRANT_TIMEOUT,
                )
                # The connectivity check that was missing. Cheap, and the only
                # thing that distinguishes a live cluster from a dead one.
                candidate.get_collections()
                cls.client = candidate
                cls._last_error = None
                cls._connected_at = time.time()
                logger.info(f"✅ Qdrant connected ({cls._host_only()})")
            except Exception as e:
                cls.client = None
                cls._collections_ready = False
                cls._last_error = f"{type(e).__name__}: {e}"[:200]
                logger.error(f"❌ Qdrant unreachable ({cls._host_only()}): {cls._last_error}")
                logger.info("   Saved-prompt search and passive memory are disabled until it recovers.")
                return None

        if not cls._collections_ready:
            # Only latch when provisioning actually succeeded. This used to be
            # set unconditionally, so one failed startup left the collections
            # uncreated for the life of the process.
            results = [cls._ensure_collection(settings.COLLECTION_NAME),
                       cls._ensure_collection(cls.SAVED_COLLECTION)]
            cls._collections_ready = all(results)
            if not cls._collections_ready:
                logger.warning("⚠️ Qdrant collections not ready — will retry on the next call.")

        return cls.client

    @classmethod
    def _host_only(cls) -> str:
        """Host without credentials, for logs."""
        raw = settings.QDRANT_URL or ""
        return raw.split("://")[-1].split("/")[0] or raw

    @classmethod
    def reset(cls):
        """Drop the cached client so the next call reconnects."""
        cls.client = None
        cls._collections_ready = False
        cls._last_attempt = 0.0

    @classmethod
    def _ensure_collection(cls, name: str) -> bool:
        """
        Make sure `name` exists with the right geometry. Returns success.

        Returned None before, and its caller ignored it either way.
        """
        try:
            info = cls.client.get_collection(name)
            size = cls._configured_size(info)
            if size is not None and size != cls.VECTOR_SIZE:
                # Loud on purpose: every upsert would be rejected, and the
                # symptom is once again "search returns nothing".
                cls._last_error = (
                    f"collection '{name}' is {size}-dim, embeddings are "
                    f"{cls.VECTOR_SIZE}-dim"
                )
                logger.error(f"❌ {cls._last_error}. Recreate the collection or change the model.")
                return False
        except Exception:
            try:
                cls.client.create_collection(
                    collection_name=name,
                    vectors_config=VectorParams(size=cls.VECTOR_SIZE, distance=Distance.COSINE),
                )
                logger.info(f"✅ Created Qdrant collection '{name}'")
            except Exception as e:
                cls._last_error = f"create '{name}' failed: {e}"[:200]
                logger.warning(f"⚠️ {cls._last_error}")
                return False

        # Filtering is by user_id on every search and by mongo_id on delete.
        # Unindexed payload filters still work but scan, so this is performance,
        # not correctness — a failure here should not fail provisioning.
        for field in ("user_id", "mongo_id"):
            try:
                cls.client.create_payload_index(
                    collection_name=name, field_name=field, field_schema="keyword",
                )
            except Exception:
                pass
        return True

    @staticmethod
    def _configured_size(info):
        """Vector width from a collection description, across client versions."""
        try:
            vectors = info.config.params.vectors
            return getattr(vectors, "size", None) if not isinstance(vectors, dict) else None
        except Exception:
            return None

    @classmethod
    def health(cls) -> dict:
        """
        Whether the vector store is actually usable, and what is in it.

        Saved-prompt search failed in production with no way to tell why from
        outside. The embedding model turned out to be fine, which left Qdrant —
        but "is it connected, do the collections exist, is anything in them"
        was unanswerable without shell access to the Space. Point counts are
        the decisive signal: an empty collection means writes never landed, a
        populated one means the query or its filter is at fault.

        Deliberately reports no URL or key.
        """
        # Structural facts about the configured URL, never the URL itself: a
        # missing :6333 port is the most common cause of "connection reset by
        # peer" against Qdrant Cloud, and it is indistinguishable from a dead
        # cluster without knowing which of the two you are looking at.
        raw = (settings.QDRANT_URL or "").strip()
        host_part = raw.split("://")[-1]
        out = {
            "connected": False,
            "collections": {},
            "error": None,
            "config": {
                "configured": bool(raw) and raw != ":memory:",
                "in_memory": raw == ":memory:",
                "scheme": raw.split("://")[0] if "://" in raw else None,
                "has_port": ":" in host_part.split("/")[0],
                "looks_like_cloud": "cloud.qdrant.io" in raw,
                "api_key_set": bool(settings.QDRANT_API_KEY),
            },
        }
        try:
            client = cls.get_client()
            out["last_error"] = cls._last_error
            out["collections_ready"] = cls._collections_ready
            if client is None:
                out["error"] = cls._last_error or "client unavailable"
                return out
            out["connected"] = True
            for name in (settings.COLLECTION_NAME, cls.SAVED_COLLECTION):
                try:
                    info = client.get_collection(name)
                    out["collections"][name] = {
                        "exists": True,
                        "points": getattr(info, "points_count", None),
                    }
                except Exception as e:
                    out["collections"][name] = {"exists": False, "error": str(e)[:120]}
        except Exception as e:
            out["error"] = str(e)[:160]
        return out


# In-Memory Fallbacks
in_memory_users = {}
in_memory_prompt_logs = []
in_memory_saved_prompts = {}  # {prompt_id: {doc}}
in_memory_analytics_events = []