Adopt Alembic migrations (match the sibling-service house pattern)

All sibling services on the box manage schema with Alembic (async env, `alembic upgrade head` in the
compose command). matchmaking-v2 was the outlier (Base.metadata.create_all + no migration for the new
user_feed.exhausted_at column). Bring it onto the pattern:
- alembic.ini + alembic/env.py (async, URL from settings.DATABASE_URL, target = Base.metadata) +
  script.py.mako — copied/adapted from user-service.
- versions/0001_baseline: the current prod schema (user_feed WITHOUT exhausted_at + opportunity_state).
- versions/0002_pool_and_exhausted: CREATE user_job_pool + ADD user_feed.exhausted_at (additive).
- Dockerfile: COPY alembic.ini + alembic/. requirements: alembic>=1.13.0.

Verified both paths: fresh DB → upgrade head builds all 3 tables; prod-like (existing tables+data) →
stamp 0001 → upgrade runs ONLY 0002, existing rows survive. 64 tests pass.

PROD ADOPTION (one-time): `alembic stamp 0001_baseline` on RDS before the first deploy, then the
compose's `alembic upgrade head` applies 0002.

Dockerfile

@@ -8,6 +8,9 @@ COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 
 COPY app ./app
+COPY alembic.ini .
+COPY alembic/ ./alembic/
 
 EXPOSE 8000
+# migrations run from the compose command (alembic upgrade head && uvicorn …) — the house pattern
 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

alembic.ini

@@ -0,0 +1,39 @@
+[alembic]
+script_location = alembic
+prepend_sys_path = .
+
+# Default (overridden in alembic/env.py from DATABASE_URL)
+sqlalchemy.url = postgresql+asyncpg://matchmaking:matchmaking@localhost:5446/matchmaking
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

alembic/env.py

@@ -0,0 +1,67 @@
+"""Alembic async env — matches the sibling-service house pattern (user-service et al.).
+
+Reads the live DB URL from app settings (overriding the alembic.ini placeholder), targets
+Base.metadata (all models live in app.db.models), and runs migrations on the async engine.
+"""
+import asyncio
+import sys
+from logging.config import fileConfig
+from pathlib import Path
+
+# Ensure the project root is importable when alembic runs via the console script.
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from alembic import context
+from sqlalchemy import pool
+from sqlalchemy.ext.asyncio import async_engine_from_config
+
+from app.config import get_settings
+from app.db.models import Base  # importing Base registers every model on Base.metadata
+
+settings = get_settings()
+config = context.config
+
+# Live URL from settings (DATABASE_URL) overrides the ini placeholder; fall back to the ini value.
+config.set_main_option("sqlalchemy.url", settings.DATABASE_URL or config.get_main_option("sqlalchemy.url"))
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+
+def run_migrations_offline() -> None:
+    url = config.get_main_option("sqlalchemy.url")
+    context.configure(url=url, target_metadata=target_metadata, literal_binds=True,
+                      dialect_opts={"paramstyle": "named"})
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def do_run_migrations(connection):
+    context.configure(connection=connection, target_metadata=target_metadata)
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+async def run_async_migrations() -> None:
+    connectable = async_engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+    async with connectable.connect() as connection:
+        await connection.run_sync(do_run_migrations)
+    await connectable.dispose()
+
+
+def run_migrations_online() -> None:
+    asyncio.run(run_async_migrations())
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

alembic/script.py.mako

@@ -0,0 +1,24 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, None] = ${repr(down_revision)}
+branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

alembic/versions/0001_baseline.py

@@ -0,0 +1,51 @@
+"""baseline — the schema as it exists in prod today (from the cutover's create_all):
+user_feed (WITHOUT exhausted_at) + opportunity_state. On prod this is STAMPED (the tables already
+exist); on a fresh DB `upgrade` creates them. The exhausted_at column + user_job_pool are added in 0002.
+
+Revision ID: 0001_baseline
+Revises:
+Create Date: 2026-06-26
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0001_baseline"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "user_feed",
+        sa.Column("user_id", sa.String(), primary_key=True),
+        sa.Column("prefs", sa.JSON(), nullable=True),
+        sa.Column("opportunities", sa.JSON(), nullable=True),
+        sa.Column("sources", sa.JSON(), nullable=True),
+        sa.Column("engine", sa.String(), nullable=True),
+        sa.Column("scanned", sa.Integer(), nullable=True),
+        sa.Column("query_sig", sa.String(), nullable=True),
+        sa.Column("cursors", sa.JSON(), nullable=True),
+        sa.Column("search_count", sa.Integer(), server_default="0", nullable=True),
+        sa.Column("salary_peak_sum", sa.Float(), server_default="0", nullable=True),
+        sa.Column("salary_peak_count", sa.Integer(), server_default="0", nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True),
+    )
+    op.create_table(
+        "opportunity_state",
+        sa.Column("user_id", sa.String(), primary_key=True),
+        sa.Column("opportunity_id", sa.String(), primary_key=True),
+        sa.Column("status", sa.String(), nullable=True),
+        sa.Column("tailored_resume_id", sa.String(), nullable=True),
+        sa.Column("tailored_version_id", sa.String(), nullable=True),
+        sa.Column("cover_letter_id", sa.String(), nullable=True),
+        sa.Column("seen", sa.Boolean(), server_default="false", nullable=True),
+        sa.Column("viewed", sa.Boolean(), server_default="false", nullable=True),
+        sa.Column("saved", sa.Boolean(), server_default="false", nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("opportunity_state")
+    op.drop_table("user_feed")

alembic/versions/0002_pool_and_exhausted.py

@@ -0,0 +1,33 @@
+"""warm pool — create user_job_pool + add user_feed.exhausted_at.
+
+The actual change this deploy ships: the warm-pool table and the exhaustion-guard column. Additive +
+non-breaking (old code ignores them). On prod: stamp 0001 (existing schema), then this runs.
+
+Revision ID: 0002_pool_and_exhausted
+Revises: 0001_baseline
+Create Date: 2026-06-26
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0002_pool_and_exhausted"
+down_revision = "0001_baseline"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "user_job_pool",
+        sa.Column("user_id", sa.String(), primary_key=True),
+        sa.Column("query_sig", sa.String(), primary_key=True),
+        sa.Column("job_id", sa.String(), primary_key=True),
+        sa.Column("job", sa.JSON(), nullable=True),
+        sa.Column("pooled_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True),
+    )
+    op.add_column("user_feed", sa.Column("exhausted_at", sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("user_feed", "exhausted_at")
+    op.drop_table("user_job_pool")

requirements.txt

@@ -11,3 +11,4 @@ openai>=1.40
 numpy
 sqlalchemy[asyncio]>=2.0
 asyncpg
+alembic>=1.13.0