Add ft service stats db updates

migrations/.current-alembic-head

@@ -1 +1 @@
-0551_drop_ntfcns_failed_idx
+0554_create_service_stats

migrations/versions/0552_create_replacation_slot.py

@@ -0,0 +1,22 @@
+"""
+Create Date: 2026-05-14T16:39:58
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision = "0552_create_replacation_slot"
+down_revision = "0551_drop_ntfcns_failed_idx"
+
+
+def upgrade():
+    op.execute(
+        "SELECT * FROM pg_create_logical_replication_slot('notify_dashboard_replication_slot', 'wal2json');"
+    )
+
+
+def downgrade():
+    op.execute(
+        "SELECT * FROM pg_drop_replication_slot('notify_dashboard_replication_slot');"
+    )

migrations/versions/0553_notifications_id_status_idx.py

@@ -0,0 +1,64 @@
+"""
+Create Date: 2026-05-19T00:00:00
+"""
+
+from alembic import op
+
+revision = "0553_notifications_id_status_idx"
+down_revision = "0552_create_replacation_slot"
+
+
+def upgrade():
+    op.execute("SET lock_timeout = '1s';")
+    op.execute("SET statement_timeout = '5s';")
+
+    # We need a unique index on (id, notification_status) for REPLICA IDENTITY USING INDEX.
+    # Build a validated check first so PostgreSQL can avoid a table scan when setting NOT NULL.
+    with op.get_context().autocommit_block():
+        op.execute("BEGIN;")
+        op.execute(
+            "ALTER TABLE notifications "
+            "ADD CONSTRAINT ck_notifications_notification_status_not_null "
+            "CHECK (notification_status IS NOT NULL) NOT VALID;"
+        )
+        op.execute("COMMIT;")
+
+    with op.get_context().autocommit_block():
+        op.execute("BEGIN;")
+        op.execute(
+            "ALTER TABLE notifications "
+            "VALIDATE CONSTRAINT ck_notifications_notification_status_not_null;"
+        )
+        op.execute("COMMIT;")
+
+    op.execute("ALTER TABLE notifications ALTER COLUMN notification_status SET NOT NULL;")
+
+    with op.get_context().autocommit_block():
+        op.execute(
+            "CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "
+            "ix_notifications_id_notification_status "
+            "ON notifications (id, notification_status);"
+        )
+
+    op.execute("ALTER TABLE notifications REPLICA IDENTITY USING INDEX ix_notifications_id_notification_status;")
+
+
+def downgrade():
+    op.execute("ALTER TABLE notifications REPLICA IDENTITY DEFAULT;")
+
+    op.execute("ALTER TABLE notifications ALTER COLUMN notification_status DROP NOT NULL;")
+    op.execute("ALTER TABLE notifications DROP CONSTRAINT IF EXISTS ck_notifications_notification_status_not_null;")
+
+    with op.get_context().autocommit_block():
+        op.drop_index(
+            "ix_notifications_id_notification_status",
+            table_name="notifications",
+            postgresql_concurrently=True,
+        )
+        op.create_index(
+            "ix_notifications_id_notification_status",
+            "notifications",
+            ["id", "notification_status"],
+            unique=False,
+            postgresql_concurrently=True,
+        )

migrations/versions/0554_create_service_stats.py

@@ -0,0 +1,69 @@
+"""
+Create service_stats table for aggregated notification status counts.
+
+Revision ID: 0554_create_service_stats
+Revises: 0553_notifications_id_status_idx
+Create Date: 2026-05-19 00:00:00
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision = "0554_create_service_stats"
+down_revision = "0553_notifications_id_status_idx"
+
+
+def upgrade():
+    op.create_table(
+        "ft_service_stats",
+        sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("service_id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("template_id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("notification_type", postgresql.ENUM(name="notification_type", create_type=False), nullable=False),
+        sa.Column("notification_status", sa.Text(), nullable=False),
+        sa.Column("count", sa.BigInteger(), nullable=False, server_default=sa.text("0")),
+        sa.ForeignKeyConstraint(["service_id"], ["services.id"]),
+        sa.ForeignKeyConstraint(["template_id"], ["templates.id"]),
+        sa.ForeignKeyConstraint(["notification_status"], ["notification_status_types.name"]),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "service_id",
+            "template_id",
+            "notification_type",
+            "notification_status",
+            name="uix_ft_service_stats_dimensions",
+        ),
+    )
+
+    op.create_index(
+        "ix_ft_svc_stats_svc_ntype_nstatus",
+        "ft_service_stats",
+        ["service_id", "notification_type", "notification_status"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_ft_svc_stats_tmpl_ntype_nstatus",
+        "ft_service_stats",
+        ["template_id", "notification_type", "notification_status"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_ft_service_stats_service_id_template_id",
+        "ft_service_stats",
+        ["service_id", "template_id"],
+        unique=False,
+    )
+
+
+def downgrade():
+    op.drop_index("ix_ft_service_stats_service_id_template_id", table_name="ft_service_stats")
+    op.drop_index(
+        "ix_ft_svc_stats_tmpl_ntype_nstatus",
+        table_name="ft_service_stats",
+    )
+    op.drop_index(
+        "ix_ft_svc_stats_svc_ntype_nstatus",
+        table_name="ft_service_stats",
+    )
+    op.drop_table("ft_service_stats")