diff --git a/surfsense_backend/alembic/versions/158_evolve_podcasts_lifecycle.py b/surfsense_backend/alembic/versions/158_evolve_podcasts_lifecycle.py index ae595cc1b..f3b194cbd 100644 --- a/surfsense_backend/alembic/versions/158_evolve_podcasts_lifecycle.py +++ b/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;") diff --git a/surfsense_backend/app/tasks/celery_tasks/auto_reload_task.py b/surfsense_backend/app/tasks/celery_tasks/auto_reload_task.py index 41a6f0b70..385cdde88 100644 --- a/surfsense_backend/app/tasks/celery_tasks/auto_reload_task.py +++ b/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: diff --git a/surfsense_backend/app/zero_publication.py b/surfsense_backend/app/zero_publication.py index 139286ee6..b14ee14d1 100644 --- a/surfsense_backend/app/zero_publication.py +++ b/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: - return ( - conn.execute( - text( - "SELECT 1 FROM information_schema.columns " - "WHERE table_schema = current_schema() " - "AND table_name = :table AND column_name = :column" - ), - {"table": table, "column": column}, - ).fetchone() - is not None - ) +def _table_columns(conn: Connection, table: str) -> set[str]: + rows = conn.execute( + text( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = current_schema() AND table_name = :table" + ), + {"table": table}, + ).fetchall() + return {row[0] for row in rows} 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( - conn, table, "_0_version" + if table in {"documents", "user", "podcasts"} and "_0_version" in _table_columns( + conn, table ): expected.append("_0_version") return expected -def _format_table_entry(conn: Connection, table: str) -> str: - columns = _expected_columns(conn, table) +def _format_table_entry(conn: Connection, table: str) -> str | None: + """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( - _format_table_entry(conn, table) for table in ZERO_PUBLICATION - ) + entries = [_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}"