-- IM private-message Snowflake ID compatibility migration.
-- Scope: the shared business database (for example meeting_im).
-- Review and back up before execution. This script does not replay dead letters.

-- 1. Preflight: every listed id column must become BIGINT.
SELECT TABLE_NAME, COLUMN_TYPE, IS_NULLABLE, EXTRA
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE()
  AND (TABLE_NAME = 'im_messages' OR TABLE_NAME REGEXP '^im_[0-9]+_messages$')
  AND COLUMN_NAME = 'id'
ORDER BY TABLE_NAME;

-- 2. Upgrade the template/base table used by new tenant shards.
ALTER TABLE `im_messages`
  MODIFY COLUMN `id` BIGINT NOT NULL AUTO_INCREMENT;

-- 3. Upgrade every existing tenant private-message shard.
DELIMITER $$

DROP PROCEDURE IF EXISTS migrate_private_message_id_to_bigint$$
CREATE PROCEDURE migrate_private_message_id_to_bigint()
BEGIN
    DECLARE done INT DEFAULT 0;
    DECLARE shard_table VARCHAR(64);
    DECLARE shard_cursor CURSOR FOR
        SELECT TABLE_NAME
        FROM information_schema.TABLES
        WHERE TABLE_SCHEMA = DATABASE()
          AND TABLE_NAME REGEXP '^im_[0-9]+_messages$'
        ORDER BY TABLE_NAME;
    DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1;

    OPEN shard_cursor;
    shard_loop: LOOP
        FETCH shard_cursor INTO shard_table;
        IF done = 1 THEN
            LEAVE shard_loop;
        END IF;
        SET @alter_message_id_sql = CONCAT(
            'ALTER TABLE `', REPLACE(shard_table, '`', '``'),
            '` MODIFY COLUMN `id` BIGINT NOT NULL AUTO_INCREMENT'
        );
        PREPARE alter_message_id_stmt FROM @alter_message_id_sql;
        EXECUTE alter_message_id_stmt;
        DEALLOCATE PREPARE alter_message_id_stmt;
    END LOOP;
    CLOSE shard_cursor;
END$$

CALL migrate_private_message_id_to_bigint()$$
DROP PROCEDURE migrate_private_message_id_to_bigint$$

DELIMITER ;

-- 4. Post-check: this query must return zero rows.
SELECT TABLE_NAME, COLUMN_TYPE, IS_NULLABLE, EXTRA
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE()
  AND (TABLE_NAME = 'im_messages' OR TABLE_NAME REGEXP '^im_[0-9]+_messages$')
  AND COLUMN_NAME = 'id'
  AND DATA_TYPE <> 'bigint';

-- 5. Observe pending recovery work. Replay only after the post-check passes.
SELECT status, last_reason, COUNT(*) AS message_count,
       MIN(created_at) AS first_created_at, MAX(created_at) AS last_created_at
FROM im_kafka_dead_letters
WHERE event_type = 'message.persist.p2p'
GROUP BY status, last_reason
ORDER BY status, last_reason;