channel require and db update

db/database_setup.py

@@ -4,7 +4,7 @@ from sqlalchemy.orm import sessionmaker
 
 from config.settings import Settings
 from .models import Base
-from .migrator import run_simple_migrations
+from .migrator import run_database_migrations
 
 async_engine = None
 
@@ -63,8 +63,7 @@ async def init_db(settings: Settings, session_factory: sessionmaker):
 
     async with async_engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
-        # Run lightweight, idempotent migrations to add any missing columns
-        await conn.run_sync(run_simple_migrations)
+        await conn.run_sync(run_database_migrations)
     logging.info(
         "PostgreSQL database initialized/checked successfully using SQLAlchemy."
     )

db/migrator.py

@@ -1,66 +1,98 @@
 import logging
-from typing import Set
+from dataclasses import dataclass
+from typing import Callable, List, Set
 
 from sqlalchemy import inspect, text
 from sqlalchemy.engine import Connection
 
-from .models import Base
+
+@dataclass(frozen=True)
+class Migration:
+    id: str
+    description: str
+    upgrade: Callable[[Connection], None]
 
 
-def _add_missing_columns(connection: Connection) -> None:
+def _ensure_migrations_table(connection: Connection) -> None:
+    connection.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS schema_migrations (
+                id VARCHAR(255) PRIMARY KEY,
+                applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+            )
+            """
+        )
+    )
+
+
+def _migration_0001_add_channel_subscription_fields(connection: Connection) -> None:
     inspector = inspect(connection)
-    metadata = Base.metadata
+    columns: Set[str] = {col["name"] for col in inspector.get_columns("users")}
+    statements: List[str] = []
 
-    existing_tables: Set[str] = set(inspector.get_table_names())
+    if "channel_subscription_verified" not in columns:
+        statements.append(
+            "ALTER TABLE users ADD COLUMN channel_subscription_verified BOOLEAN"
+        )
+    if "channel_subscription_checked_at" not in columns:
+        statements.append(
+            "ALTER TABLE users ADD COLUMN channel_subscription_checked_at TIMESTAMPTZ"
+        )
+    if "channel_subscription_verified_for" not in columns:
+        statements.append(
+            "ALTER TABLE users ADD COLUMN channel_subscription_verified_for BIGINT"
+        )
 
-    for table in metadata.tables.values():
-        table_name = table.name
-        if table_name not in existing_tables:
-            # Tables are created elsewhere via create_all; skip here.
+    for stmt in statements:
+        connection.execute(text(stmt))
+
+
+MIGRATIONS: List[Migration] = [
+    Migration(
+        id="0001_add_channel_subscription_fields",
+        description="Add columns to track required channel subscription verification",
+        upgrade=_migration_0001_add_channel_subscription_fields,
+    ),
+]
+
+
+def run_database_migrations(connection: Connection) -> None:
+    """
+    Apply pending migrations sequentially. Already applied revisions are skipped.
+    """
+    _ensure_migrations_table(connection)
+
+    applied_revisions: Set[str] = {
+        row[0]
+        for row in connection.execute(
+            text("SELECT id FROM schema_migrations")
+        )
+    }
+
+    for migration in MIGRATIONS:
+        if migration.id in applied_revisions:
             continue
 
-        existing_columns = {col_info["name"] for col_info in inspector.get_columns(table_name)}
-
-        for desired_column in table.columns:
-            if desired_column.name in existing_columns:
-                continue
-
-            # Build ADD COLUMN DDL
-            preparer = connection.dialect.identifier_preparer
-            table_quoted = preparer.format_table(table)
-            column_name_quoted = preparer.quote(desired_column.name)
-            column_type_sql = desired_column.type.compile(dialect=connection.dialect)
-
-            default_clause = ""
-            server_default = getattr(desired_column, "server_default", None)
-            if server_default is not None and getattr(server_default, "arg", None) is not None:
-                try:
-                    compiled_default = str(
-                        server_default.arg.compile(dialect=connection.dialect)
-                    )
-                    default_clause = f" DEFAULT {compiled_default}"
-                except Exception:  # best-effort
-                    pass
-
-            # For safety, add new columns as NULLable to avoid failures on existing rows
-            # If strict NOT NULL is needed, it can be enforced manually later.
-            ddl = f"ALTER TABLE {table_quoted} ADD COLUMN {column_name_quoted} {column_type_sql}{default_clause}"
-
-            logging.info(
-                f"Migrator: adding missing column {desired_column.name} to table {table_name}"
+        logging.info(
+            "Migrator: applying %s – %s", migration.id, migration.description
+        )
+        try:
+            with connection.begin_nested():
+                migration.upgrade(connection)
+                connection.execute(
+                    text(
+                        "INSERT INTO schema_migrations (id) VALUES (:revision)"
+                    ),
+                    {"revision": migration.id},
+                )
+        except Exception as exc:
+            logging.error(
+                "Migrator: failed to apply %s (%s)",
+                migration.id,
+                migration.description,
+                exc_info=True,
             )
-            connection.execute(text(ddl))
-
-
-def run_simple_migrations(connection: Connection) -> None:
-    """
-    Run lightweight, idempotent migrations:
-    - Ensure missing columns are added to existing tables to match models in db/models.py
-    Note: Table creation is handled separately via Base.metadata.create_all.
-    """
-    try:
-        _add_missing_columns(connection)
-        logging.info("Migrator: schema synchronized (columns added as needed).")
-    except Exception as e:
-        logging.error(f"Migrator: failed to run simple migrations: {e}", exc_info=True)
-        raise
+            raise exc
+        else:
+            logging.info("Migrator: migration %s applied successfully", migration.id)

db/models.py

@@ -24,6 +24,10 @@ class User(Base):
     referred_by_id = Column(BigInteger,
                             ForeignKey("users.user_id"),
                             nullable=True)
+    channel_subscription_verified = Column(Boolean, nullable=True)
+    channel_subscription_checked_at = Column(DateTime(timezone=True),
+                                             nullable=True)
+    channel_subscription_verified_for = Column(BigInteger, nullable=True)
 
     referrer = relationship("User", remote_side=[user_id], backref="referrals")
     subscriptions = relationship("Subscription",