-- Author: fruit.com.cn

-- FruitShorts 短剧服务 · 分区维护（PostgreSQL 18）
--
-- 为什么要单独一个脚本：
--   声明式分区只定义了"分片规则"，不会自动建分区。如果只建了 default 分区，
--   所有数据都会堆在 default 里，分区等于白做，而且越晚拆越贵。
--   所以这里提供三个东西：建分区的函数、清理过期分区的函数、以及现有库的转换脚本。
--
-- 应用侧也有兜底：com.fruitshorts.framework.partition.PartitionMaintainer
-- 启动时 + 每天凌晨各跑一次 ensure，保证不需要人工介入也不会漏建月份。
--
-- 幂等：可重复执行。

-- ============================================================
-- 一、分区管理函数
-- ============================================================

/**
 * 为按月分区的表补齐「从 p_from 所在月开始、连续 p_months 个月」的分区。
 * 分区命名 <table>_YYYYMM，已存在则跳过，返回实际新建的个数。
 */
CREATE OR REPLACE FUNCTION ensure_monthly_partitions(
    p_table        text,
    p_from         date DEFAULT CURRENT_DATE,
    p_months       int  DEFAULT 3
) RETURNS int AS $$
DECLARE
    i        int;
    v_start  date;
    v_end    date;
    v_name   text;
    v_created int := 0;
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_partitioned_table pt
                   JOIN pg_class c ON c.oid = pt.partrelid
                   WHERE c.relname = p_table) THEN
        RAISE EXCEPTION '% 不是分区表，无法维护分区', p_table;
    END IF;

    FOR i IN 0 .. (p_months - 1) LOOP
        v_start := (date_trunc('month', p_from) + (i || ' month')::interval)::date;
        v_end   := (v_start + interval '1 month')::date;
        v_name  := format('%s_%s', p_table, to_char(v_start, 'YYYYMM'));
        IF NOT EXISTS (SELECT 1 FROM pg_class WHERE relname = v_name) THEN
            EXECUTE format('CREATE TABLE %I PARTITION OF %I FOR VALUES FROM (%L) TO (%L)',
                           v_name, p_table, v_start, v_end);
            v_created := v_created + 1;
        END IF;
    END LOOP;
    RETURN v_created;
END;
$$ LANGUAGE plpgsql;

/**
 * 删除 p_before 之前的月分区，返回删除个数。
 * 只认 <table>_YYYYMM 这种命名的分区，default 分区不动。
 */
CREATE OR REPLACE FUNCTION drop_partitions_before(
    p_table  text,
    p_before date
) RETURNS int AS $$
DECLARE
    rec     record;
    dropped int := 0;
    cutoff  text := to_char(p_before, 'YYYYMM');
BEGIN
    FOR rec IN
        SELECT c.relname
        FROM pg_inherits i
        JOIN pg_class c ON c.oid = i.inhrelid
        JOIN pg_class p ON p.oid = i.inhparent
        WHERE p.relname = p_table
          AND c.relname ~ ('^' || p_table || '_[0-9]{6}$')
    LOOP
        IF substring(rec.relname from '([0-9]{6})$') < cutoff THEN
            EXECUTE format('DROP TABLE IF EXISTS %I', rec.relname);
            dropped := dropped + 1;
        END IF;
    END LOOP;
    RETURN dropped;
END;
$$ LANGUAGE plpgsql;

-- ============================================================
-- 二、把两张高写入表转成按月分区（只处理空表，有数据直接中止）
-- ============================================================

DO $$
BEGIN
    -- playback_ticket：每次签发播放票据一行，1 万并发下量级最大
    IF EXISTS (SELECT 1 FROM information_schema.tables
               WHERE table_schema = 'public' AND table_name = 'playback_ticket')
       AND NOT EXISTS (SELECT 1 FROM pg_partitioned_table pt
                       JOIN pg_class c ON c.oid = pt.partrelid
                       WHERE c.relname = 'playback_ticket') THEN
        IF (SELECT count(*) FROM playback_ticket) > 0 THEN
            RAISE EXCEPTION 'playback_ticket 已有数据，请先归档再转分区（勿直接删表）';
        END IF;
        DROP TABLE playback_ticket;
    END IF;

    CREATE TABLE IF NOT EXISTS playback_ticket (
      id         BIGINT,
      ticket_no  VARCHAR(40)  NOT NULL,
      user_id    BIGINT       NOT NULL DEFAULT 0,
      drama_id   BIGINT       NOT NULL,
      episode_id BIGINT       NOT NULL,
      quality    VARCHAR(16)  NOT NULL DEFAULT 'AUTO',
      play_url   VARCHAR(1024) NOT NULL DEFAULT '',
      expire_at  TIMESTAMP(3) NOT NULL,
      ip         VARCHAR(64)  NOT NULL DEFAULT '',
      device_id  VARCHAR(128) NOT NULL DEFAULT '',
      status     VARCHAR(16)  NOT NULL DEFAULT 'ISSUED',
      created_at TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
      PRIMARY KEY (id, created_at)
    ) PARTITION BY RANGE (created_at);

    -- 分区表的唯一索引必须包含分区键，所以带上 created_at；
    -- ticket_no 是 UUID，实际唯一性不受影响。
    CREATE UNIQUE INDEX IF NOT EXISTS uk_playback_ticket_no
        ON playback_ticket (ticket_no, created_at);
    CREATE INDEX IF NOT EXISTS idx_playback_ticket_user
        ON playback_ticket (user_id, created_at DESC);

    -- order_event：订单状态流转流水，支付争议的证据链
    IF EXISTS (SELECT 1 FROM information_schema.tables
               WHERE table_schema = 'public' AND table_name = 'order_event')
       AND NOT EXISTS (SELECT 1 FROM pg_partitioned_table pt
                       JOIN pg_class c ON c.oid = pt.partrelid
                       WHERE c.relname = 'order_event') THEN
        IF (SELECT count(*) FROM order_event) > 0 THEN
            RAISE EXCEPTION 'order_event 已有数据，请先归档再转分区（勿直接删表）';
        END IF;
        DROP TABLE order_event;
    END IF;

    CREATE TABLE IF NOT EXISTS order_event (
      id          BIGINT,
      order_no    VARCHAR(40)  NOT NULL,
      event       VARCHAR(32)  NOT NULL,
      from_status VARCHAR(16)  NOT NULL DEFAULT '',
      to_status   VARCHAR(16)  NOT NULL DEFAULT '',
      source      VARCHAR(16)  NOT NULL DEFAULT 'SERVER',
      payload     TEXT,
      operator_id BIGINT       NOT NULL DEFAULT 0,
      created_at  TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
      PRIMARY KEY (id, created_at)
    ) PARTITION BY RANGE (created_at);

    CREATE INDEX IF NOT EXISTS idx_order_event_no
        ON order_event (order_no, created_at);
END $$;

-- ============================================================
-- 三、预建当前月 + 未来 3 个月的分区
-- ============================================================

SELECT ensure_monthly_partitions('watch_history',   CURRENT_DATE, 4);
SELECT ensure_monthly_partitions('playback_ticket', CURRENT_DATE, 4);
SELECT ensure_monthly_partitions('order_event',     CURRENT_DATE, 4);

-- ============================================================
-- 四、保留策略（默认只演示，不自动删；上线后由应用定时任务按配置执行）
-- ============================================================
-- 需要手工清理时执行：
--   SELECT drop_partitions_before('watch_history',   CURRENT_DATE - INTERVAL '24 months');
--   SELECT drop_partitions_before('playback_ticket', CURRENT_DATE - INTERVAL '6 months');
--   SELECT drop_partitions_before('order_event',     CURRENT_DATE - INTERVAL '36 months');