Merge pull request #1486 from MODSetter/dev

fix(auto_reload): update task to use a lambda for user_id in async call

surfsense_backend/alembic/versions/158_evolve_podcasts_lifecycle.py

@@ -6,6 +6,8 @@ Revises: 157
 
 from collections.abc import Sequence
 
+import sqlalchemy as sa
+
 from alembic import op
 
 revision: str = "158"
@@ -14,7 +16,29 @@ branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 
 
+def _drop_podcasts_from_publication() -> None:
+    """Detach podcasts from zero_publication so status can be retyped.
+
+    Postgres refuses ``ALTER COLUMN ... TYPE`` on a column a publication
+    depends on. Some databases reach this migration with podcasts already
+    published (an interim apply_publication ran during 156); drop it here and
+    let migration 159 reconcile the publication to the canonical shape.
+    """
+    conn = op.get_bind()
+    published = conn.execute(
+        sa.text(
+            "SELECT 1 FROM pg_publication_tables "
+            "WHERE pubname = 'zero_publication' "
+            "AND schemaname = current_schema() AND tablename = 'podcasts'"
+        )
+    ).fetchone()
+    if published:
+        op.execute('ALTER PUBLICATION "zero_publication" DROP TABLE "podcasts";')
+
+
 def upgrade() -> None:
+    _drop_podcasts_from_publication()
+
     # Retype the status enum by swapping in a fresh type and casting existing
     # rows. The legacy transient value 'generating' maps onto 'rendering'.
     op.execute("ALTER TYPE podcast_status RENAME TO podcast_status_old;")

surfsense_backend/app/tasks/celery_tasks/auto_reload_task.py

@@ -59,7 +59,7 @@ def _card_error_payment_intent_id(exc: CardError) -> str | None:
 @celery_app.task(name="auto_reload_credits")
 def auto_reload_credits_task(user_id: str):
     """Charge the user's saved card to top up credits when below threshold."""
-    return run_async_celery_task(_auto_reload_credits, user_id)
+    return run_async_celery_task(lambda: _auto_reload_credits(user_id))
 
 
 async def _auto_reload_credits(user_id: str) -> None:

surfsense_backend/app/zero_publication.py

@@ -86,18 +86,15 @@ def _quote_identifier(identifier: str) -> str:
     return '"' + identifier.replace('"', '""') + '"'
 
 
-def _column_exists(conn: Connection, table: str, column: str) -> bool:
+def _table_columns(conn: Connection, table: str) -> set[str]:
-    return (
+    rows = conn.execute(
-        conn.execute(
+        text(
-            text(
+            "SELECT column_name FROM information_schema.columns "
-                "SELECT 1 FROM information_schema.columns "
+            "WHERE table_schema = current_schema() AND table_name = :table"
-                "WHERE table_schema = current_schema() "
+        ),
-                "AND table_name = :table AND column_name = :column"
+        {"table": table},
-            ),
+    ).fetchall()
-            {"table": table, "column": column},
+    return {row[0] for row in rows}
-        ).fetchone()
-        is not None
-    )
 
 
 def _expected_columns(conn: Connection, table: str) -> list[str] | None:
@@ -106,19 +103,39 @@ def _expected_columns(conn: Connection, table: str) -> list[str] | None:
         return None
 
     expected = list(columns)
-    if table in {"documents", "user", "podcasts"} and _column_exists(
+    if table in {"documents", "user", "podcasts"} and "_0_version" in _table_columns(
-        conn, table, "_0_version"
+        conn, table
     ):
         expected.append("_0_version")
     return expected
 
 
-def _format_table_entry(conn: Connection, table: str) -> str:
+def _format_table_entry(conn: Connection, table: str) -> str | None:
-    columns = _expected_columns(conn, table)
+    """Render one SET TABLE entry, or ``None`` if the table isn't ready.
+
+    Historical migrations (e.g. 155/156) call ``apply_publication`` while the
+    schema is still mid-history, before later migrations add columns that the
+    canonical shape references. A table is only published once it exists AND
+    every canonical column exists; otherwise it is omitted entirely and a later
+    reconcile migration (e.g. 159) picks it up once its columns land. Partial
+    column lists are deliberately avoided: publishing a column early would
+    block later ``ALTER COLUMN ... TYPE`` migrations on it (Postgres forbids
+    retyping columns a publication depends on). ``verify_publication`` remains
+    strict against the unfiltered canonical shape.
+    """
+
+    actual = _table_columns(conn, table)
+    if not actual:
+        return None
+
     table_sql = _quote_identifier(table)
+    columns = _expected_columns(conn, table)
     if columns is None:
         return table_sql
 
+    if any(column not in actual for column in columns):
+        return None
+
     column_sql = ", ".join(_quote_identifier(column) for column in columns)
     return f"{table_sql} ({column_sql})"
 
@@ -126,9 +143,8 @@ def _format_table_entry(conn: Connection, table: str) -> str:
 def build_set_table_sql(conn: Connection) -> str:
     """Build the canonical plain SET TABLE statement for Zero's event triggers."""
 
-    table_list = ", ".join(
+    entries = [_format_table_entry(conn, table) for table in ZERO_PUBLICATION]
-        _format_table_entry(conn, table) for table in ZERO_PUBLICATION
+    table_list = ", ".join(entry for entry in entries if entry is not None)
-    )
     return f"ALTER PUBLICATION {_quote_identifier(PUBLICATION_NAME)} SET TABLE {table_list}"