"""Day-90 onboarding-experience survey push.

1. ``identity_persons.day90_survey_sent_at`` — idempotency stamp for the
   90-day survey invite (Feishu card + email).
2. Seeds the ``hr.day90_survey_notify`` scheduled job (daily 02:00 UTC =
   10:00 Beijing): invites hires who reached 90 days of tenure (their
   LOCAL calendar date) to fill the onboarding-experience survey; a
   90–97-day catch-up window absorbs missed ticks without ever storming
   long-tenured employees at rollout.

Revision ID: 377
Revises: 376
"""
import uuid

import sqlalchemy as sa

from alembic import op

revision = "377"
down_revision = "376"
branch_labels = None
depends_on = None

JOB = "hr.day90_survey_notify"


def upgrade() -> None:
    op.add_column(
        "identity_persons",
        sa.Column("day90_survey_sent_at", sa.DateTime(timezone=True), nullable=True),
    )

    bind = op.get_bind()
    job_exists = bind.execute(
        sa.text("SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1"),
        {"n": JOB},
    ).scalar()
    if not job_exists:
        bind.execute(
            sa.text(
                """
                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)
                VALUES
                    (:id, :name, :display_name, :description, :name,
                     'cron', NULL, '0 2 * * *', 'UTC',
                     1, '{}', 0, 0, NOW(), NOW())
                """
            ),
            {
                "id": str(uuid.uuid4()),
                "name": JOB,
                "display_name": "入职 90 天 · 体验问卷推送",
                "description": (
                    "每天 02:00 UTC 按员工本地日历给入职满 90 天的新员工"
                    "推送入职体验问卷邀请(飞书卡片 + 邮件,双语);"
                    "幂等键 identity_persons.day90_survey_sent_at;"
                    "90–97 天补发窗口兜底漏跑,不会打扰老员工。"
                ),
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    bind.execute(sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": JOB})
    op.drop_column("identity_persons", "day90_survey_sent_at")