"""prd011: add scheduler_configs and scheduler_logs tables. Introduces two tables that back the Scheduler Admin UI (PRD-011-HWADM): scheduler_configs — one row per known Celery Beat task; controls whether the task is enabled, exposes metadata for the UI, and records who last changed the config. scheduler_logs — append-only execution log; one row per task run (RUNNING → SUCCESS / FAILED / SKIPPED). Child / per-merchant sub-runs link back via parent_log_id. Seeds 11 rows into scheduler_configs for the tasks that already exist in the codebase so the UI is immediately populated after migration. Revision ID: prd011_add_scheduler_tables Revises: prd010_add_invoice_split_id Create Date: 2026-05-06 """ from typing import Union, Sequence import sqlalchemy as sa from alembic import op from sqlalchemy import text revision: str = "prd011_add_scheduler_tables" down_revision: Union[str, Sequence[str], None] = "prd010_add_invoice_split_id" branch_labels = None depends_on = None # --------------------------------------------------------------------------- # upgrade # --------------------------------------------------------------------------- def upgrade() -> None: # ------------------------------------------------------------------ # scheduler_configs # ------------------------------------------------------------------ op.create_table( "scheduler_configs", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("task_name", sa.String(200), nullable=False), sa.Column("display_name", sa.String(200), nullable=False), sa.Column("description", sa.Text(), nullable=True), sa.Column("schedule_expression", sa.String(100), nullable=False), sa.Column( "is_enabled", sa.Boolean(), nullable=False, server_default=sa.text("true"), ), sa.Column( "supports_merchant_filter", sa.Boolean(), nullable=False, server_default=sa.text("false"), ), sa.Column("alert_threshold_ms", sa.Integer(), nullable=True), sa.Column( "log_retention_days", sa.Integer(), nullable=False, server_default=sa.text("90"), ), sa.Column( "last_modified_by_admin_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=True, ), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False, ), sa.Column( "updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False, ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("task_name", name="uq_scheduler_configs_task_name"), ) # ------------------------------------------------------------------ # scheduler_logs # ------------------------------------------------------------------ op.create_table( "scheduler_logs", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("task_name", sa.String(200), nullable=False), sa.Column("display_name", sa.String(200), nullable=False), sa.Column("run_status", sa.String(20), nullable=False), sa.Column("triggered_by", sa.String(20), nullable=False), sa.Column( "triggered_by_admin_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=True, ), sa.Column( "parent_log_id", sa.Integer(), sa.ForeignKey("scheduler_logs.id"), nullable=True, ), sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("duration_ms", sa.Integer(), nullable=True), sa.Column("records_processed", sa.Integer(), nullable=True), sa.Column("log_output", sa.Text(), nullable=True), sa.Column("error_message", sa.Text(), nullable=True), sa.Column("traceback", sa.Text(), nullable=True), sa.Column( "threshold_exceeded", sa.Boolean(), nullable=False, server_default=sa.text("false"), ), sa.Column("celery_task_id", sa.String(200), nullable=True), sa.Column("merchant_ids", sa.JSON(), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False, ), sa.PrimaryKeyConstraint("id"), ) # scheduler_logs indexes # Simple column indexes op.create_index( "ix_scheduler_logs_task_name", "scheduler_logs", ["task_name"], ) op.create_index( "ix_scheduler_logs_run_status", "scheduler_logs", ["run_status"], ) op.create_index( "ix_scheduler_logs_celery_task_id", "scheduler_logs", ["celery_task_id"], ) op.create_index( "ix_scheduler_logs_created_at", "scheduler_logs", ["created_at"], ) # Composite indexes for the two most common admin UI list queries: # 1. "show me all runs for task X, ordered by newest first" # 2. "show me runs for task X that have status Y" op.create_index( "ix_scheduler_logs_task_name_created_at", "scheduler_logs", ["task_name", sa.text("created_at DESC")], postgresql_using="btree", ) op.create_index( "ix_scheduler_logs_task_name_run_status", "scheduler_logs", ["task_name", "run_status"], ) # ------------------------------------------------------------------ # Seed scheduler_configs # ------------------------------------------------------------------ _seed_scheduler_configs() # --------------------------------------------------------------------------- # downgrade # --------------------------------------------------------------------------- def downgrade() -> None: # Drop logs first (has FK to itself and to users, but no FK from configs) op.drop_index("ix_scheduler_logs_task_name_run_status", table_name="scheduler_logs") op.drop_index("ix_scheduler_logs_task_name_created_at", table_name="scheduler_logs") op.drop_index("ix_scheduler_logs_created_at", table_name="scheduler_logs") op.drop_index("ix_scheduler_logs_celery_task_id", table_name="scheduler_logs") op.drop_index("ix_scheduler_logs_run_status", table_name="scheduler_logs") op.drop_index("ix_scheduler_logs_task_name", table_name="scheduler_logs") op.drop_table("scheduler_logs") op.drop_table("scheduler_configs") # --------------------------------------------------------------------------- # seed helper # --------------------------------------------------------------------------- def _seed_scheduler_configs() -> None: """ Insert the 11 known scheduler tasks into scheduler_configs so the admin UI is populated immediately after migration. Uses ON CONFLICT DO NOTHING so re-running upgrade() on a non-empty DB is safe. """ bind = op.get_bind() rows = [ # (task_name, display_name, schedule_expression, supports_merchant_filter, description) ( "invoice.mark_overdue_invoices", "Mark Overdue Invoices", "Daily 00:05 UTC", True, "Scans open invoices whose due_date has passed and transitions them to OVERDUE status.", ), ( "invoice.send_reminder_poll", "Invoice Reminder Poll", "Every 15 min", True, "Checks scheduled invoice reminders and dispatches emails/SMS when the send time arrives.", ), ( "subscription.update_subscription_statuses", "Update Subscription Statuses", "Daily 00:30 UTC", True, "Transitions subscriptions to ACTIVE, PAST_DUE, or CANCELLED based on billing cycle dates.", ), ( "subscription.generate_upcoming_invoices", "Generate Upcoming Invoices", "Daily 01:00 UTC", True, "Pre-generates invoices for subscriptions due to bill within the lookahead window.", ), ( "update_maxmind_db", "MaxMind DB Update", "Weekly Sun 02:00 UTC", False, "Downloads the latest MaxMind GeoIP2 database used for fraud-risk geolocation.", ), ( "cart_plugin.expire_cart_sessions", "Expire Cart Sessions", "Every 15 min", True, "Expires abandoned cart sessions that have exceeded their configured TTL.", ), ( "checkout.expire_stale_checkouts", "Expire Stale Checkouts", "Every 1 hr", True, "Marks checkout sessions as EXPIRED when they exceed the maximum pending duration.", ), ( "payment_requests.process_due_payment_requests", "Process Due Payment Requests", "Daily 08:00 UTC", True, "Picks up recurring payment requests whose next_run_date is today and submits charges.", ), ( "tax.refresh_tax_rate_cache", "Refresh Tax Rate Cache", "Monthly 1st 02:00 UTC", False, "Refreshes cached tax rates from the configured tax provider (e.g. TaxJar / Avalara).", ), ( "user-profile-cleanup-expired-email-tokens", "Cleanup Expired Email Tokens", "Daily 02:00 UTC", False, "Purges expired email-verification and email-change tokens from the database.", ), ( "admin.purge_old_scheduler_logs", "Purge Old Scheduler Logs", "Daily 03:00 UTC", False, "Deletes scheduler_logs rows older than each task's log_retention_days threshold.", ), ] for task_name, display_name, schedule_expression, supports_merchant_filter, description in rows: bind.execute( text( """ INSERT INTO scheduler_configs (task_name, display_name, description, schedule_expression, is_enabled, supports_merchant_filter, log_retention_days, created_at, updated_at) VALUES (:task_name, :display_name, :description, :schedule_expression, true, :supports_merchant_filter, 90, now(), now()) ON CONFLICT (task_name) DO NOTHING """ ), { "task_name": task_name, "display_name": display_name, "description": description, "schedule_expression": schedule_expression, "supports_merchant_filter": supports_merchant_filter, }, )