h

database/db.py

@@ -1,61 +1,101 @@
 """
 database/db.py — Async SQLAlchemy engine and session factory.
 
-ROOT CAUSE OF THE BUG:
-  Supabase uses PgBouncer in **Transaction Pooler** mode.
-  PgBouncer reuses the same underlying DB connection for many different
-  clients. asyncpg caches "prepared statements" per-connection. When
-  PgBouncer hands the same connection to another client, asyncpg tries
-  to create the same prepared statement again and PostgreSQL raises:
-
-      DuplicatePreparedStatementError:
-      prepared statement "__asyncpg_stmt_2__" already exists
-
-THE FIX — use NullPool:
-  SQLAlchemy's NullPool opens a brand-new raw connection for every
-  session and closes it immediately when the session ends.
-  Because every connection is fresh, there is never a leftover prepared
-  statement to conflict with. PgBouncer handles the actual pooling on
-  its side, so SQLAlchemy doesn't need its own pool at all.
-
-  This is the officially recommended approach for Supabase + asyncpg.
+═══════════════════════════════════════════════════════════════════
+ROOT CAUSE (why previous fixes failed):
+═══════════════════════════════════════════════════════════════════
+Even with NullPool + statement_cache_size=0 in connect_args,
+SQLAlchemy's asyncpg *dialect* runs this internally during
+connection setup:
+
+    [SQL: select pg_catalog.version()]
+
+It executes this as a prepared statement using its OWN internal
+mechanism — BEFORE our connect_args are applied to the session.
+PgBouncer sees the second client try to prepare the same statement
+on the recycled connection and raises:
+
+    DuplicatePreparedStatementError:
+    prepared statement "__asyncpg_stmt_2__" already exists
+
+═══════════════════════════════════════════════════════════════════
+THE DEFINITIVE FIX — async_creator:
+═══════════════════════════════════════════════════════════════════
+SQLAlchemy 2.x supports an `async_creator` parameter.
+When provided, SQLAlchemy skips its entire dialect connection
+bootstrap and uses our raw asyncpg connection directly.
+We create that connection with statement_cache_size=0, so no
+prepared statements are EVER created — the conflict is impossible.
+
+This is the only approach that fully resolves the issue with
+Supabase Transaction Pooler (PgBouncer in transaction mode).
 """
 
+import ssl as ssl_module
+import asyncpg
+
 from sqlalchemy.ext.asyncio import (
     AsyncSession,
     create_async_engine,
     async_sessionmaker,
 )
-from sqlalchemy.pool import NullPool          # ← THE KEY FIX
+from sqlalchemy.pool import NullPool
 
 from config import config
 from database.models import Base
 
 
+# ── Build a plain DSN for asyncpg (no SQLAlchemy dialect prefix) ──────────
+# asyncpg.connect() expects "postgresql://..." not "postgresql+asyncpg://..."
+_PLAIN_DSN: str = config.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
+
+# Build an SSL context that requires SSL but doesn't verify the cert
+# (Supabase's cert chain can vary by region; this keeps it simple).
+_SSL_CTX = ssl_module.create_default_context()
+_SSL_CTX.check_hostname = False
+_SSL_CTX.verify_mode = ssl_module.CERT_NONE
+
+
+async def _asyncpg_creator() -> asyncpg.Connection:
+    """
+    Raw asyncpg connection factory passed to SQLAlchemy as async_creator.
+
+    Key parameters:
+      statement_cache_size=0  — disables ALL prepared-statement caching
+                                inside asyncpg. This is the single setting
+                                that PgBouncer's own error message tells us
+                                to use. By using async_creator we guarantee
+                                it applies to *every* connection, including
+                                the internal dialect init queries.
+    """
+    return await asyncpg.connect(
+        dsn=_PLAIN_DSN,
+        ssl=_SSL_CTX,
+        statement_cache_size=0,   # ← THE critical fix
+        timeout=30,
+        command_timeout=60,
+    )
+
+
+# ── Engine ────────────────────────────────────────────────────────────────
 engine = create_async_engine(
+    # The URL still needs to be a valid SQLAlchemy URL so the dialect is
+    # loaded; async_creator overrides the actual connection creation.
     config.DATABASE_URL,
 
-    # ── NullPool: no SQLAlchemy-level connection pooling ─────
-    # PgBouncer (Supabase) is the pool. SQLAlchemy should not
-    # add another caching layer on top of it.
-    poolclass=NullPool,
-
-    # ── asyncpg connection arguments ─────────────────────────
-    connect_args={
-        "ssl"            : "require",   # Mandatory for Supabase
-        "timeout"        : 30,          # Connect timeout (seconds)
-        "command_timeout": 60,          # Query timeout (seconds)
+    # async_creator: bypass SQLAlchemy's dialect connection bootstrap.
+    # Our _asyncpg_creator() is called for every new connection instead.
+    async_creator=_asyncpg_creator,
 
-        # Belt-and-suspenders: also disable the statement cache
-        # inside asyncpg itself (the NullPool already prevents
-        # the conflict, but this makes doubly sure).
-        "statement_cache_size": 0,
-    },
+    # NullPool: don't let SQLAlchemy pool connections.
+    # PgBouncer (Supabase) is the pool — two pools stacked = conflicts.
+    poolclass=NullPool,
 
-    echo=False,   # Set True temporarily to log all SQL for debugging
+    echo=False,  # Set True temporarily to log SQL queries while debugging
 )
 
 
+# ── Session factory ───────────────────────────────────────────────────────
 AsyncSessionFactory = async_sessionmaker(
     bind=engine,
     class_=AsyncSession,
@@ -64,8 +104,10 @@ AsyncSessionFactory = async_sessionmaker(
 )
 
 
+# ── Public helpers ────────────────────────────────────────────────────────
+
 async def init_db() -> None:
-    """Create all tables defined in models.py if they don't exist yet."""
+    """Create all tables defined in models.py (idempotent — safe to re-run)."""
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
     print("✅ Database tables initialized.")
@@ -75,7 +117,7 @@ async def get_session() -> AsyncSession:
     """
     Async context-manager that yields a DB session.
 
-    Usage:
+    Usage (in every crud/handler function):
         async with AsyncSessionFactory() as session:
             result = await session.execute(...)
     """