"""Scheduled jobs framework — table + seed built-in jobs. Revision ID: 037 Revises: 036 Create Date: 2026-04-20 Introduces a DB-driven scheduler: every recurring task (HRIS incremental sync, overdue-leaver check, domain-expiry alert, future "daily reminder to managers", ...) is a row in ``scheduled_jobs`` with its cadence, enabled flag, and last-run telemetry. ``scheduled_job_runs`` holds the per-execution history so the admin console can show "last 20 runs" without parsing stdout. The migration also seeds three rows so existing timers keep running after deploy: the ad-hoc loops in ``app/worker.py`` are being replaced by a scheduler that reads these rows. """ from __future__ import annotations import sqlalchemy as sa from alembic import op revision = "037" down_revision = "036" branch_labels = None depends_on = None def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n" ), {"n": name}, ).scalar() is not None def upgrade() -> None: """Idempotent — safe to re-run after a previous partial failure. MySQL/TiDB DDL isn't transactional; a mid-migration crash can leave one table created and the other missing. Guard each ``CREATE TABLE`` with an information_schema check so a retry after partial success doesn't die on ``1050 "Table ... already exists"``. """ bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): # Either first run succeeded then something later rolled back, # or operator hand-created it. Either way skip the CREATE and # fall through to the index / seed blocks (also idempotent). pass else: _create_scheduled_jobs_table() if not _table_exists(bind, "scheduled_job_runs"): _create_scheduled_job_runs_table() _seed_builtin_jobs() def _create_scheduled_jobs_table() -> None: op.create_table( "scheduled_jobs", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("name", sa.String(128), nullable=False), sa.Column("display_name", sa.String(128), nullable=False), sa.Column("description", sa.Text, nullable=True), sa.Column("handler_key", sa.String(128), nullable=False), sa.Column("trigger_type", sa.String(16), nullable=False), sa.Column("interval_seconds", sa.Integer, nullable=True), sa.Column("cron_expr", sa.String(128), nullable=True), sa.Column( "timezone", sa.String(32), nullable=False, server_default="UTC", ), sa.Column("enabled", sa.Boolean, nullable=False, server_default=sa.true()), sa.Column("config", sa.JSON, nullable=True), sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True), sa.Column("last_status", sa.String(16), nullable=True), sa.Column("last_duration_ms", sa.Integer, nullable=True), sa.Column("last_error", sa.Text, nullable=True), sa.Column("last_result_summary", sa.JSON, nullable=True), sa.Column("next_run_at", sa.DateTime(timezone=True), nullable=True), sa.Column("run_count", sa.Integer, nullable=False, server_default="0"), sa.Column("failure_count", sa.Integer, nullable=False, server_default="0"), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint("name", name="uq_scheduled_jobs_name"), ) op.create_index( "ix_scheduled_jobs_enabled", "scheduled_jobs", ["enabled"], ) def _create_scheduled_job_runs_table() -> None: op.create_table( "scheduled_job_runs", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "job_id", sa.CHAR(36), sa.ForeignKey("scheduled_jobs.id", ondelete="CASCADE"), nullable=False, ), sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), sa.Column("duration_ms", sa.Integer, nullable=True), sa.Column("status", sa.String(16), nullable=False), sa.Column("triggered_by", sa.String(64), nullable=False), sa.Column("error_message", sa.Text, nullable=True), sa.Column("result_summary", sa.JSON, nullable=True), ) op.create_index( "ix_scheduled_job_runs_job_started", "scheduled_job_runs", ["job_id", "started_at"], ) def _seed_builtin_jobs() -> None: # Seed the 3 built-in jobs so existing behaviour survives the # worker refactor. Rows are inserted only if absent (idempotent). op.execute(""" INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) SELECT UUID(), 'hris.incremental_sync', 'HRIS 增量同步', '遍历所有已启用的 HRIS 集成,按 lastModifiedDateTime 游标拉取变更并自动确认。', 'hris.incremental_sync', 'interval', 14400, NULL, 'UTC', 1, JSON_OBJECT('lookback_hours', 25), 0, 0, NOW(), NOW() FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM scheduled_jobs WHERE name = 'hris.incremental_sync') """) op.execute(""" INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) SELECT UUID(), 'hr.overdue_leavers_check', '逾期离职员工状态推进', '把 termination_date 已过期的 leaver_pending 员工推进到 left;并跑经理升级 tick。', 'hr.overdue_leavers_check', 'interval', 3600, NULL, 'UTC', 1, NULL, 0, 0, NOW(), NOW() FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM scheduled_jobs WHERE name = 'hr.overdue_leavers_check') """) op.execute(""" INSERT INTO scheduled_jobs (id, name, display_name, description, handler_key, trigger_type, interval_seconds, cron_expr, timezone, enabled, config, run_count, failure_count, created_at, updated_at) SELECT UUID(), 'cmdb.domain_expiry_alert', '域名到期提醒', '扫描 CMDB 域名,30 天内到期或已过期时通过 Feishu 告警。', 'cmdb.domain_expiry_alert', 'cron', NULL, '0 9 * * *', 'Asia/Shanghai', 1, JSON_OBJECT('warn_days', 30), 0, 0, NOW(), NOW() FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM scheduled_jobs WHERE name = 'cmdb.domain_expiry_alert') """) def downgrade() -> None: op.drop_index("ix_scheduled_job_runs_job_started", table_name="scheduled_job_runs") op.drop_table("scheduled_job_runs") op.drop_index("ix_scheduled_jobs_enabled", table_name="scheduled_jobs") op.drop_table("scheduled_jobs")