leocrm/alembic/versions/0120_notification_to_comm_system_channel.py

"""Add is_system column to comm_conversations and migrate notifications to system channel (B-NOTIF-*).

Adds is_system boolean to comm_conversations for system channel support.
Migrates existing notifications into the system channel as CommMessages.
Creates a view notifications_legacy as a compatibility layer over the old notifications table.

Revision ID: 0120
Revises: 0119
"""

from alembic import op
import sqlalchemy as sa

revision = "0120"
down_revision = "0119"
branch_labels = None
depends_on = None


def _table_exists(conn, table_name: str) -> bool:
    """True when the table exists (dual-path convergence, Gate B).

    On a fresh install the kommunikation plugin SQL migration has not run
    yet when Alembic reaches this revision — skip the comm_* parts instead
    of failing. The plugin-side migration adds the same column idempotently.
    """
    row = conn.execute(
        sa.text("SELECT to_regclass(:tname) IS NOT NULL"),
        {"tname": f"public.{table_name}"},
    ).scalar()
    return bool(row)


def upgrade() -> None:
    conn = op.get_bind()
    if _table_exists(conn, "comm_conversations"):
        # 1. Add is_system column to comm_conversations
        op.add_column(
            "comm_conversations",
            sa.Column("is_system", sa.Boolean(), nullable=False, server_default=sa.text("false")),
        )
        op.create_index(
            "ix_comm_conversations_tenant_system",
            "comm_conversations",
            ["tenant_id", "is_system"],
        )

        # 2. Create system channel per tenant (for tenants that have notifications)
        op.execute("""
            INSERT INTO comm_conversations (id, tenant_id, title, is_pinned, is_locked, is_direct, is_archived, is_system, created_by, created_by_type, metadata, created_at, updated_at)
            SELECT
                gen_random_uuid(),
                n.tenant_id,
                'System Channel',
                false,
                true,
                false,
                false,
                true,
                NULL,
                'system',
                '{}'::jsonb,
                NOW(),
                NOW()
            FROM (
                SELECT DISTINCT tenant_id FROM notifications WHERE deleted_at IS NULL
            ) n
            WHERE NOT EXISTS (
                SELECT 1 FROM comm_conversations cc
                WHERE cc.tenant_id = n.tenant_id AND cc.is_system = true AND cc.deleted_at IS NULL
            );
        """)

        # 3. Insert notifications as CommMessages in the system channel
        op.execute("""
            INSERT INTO comm_messages (id, tenant_id, conversation_id, sender_id, sender_type, content, content_format, metadata, created_at, updated_at)
            SELECT
                gen_random_uuid(),
                n.tenant_id,
                sc.id,
                n.user_id,
                'system',
                COALESCE(n.title, '') || CASE WHEN n.body IS NOT NULL THEN E'\n' || n.body ELSE '' END,
                'text',
                jsonb_build_object(
                    'notification_type', n.type,
                    'severity', 'info',
                    'entity_ref', CASE WHEN n.entity_type IS NOT NULL THEN jsonb_build_object('entity_type', n.entity_type, 'entity_id', n.entity_id::text) ELSE NULL END,
                    'migrated_from_notification', true,
                    'original_notification_id', n.id::text
                ),
                n.created_at,
                COALESCE(n.read_at, n.created_at)
            FROM notifications n
            JOIN comm_conversations sc ON sc.tenant_id = n.tenant_id AND sc.is_system = true AND sc.deleted_at IS NULL
            WHERE n.deleted_at IS NULL;
        """)

        # 4. Insert text blocks for each migrated message
        op.execute("""
            INSERT INTO comm_message_blocks (id, tenant_id, message_id, block_type, block_data, sort_order)
            SELECT
                gen_random_uuid(),
                cm.tenant_id,
                cm.id,
                'text',
                jsonb_build_object('text', cm.content),
                0
            FROM comm_messages cm
            WHERE cm.metadata->>'migrated_from_notification' = 'true';
        """)

        # 5. Insert action_card blocks for messages with entity references
        op.execute("""
            INSERT INTO comm_message_blocks (id, tenant_id, message_id, block_type, block_data, sort_order)
            SELECT
                gen_random_uuid(),
                cm.tenant_id,
                cm.id,
                'action_card',
                jsonb_build_object(
                    'label', 'Open',
                    'entity_type', (cm.metadata->'entity_ref'->>'entity_type'),
                    'entity_id', (cm.metadata->'entity_ref'->>'entity_id')
                ),
                1
            FROM comm_messages cm
            WHERE cm.metadata->>'migrated_from_notification' = 'true'
              AND cm.metadata->'entity_ref' IS NOT NULL;
        """)

        # 6. For read notifications, create CommMessageRead entries
        op.execute("""
            INSERT INTO comm_message_reads (id, tenant_id, conversation_id, user_id, last_read_msg_id, last_read_at)
            SELECT
                gen_random_uuid(),
                cm.tenant_id,
                cm.conversation_id,
                cm.sender_id,
                cm.id,
                COALESCE(n.read_at, n.created_at)
            FROM comm_messages cm
            JOIN notifications n ON n.id::text = cm.metadata->>'original_notification_id'
            WHERE cm.metadata->>'migrated_from_notification' = 'true'
              AND n.read_at IS NOT NULL
              AND n.deleted_at IS NULL;
        """)

    # 7. Legacy view over the CORE notifications table — exists on both paths
    op.execute("DROP VIEW IF EXISTS notifications_legacy")
    op.execute("CREATE VIEW notifications_legacy AS SELECT * FROM notifications")


def downgrade() -> None:
    conn = op.get_bind()
    op.execute("DROP VIEW IF EXISTS notifications_legacy")
    if not _table_exists(conn, "comm_conversations"):
        return
    op.execute("DELETE FROM comm_message_blocks WHERE message_id IN (SELECT id FROM comm_messages WHERE metadata->>'migrated_from_notification' = 'true')")
    op.execute("DELETE FROM comm_messages WHERE metadata->>'migrated_from_notification' = 'true'")
    op.execute("DELETE FROM comm_conversations WHERE is_system = true AND title = 'System Channel'")
    op.drop_index("ix_comm_conversations_tenant_system", table_name="comm_conversations")
    op.drop_column("comm_conversations", "is_system")