Process replication slot changes

[DilwoarH]

** DO NOT MERGE UNTIL #4946 IS MERGED IN FIRST**

What

The application code for processing slot changes for notifications table

Why

This is needed to process WAL logs in the database

Test results

Ran a load test on dev-c - this is the result after 2400 emails.

SELECT
	(SELECT count(*) FROM pg_logical_slot_peek_changes('notify_dashboard_replication_slot', NULL, NULL)) AS still_processing,
    (SELECT SUM(notification_count) FROM ft_service_stats) AS ft_stats_count,
    COUNT(*) AS notif_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'created') AS ft_created_count,
    COUNT(CASE WHEN notification_status = 'created' THEN 1 END) AS notif_created_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'sending') AS ft_sending_count,
    COUNT(CASE WHEN notification_status = 'sending' THEN 1 END) AS notif_sending_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'delivered') AS ft_delivered_count,
    COUNT(CASE WHEN notification_status = 'delivered' THEN 1 END) AS notif_delivered_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'temporary-failure') AS ft_temp_fail_count,
    COUNT(CASE WHEN notification_status = 'temporary-failure' THEN 1 END) AS notif_temp_fail_count,
    (SELECT SUM(notification_count) FROM ft_service_stats where notification_status = 'permanent-failure') AS ft_perm_fail_count,
    COUNT(CASE WHEN notification_status = 'permanent-failure' THEN 1 END) AS notif_perm_fail_count
FROM
    public.notifications;

still_processing	ft_stats_count	notif_count	ft_created_count	notif_created_count	ft_sending_count	notif_sending_count	ft_delivered_count	notif_delivered_count	ft_temp_fail_count	notif_temp_fail_count	ft_perm_fail_count	notif_perm_fail_count
0	2400	2400	0	0	0	0	1206	1206	234	234	960	960

How it works

[DilwoarH]

requires #4936 to be merged first

[spatel033]

filters is used only in else block so better to move inside else block to avoid evaluating the SQLAlchemy expressions on increments

[spatel033]

There is a confusing order swap between template_id and service_id, is this intentional?

[DilwoarH]

no - good spot

[spatel033]

Do we actually need this condition check? Updated notifications always have a previous notification_status from the database, this condition seems redundant and a bit confusing.