"""Feed domain — 动态广场: posts, comments, likes, blessings, announcements.

Revision ID: 235
Revises: 234
Create Date: 2026-06-01

Adds the employee 动态广场 (employee social feed) feature:

* ``identity_persons.birthday`` — DATE, nullable. Sourced from
  SuccessFactors ``PerPersonal.dateOfBirth`` (next HRIS sync after
  this migration ships) and/or self-edited by the person on
  ``/me/settings``. Display strategy is governed by
  ``identity_person_preferences`` key ``feed.birthday_visibility``
  (``month_day`` default, ``hidden`` to suppress).

* Nine new tables under the ``feed_`` prefix:

  - ``feed_posts`` — top-level post stream. ``post_type`` ∈
    {``user``, ``system_birthday``, ``system_anniversary``,
    ``aggregate_birthday``, ``aggregate_anniversary``}. ``visibility``
    ∈ {``public``, ``department``}. System posts are always public.
  - ``feed_comments`` — flat comments with optional
    ``reply_to_comment_id`` for "回复 @xxx" display.
  - ``feed_likes`` — one row per (post, person). UNIQUE.
  - ``feed_mentions`` — denormalized @-mention index, post OR comment.
  - ``feed_attachments`` — image attachments, ordered by ``position``.
  - ``feed_celebrations`` — one row per (person, occasion_date,
    celebration_type). The source-of-truth event row that the system
    posts hang off. UNIQUE so the hourly cron is idempotent.
  - ``feed_blessings`` — one row per (celebration, sender). UNIQUE.
    Mirrors a row into ``feed_comments`` on the system post for the
    same celebration so the unified comment thread renders.
  - ``feed_announcements`` — rich-text announcements on the right
    rail. Markdown body, scheduled publish + expire, pinning,
    category, color.

* ``identity_person_preferences`` — generic per-person KV store. First
  keys: ``feed.birthday_visibility``, ``feed.notify_comment_feishu``,
  ``feed.notify_like_aggregate``.

* Three rows in ``scheduled_jobs`` for the celebration scanner,
  celebrant/lead notifier, and announcement pin-expiry — enabled at
  seed time (no destructive side effects, posts are appendable).

Idempotent: every DDL/seed is guarded with information_schema /
NOT EXISTS. Downgrade drops the new tables + birthday column.
"""
from __future__ import annotations

import json
import uuid

import sqlalchemy as sa
from alembic import op

from app.core.migration_helpers import table_kw_for

revision = "235"
down_revision = "234"
branch_labels = None
depends_on = None


def _table_exists(table: str) -> bool:
    conn = op.get_bind()
    result = conn.execute(
        sa.text(
            "SELECT COUNT(*) FROM information_schema.tables "
            "WHERE table_schema = DATABASE() AND table_name = :t"
        ),
        {"t": table},
    )
    return (result.scalar() or 0) > 0


def _column_exists(table: str, column: str) -> bool:
    conn = op.get_bind()
    result = conn.execute(
        sa.text(
            "SELECT COUNT(*) FROM information_schema.columns "
            "WHERE table_schema = DATABASE() "
            "AND table_name = :t AND column_name = :c"
        ),
        {"t": table, "c": column},
    )
    return (result.scalar() or 0) > 0


def _feed_table_has_varchar_id(table: str) -> bool:
    """Detect the prior-bad state: an earlier cut of this migration
    declared id / FK columns as ``sa.String(36)`` → VARCHAR(36) under
    MySQL. Local CI MySQL accepted FKs VARCHAR(36) → CHAR(36); prod
    TiDB rejects them with error 3780 ("Referencing column … are
    incompatible"). If a local DB has the feed table with id =
    VARCHAR, the migration must DROP it so the corrected CHAR(36)
    CREATE TABLE below converges. On a clean prod deploy the table
    doesn't exist yet → guard returns False, no drop happens.
    """
    conn = op.get_bind()
    row = conn.execute(
        sa.text(
            "SELECT DATA_TYPE FROM information_schema.columns "
            "WHERE table_schema = DATABASE() AND table_name = :t "
            "AND column_name = 'id' LIMIT 1"
        ),
        {"t": table},
    ).scalar()
    return bool(row) and str(row).lower() == "varchar"


# Feed tables in reverse-FK order so DROP doesn't violate parent refs.
_FEED_TABLES_REVERSE_FK = [
    "feed_announcements",
    "feed_blessings",
    "feed_attachments",
    "feed_mentions",
    "feed_likes",
    "feed_comments",
    "feed_posts",
    "feed_celebrations",
    "identity_person_preferences",
]


def upgrade() -> None:
    # ── 0. TiDB-FK-strictness cleanup ─────────────────────────────
    # See ``_feed_table_has_varchar_id`` docstring. Feed tables are
    # empty on first deploy + the DB is the only data source, so the
    # drop loses no data. Subsequent CREATE TABLE blocks recreate
    # them with the correct CHAR(36) type that matches
    # ``identity_persons.id``'s CHAR(36) on prod TiDB.
    for tbl in _FEED_TABLES_REVERSE_FK:
        if _feed_table_has_varchar_id(tbl):
            op.execute(f"DROP TABLE IF EXISTS `{tbl}`")

    # ── 1. identity_persons.birthday ──────────────────────────────
    if not _column_exists("identity_persons", "birthday"):
        op.add_column(
            "identity_persons",
            sa.Column("birthday", sa.Date(), nullable=True),
        )

    # ── 2. identity_person_preferences ────────────────────────────
    if not _table_exists("identity_person_preferences"):
        op.create_table(
            "identity_person_preferences",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            sa.Column("pref_key", sa.String(length=64), nullable=False),
            sa.Column("pref_value", sa.JSON(), nullable=True),
            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(),
                server_onupdate=sa.func.now(),
            ),
            sa.PrimaryKeyConstraint("id"),
            sa.UniqueConstraint(
                "person_id", "pref_key",
                name="uq_identity_person_preferences_key",
            ),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_identity_person_preferences_person",
            "identity_person_preferences",
            ["person_id"],
        )

    # ── 3. feed_celebrations (built before feed_posts so the FK
    #       feed_posts.celebration_id resolves on CREATE) ──────────
    if not _table_exists("feed_celebrations"):
        op.create_table(
            "feed_celebrations",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            sa.Column(
                "celebration_type",
                sa.String(length=16),
                nullable=False,
                comment="'birthday' | 'anniversary'",
            ),
            sa.Column("occasion_date", sa.Date(), nullable=False),
            sa.Column(
                "years",
                sa.Integer(),
                nullable=False,
                server_default="0",
            ),
            sa.Column(
                "display_name_snapshot",
                sa.String(length=128),
                nullable=True,
            ),
            sa.Column(
                "department_snapshot",
                sa.String(length=128),
                nullable=True,
            ),
            sa.Column(
                "avatar_url_snapshot",
                sa.String(length=512),
                nullable=True,
            ),
            sa.Column(
                "illustration_key",
                sa.String(length=32),
                nullable=False,
                server_default="default",
            ),
            # Notify state — JSON map of which targets have been
            # notified for this celebration so the notifier cron is
            # idempotent across re-runs. Shape:
            # {"self": "<isoformat>", "manager": "...", "hrbp": "..."}
            sa.Column(
                "notify_state_json",
                sa.JSON(),
                nullable=True,
            ),
            sa.Column(
                "created_at",
                sa.DateTime(timezone=True),
                nullable=False,
                server_default=sa.func.now(),
            ),
            sa.PrimaryKeyConstraint("id"),
            sa.UniqueConstraint(
                "person_id", "occasion_date", "celebration_type",
                name="uq_feed_celebrations_person_date_type",
            ),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_celebrations_date_type",
            "feed_celebrations",
            ["occasion_date", "celebration_type"],
        )
        op.create_index(
            "ix_feed_celebrations_person",
            "feed_celebrations",
            ["person_id"],
        )

    # ── 4. feed_posts ────────────────────────────────────────────
    if not _table_exists("feed_posts"):
        op.create_table(
            "feed_posts",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "author_person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=True,
                comment="NULL = system-generated",
            ),
            # user | system_birthday | system_anniversary
            #      | aggregate_birthday | aggregate_anniversary
            sa.Column("post_type", sa.String(length=32), nullable=False),
            # public | department  (system posts are always public)
            sa.Column(
                "visibility",
                sa.String(length=16),
                nullable=False,
                server_default="public",
            ),
            # general | work_plan | other (user-post tab category)
            sa.Column(
                "category",
                sa.String(length=32),
                nullable=False,
                server_default="general",
            ),
            sa.Column("body", sa.Text(), nullable=False),
            # For system posts: target person, occasion date, years.
            sa.Column(
                "celebration_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_celebrations.id"),
                nullable=True,
            ),
            # Department snapshot for visibility check on read path.
            # Captured at post-time so re-orgs don't retroactively hide
            # old posts. NULL for visibility='public'.
            sa.Column(
                "department_snapshot",
                sa.String(length=128),
                nullable=True,
            ),
            sa.Column(
                "like_count",
                sa.Integer(),
                nullable=False,
                server_default="0",
            ),
            sa.Column(
                "comment_count",
                sa.Integer(),
                nullable=False,
                server_default="0",
            ),
            sa.Column(
                "is_pinned",
                sa.Boolean(),
                nullable=False,
                server_default=sa.text("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(),
                server_onupdate=sa.func.now(),
            ),
            sa.Column(
                "deleted_at",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
            sa.PrimaryKeyConstraint("id"),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_posts_visibility_created",
            "feed_posts",
            ["visibility", "created_at"],
        )
        op.create_index(
            "ix_feed_posts_author_created",
            "feed_posts",
            ["author_person_id", "created_at"],
        )
        op.create_index(
            "ix_feed_posts_type_created",
            "feed_posts",
            ["post_type", "created_at"],
        )
        op.create_index(
            "ix_feed_posts_department",
            "feed_posts",
            ["department_snapshot", "created_at"],
        )
        op.create_index(
            "ix_feed_posts_celebration",
            "feed_posts",
            ["celebration_id"],
        )
        op.create_index(
            "ix_feed_posts_pinned",
            "feed_posts",
            ["is_pinned", "created_at"],
        )
        op.create_index(
            "ix_feed_posts_category_created",
            "feed_posts",
            ["category", "created_at"],
        )

    # ── 5. feed_comments ─────────────────────────────────────────
    if not _table_exists("feed_comments"):
        op.create_table(
            "feed_comments",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "post_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_posts.id"),
                nullable=False,
            ),
            sa.Column(
                "author_person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            sa.Column("body", sa.Text(), nullable=False),
            # Flat threading — display only ("reply @xxx" prefix).
            sa.Column(
                "reply_to_comment_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_comments.id"),
                nullable=True,
            ),
            # If this comment was sourced from a sticker-blessing (the
            # 送祝福 modal), keep the sticker key + blessing back-pointer
            # so the celebration page can render the gift icon next to
            # the message.
            sa.Column(
                "blessing_id",
                sa.CHAR(length=36),
                nullable=True,
            ),
            sa.Column("sticker_key", sa.String(length=32), nullable=True),
            sa.Column(
                "created_at",
                sa.DateTime(timezone=True),
                nullable=False,
                server_default=sa.func.now(),
            ),
            sa.Column(
                "deleted_at",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
            sa.PrimaryKeyConstraint("id"),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_comments_post_created",
            "feed_comments",
            ["post_id", "created_at"],
        )
        op.create_index(
            "ix_feed_comments_author",
            "feed_comments",
            ["author_person_id"],
        )

    # ── 6. feed_likes ────────────────────────────────────────────
    if not _table_exists("feed_likes"):
        op.create_table(
            "feed_likes",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "post_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_posts.id"),
                nullable=False,
            ),
            sa.Column(
                "person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            sa.Column(
                "created_at",
                sa.DateTime(timezone=True),
                nullable=False,
                server_default=sa.func.now(),
            ),
            sa.PrimaryKeyConstraint("id"),
            sa.UniqueConstraint(
                "post_id", "person_id",
                name="uq_feed_likes_post_person",
            ),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_likes_post",
            "feed_likes",
            ["post_id"],
        )
        op.create_index(
            "ix_feed_likes_person",
            "feed_likes",
            ["person_id"],
        )

    # ── 7. feed_mentions ─────────────────────────────────────────
    if not _table_exists("feed_mentions"):
        op.create_table(
            "feed_mentions",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "mentioned_person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            # Exactly one of post_id / comment_id is set.
            sa.Column(
                "post_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_posts.id"),
                nullable=True,
            ),
            sa.Column(
                "comment_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_comments.id"),
                nullable=True,
            ),
            sa.Column(
                "author_person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            sa.Column(
                "created_at",
                sa.DateTime(timezone=True),
                nullable=False,
                server_default=sa.func.now(),
            ),
            sa.PrimaryKeyConstraint("id"),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_mentions_mentioned",
            "feed_mentions",
            ["mentioned_person_id", "created_at"],
        )
        op.create_index(
            "ix_feed_mentions_post",
            "feed_mentions",
            ["post_id"],
        )
        op.create_index(
            "ix_feed_mentions_comment",
            "feed_mentions",
            ["comment_id"],
        )

    # ── 8. feed_attachments ──────────────────────────────────────
    if not _table_exists("feed_attachments"):
        op.create_table(
            "feed_attachments",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "post_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_posts.id"),
                nullable=False,
            ),
            sa.Column("s3_key", sa.String(length=512), nullable=False),
            sa.Column("mime_type", sa.String(length=64), nullable=True),
            sa.Column("size_bytes", sa.Integer(), nullable=True),
            sa.Column("width", sa.Integer(), nullable=True),
            sa.Column("height", sa.Integer(), nullable=True),
            sa.Column(
                "position",
                sa.Integer(),
                nullable=False,
                server_default="0",
            ),
            sa.Column(
                "created_at",
                sa.DateTime(timezone=True),
                nullable=False,
                server_default=sa.func.now(),
            ),
            sa.PrimaryKeyConstraint("id"),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_attachments_post",
            "feed_attachments",
            ["post_id", "position"],
        )

    # ── 9. feed_blessings ────────────────────────────────────────
    if not _table_exists("feed_blessings"):
        op.create_table(
            "feed_blessings",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column(
                "celebration_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_celebrations.id"),
                nullable=False,
            ),
            sa.Column(
                "sender_person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            sa.Column("sticker_key", sa.String(length=32), nullable=False),
            sa.Column("message", sa.Text(), nullable=False),
            # The mirror comment that this blessing produced on the
            # corresponding system post. Lets us cascade soft-delete
            # via blessing → comment.
            sa.Column(
                "mirror_comment_id",
                sa.CHAR(length=36),
                sa.ForeignKey("feed_comments.id"),
                nullable=True,
            ),
            sa.Column(
                "created_at",
                sa.DateTime(timezone=True),
                nullable=False,
                server_default=sa.func.now(),
            ),
            sa.PrimaryKeyConstraint("id"),
            sa.UniqueConstraint(
                "celebration_id", "sender_person_id",
                name="uq_feed_blessings_celebration_sender",
            ),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_blessings_celebration",
            "feed_blessings",
            ["celebration_id"],
        )

    # ── 10. feed_announcements ───────────────────────────────────
    if not _table_exists("feed_announcements"):
        op.create_table(
            "feed_announcements",
            sa.Column("id", sa.CHAR(length=36), nullable=False),
            sa.Column("title", sa.String(length=256), nullable=False),
            sa.Column("body_markdown", sa.Text(), nullable=False),
            # general | policy | event | benefit | urgent
            sa.Column(
                "category",
                sa.String(length=32),
                nullable=False,
                server_default="general",
            ),
            # CSS color token name (rendered on the sidebar card border).
            sa.Column(
                "color",
                sa.String(length=32),
                nullable=False,
                server_default="default",
            ),
            sa.Column(
                "is_pinned",
                sa.Boolean(),
                nullable=False,
                server_default=sa.text("0"),
            ),
            sa.Column(
                "publish_at",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
            sa.Column(
                "expire_at",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
            # draft | published | archived
            sa.Column(
                "status",
                sa.String(length=16),
                nullable=False,
                server_default="published",
            ),
            sa.Column(
                "attachments_json",
                sa.JSON(),
                nullable=True,
                comment="List of {s3_key,name,size_bytes,mime_type}",
            ),
            sa.Column(
                "author_person_id",
                sa.CHAR(length=36),
                sa.ForeignKey("identity_persons.id"),
                nullable=False,
            ),
            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(),
                server_onupdate=sa.func.now(),
            ),
            sa.Column(
                "deleted_at",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
            sa.PrimaryKeyConstraint("id"),
            **table_kw_for("identity_persons"),
        )
        op.create_index(
            "ix_feed_announcements_status_publish",
            "feed_announcements",
            ["status", "publish_at"],
        )
        op.create_index(
            "ix_feed_announcements_pinned",
            "feed_announcements",
            ["is_pinned", "publish_at"],
        )
        op.create_index(
            "ix_feed_announcements_expire",
            "feed_announcements",
            ["expire_at"],
        )

    # ── 11. Seed scheduled_jobs rows ─────────────────────────────
    _seed_scheduled_job(
        name="feed.scan_today_celebrations",
        display_name="动态广场 扫描今日生日与入职周年",
        description=(
            "每小时执行,扫描在自身本地时区里 occasion = 今天 的活跃员工,"
            "为每个人 upsert feed_celebrations 行并生成对应的 system_post + "
            "聚合 banner post。idempotent (UNIQUE 约束)。"
        ),
        handler_key="feed.scan_today_celebrations",
        cron_expr="0 * * * *",
        config={},
    )
    _seed_scheduled_job(
        name="feed.notify_celebrants_and_leads",
        display_name="动态广场 推送祝福通知",
        description=(
            "紧随扫描任务执行,给寿星本人/manager/HRBP 推送站内+飞书通知,"
            "每个 celebration 仅推一次 (notify_state_json 持久化)。"
        ),
        handler_key="feed.notify_celebrants_and_leads",
        cron_expr="10 * * * *",
        config={},
    )
    _seed_scheduled_job(
        name="feed.expire_pinned_announcements",
        display_name="动态广场 公告自动上下架",
        description=(
            "每 10 分钟扫描 feed_announcements,根据 publish_at/expire_at "
            "翻转 is_pinned + status。"
        ),
        handler_key="feed.expire_pinned_announcements",
        cron_expr="*/10 * * * *",
        config={},
    )


def _seed_scheduled_job(
    *,
    name: str,
    display_name: str,
    description: str,
    handler_key: str,
    cron_expr: str,
    config: dict,
) -> None:
    bind = op.get_bind()
    bind.execute(
        sa.text(
            "INSERT INTO scheduled_jobs "
            "(id, name, display_name, description, handler_key, trigger_type, "
            "cron_expr, timezone, enabled, config, run_count, failure_count, "
            "created_at, updated_at) "
            "SELECT :id, :name, :display, :desc, :handler, 'cron', "
            ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS "
            "(SELECT 1 FROM scheduled_jobs WHERE name = :name)"
        ),
        {
            "id": str(uuid.uuid4()),
            "name": name,
            "display": display_name,
            "desc": description,
            "handler": handler_key,
            "cron": cron_expr,
            "config": json.dumps(config),
        },
    )


def downgrade() -> None:
    # Drop scheduled_jobs seed rows first.
    bind = op.get_bind()
    bind.execute(
        sa.text(
            "DELETE FROM scheduled_jobs WHERE handler_key IN ("
            "'feed.scan_today_celebrations',"
            "'feed.notify_celebrants_and_leads',"
            "'feed.expire_pinned_announcements')"
        )
    )

    # Drop tables in reverse FK order.
    for tbl in (
        "feed_announcements",
        "feed_blessings",
        "feed_attachments",
        "feed_mentions",
        "feed_likes",
        "feed_comments",
        "feed_posts",
        "feed_celebrations",
        "identity_person_preferences",
    ):
        if _table_exists(tbl):
            op.drop_table(tbl)

    if _column_exists("identity_persons", "birthday"):
        op.drop_column("identity_persons", "birthday")