First draft to purge high volume data on an hourly basis.

This makes the assumption that we do not need NotificationHistory, we will need to ensure that for HighVolumeServices we do not recalculate the stats tables, either in the reporting_tasks nor in the commands.
The strategy to migrate to using this code is still up for design.
We will need to consider:
 - how to purge the data, safely, efficiently.
 - ensure we get the counts right for the stats table, the ft_notification_status table is particularly tricky, because as a notification moves from created -> sending -> delivered the row counts change for each status type. One idea is that we only write the `delivered`, `temporary-failure` and `permanent-failure` status into the table.

This is not production ready but just a draft to get the conversation going.
I also have an idea to stop returning "todays" notification data for HighVolumeServices on the dashboards/usages (except platform_admin) until this code is ready.

app/celery/reporting_tasks.py

@@ -42,6 +42,8 @@ def create_nightly_billing(day_start=None):
 @notify_celery.task(name="create-nightly-billing-for-day")
 @statsd(namespace="tasks")
 def create_nightly_billing_for_day(process_day):
+    # When app.celery.scheduled_tasks.purge_high_volume_notifications starts
+    # we need to exclude HighVolumeServices from the list.
     process_day = datetime.strptime(process_day, "%Y-%m-%d").date()
 
     start = datetime.utcnow()
@@ -62,6 +64,8 @@ def create_nightly_billing_for_day(process_day):
 @cronitor("create-nightly-notification-status")
 @statsd(namespace="tasks")
 def create_nightly_notification_status():
+    # When app.celery.scheduled_tasks.purge_high_volume_notifications starts
+    # we need to exclude HighVolumeServices from the list.
     yesterday = convert_utc_to_bst(datetime.utcnow()).date() - timedelta(days=1)
 
     # email and sms
@@ -84,6 +88,8 @@ def create_nightly_notification_status():
 @notify_celery.task(name="create-nightly-notification-status-for-day")
 @statsd(namespace="tasks")
 def create_nightly_notification_status_for_day(process_day, notification_type):
+    # When app.celery.scheduled_tasks.purge_high_volume_notifications starts
+    # we need to exclude HighVolumeServices from the list.
     process_day = datetime.strptime(process_day, "%Y-%m-%d").date()
 
     start = datetime.utcnow()

app/celery/scheduled_tasks.py

@@ -5,10 +5,11 @@ from datetime import (
 
 from flask import current_app
 from notifications_utils.statsd_decorators import statsd
+from notifications_utils.timezones import convert_utc_to_bst
 from sqlalchemy import and_
 from sqlalchemy.exc import SQLAlchemyError
 
-from app import notify_celery, zendesk_client
+from app import notify_celery, zendesk_client, db
 from app.celery.tasks import (
     process_job,
     get_recipient_csv_and_template_and_sender_id,
@@ -45,7 +46,7 @@ from app.models import (
     JOB_STATUS_ERROR,
     SMS_TYPE,
     EMAIL_TYPE,
-)
+    HighVolumeService, Template, Notification, KEY_TYPE_NORMAL, FactBilling, FactNotificationStatus)
 from app.notifications.process_notifications import send_notification_to_queue
 from app.v2.errors import JobIncompleteError
 
@@ -311,3 +312,112 @@ def check_for_services_with_high_failure_rates_or_sending_to_tv_numbers():
                 message=message,
                 ticket_type=zendesk_client.TYPE_INCIDENT
             )
+
+
+def purge_high_volume_notifications():
+    # Not sure what time to use here.....
+    hour_ago = datetime.utcnow() - timedelta(hours=1)
+
+    # We could also store the id in config... however this does give us a bit of control,
+    # as in we can stop this process from happening by deleting the row.
+    # The other part of it is to potentially take this tactic for other services...
+    # perhaps hour_ago is actually the timedelta from the table.
+    services = HighVolumeService.query.all()
+
+    for service in services:
+        templates = Template.query.filter_by(
+            service_id=service.service_id,
+            # going to assume we are only dealing with emails because I don't want to deal with rates at this time.
+            template_type=EMAIL_TYPE
+        ).all()
+        for status in ['delivered', 'temporary-failure', 'permanent-failure']:
+            for template in templates:
+                del_count = Notification.query.filter(
+                    Notification.service_id == service.service_id,
+                    Notification.template_id == template.id,
+                    Notification.notification_type == template.template_type,
+                    Notification.status == status,
+                    Notification.key_type == KEY_TYPE_NORMAL,
+                    Notification.created_at < hour_ago,
+                ).delete()
+
+                bst_date = convert_utc_to_bst(hour_ago).date()
+                # upsert stat data
+                upsert_ft_billing(bst_date, del_count, service.service_id, template)
+                upsert_ft_notification_status(bst_date, del_count, service.service_id, status, template)
+
+
+def upsert_ft_notification_status(bst_date, del_count, service_id, status, template):
+    ft_status_row = FactNotificationStatus.query.filter(
+        FactNotificationStatus.service_id == service_id,
+        FactNotificationStatus.template_id == template.id,
+        FactNotificationStatus.bst_date == bst_date,
+        FactNotificationStatus.notification_type == template.template_type,
+        FactNotificationStatus.notification_status == status,
+        FactNotificationStatus.key_type == KEY_TYPE_NORMAL
+    ).first()
+    if ft_status_row:
+        # How do we deal with rows in ft_status where there are.... this isn't going to work.
+        # The current process take the days total from Notifications,
+        # deletes the row for status then inserts,
+        # this means that any rows with a created status should eventually go away.
+        # Do we stop processing the HighVolumes services in the reporting tasks?
+        # If yes, then we need to thing about the migration plan, take a snapshot of data first, etc.
+        FactNotificationStatus.query.filter(
+            FactNotificationStatus.service_id == service_id,
+            FactNotificationStatus.template_id == template.id,
+            FactNotificationStatus.bst_date == bst_date,
+            FactNotificationStatus.notification_type == template.template_type,
+            FactNotificationStatus.notification_status == status,
+            FactNotificationStatus.key_type == KEY_TYPE_NORMAL
+        ).update({"notification_count": ft_status_row.notification_count + del_count})
+    else:
+        ft_status = FactNotificationStatus(
+            bst_date=bst_date,
+            template_id=template.id,
+            service_id=service_id,
+            job_id='00000000-0000-0000-0000-000000000000',
+            notification_type=template.template_type,
+            key_type=KEY_TYPE_NORMAL,
+            notification_status=status,
+            notification_count=del_count,
+            created_at=datetime.utcnow()
+        )
+        db.session.add(ft_status)
+        db.session.commit()
+
+
+def upsert_ft_billing(bst_date, del_count, service_id, template):
+    ft_billing_row = FactBilling.query.filter(
+        FactBilling.service_id == service_id,
+        FactBilling.template_id == template.id,
+        FactBilling.bst_date == bst_date,
+        FactBilling.notification_type == template.template_type
+    ).first()
+    if not ft_billing_row:
+        # insert new row
+        ft_billing = FactBilling(
+            bst_date=bst_date,
+            service_id=service_id,
+            template_id=template.id,
+            notification_type=template.template_type,
+            provider='SES',
+            rate_multiplier=0,
+            international=False,
+            rate=0,
+            billable_units=0,
+            notifications_sent=del_count,
+            created_at=datetime.utcnow(),
+            postage='none'
+        )
+        db.session.add(ft_billing)
+        db.session.commit()
+
+    else:
+        FactBilling.query.filter(
+            FactBilling.service_id == service_id,
+            FactBilling.template_id == template.id,
+            FactBilling.bst_date == bst_date,
+            FactBilling.notification_type == template.template_type
+        ).update({'notifications_sent': ft_billing_row.notifications_sent + del_count})
+        db.session.commit()

app/commands.py

@@ -11,7 +11,7 @@ from click_datetime import Datetime as click_dt
 from flask import current_app, json
 from notifications_utils.recipients import RecipientCSV
 from notifications_utils.template import SMSMessageTemplate
-from sqlalchemy.exc import IntegrityError, SQLAlchemyError
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm.exc import NoResultFound
 from notifications_utils.statsd_decorators import statsd
 
@@ -45,6 +45,7 @@ from app.dao.templates_dao import dao_get_template_by_id
 from app.dao.users_dao import delete_model_user, delete_user_verify_codes, get_user_by_email
 from app.models import (
     PROVIDERS,
+    NOTIFICATION_CREATED,
     KEY_TYPE_TEST,
     SMS_TYPE,
     EMAIL_TYPE,
@@ -56,10 +57,6 @@ from app.models import (
     Service,
     EmailBranding,
     LetterBranding,
-    NOTIFICATION_CREATED,
-    NOTIFICATION_DELIVERED,
-    NOTIFICATION_PERMANENT_FAILURE,
-    NOTIFICATION_TEMPORARY_FAILURE,
 )
 from app.performance_platform.processing_time import send_processing_time_for_start_and_end
 from app.utils import get_london_midnight_in_utc, get_midnight_for_day_before
@@ -904,105 +901,3 @@ def process_row_from_job(job_id, job_row_number):
             notification_id = process_row(row, template, job, job.service)
             current_app.logger.info("Process row {} for job {} created notification_id: {}".format(
                 job_row_number, job_id, notification_id))
-
-
-@notify_command(name='delete-high-volume-service-data')
-@click.option('-i', '--service_id', required=True, help='Service id of the high volume service')
-@click.option('-s', '--start_date', required=True, type=click_dt(format='%Y-%m-%d %H'),
-              help='Start date of process YYYY-MM-DD HH:mm')
-@click.option('-e', '--end_date', required=True, type=click_dt(format='%Y-%m-%d %H'),
-              help='End date of process YYYY-MM-DD HH:MM')
-@click.option('-t', '--notification_type', required=False, default='email',
-              help='Notification type of the data to delete')
-def delete_high_volume_service_data(service_id, start_date, end_date, notification_type):
-    str_date = start_date.strftime('%Y_%m_%d_%H')
-    str_end_date = end_date.strftime('%Y_%m_%d_%H')
-
-    bkup_tble_name = f'back_up_notifications_{str_date}_to_{str_end_date}'
-
-    print(f"""Creating back up {bkup_tble_name} starting at: {datetime.utcnow()}
-          for {notification_type} notifications for service: {service_id}, starting at {start_date} and {end_date}
-         """)
-
-    _create_notification_bkup_table(bkup_tble_name, end_date, notification_type, service_id, start_date)
-
-    hour_start = start_date
-    hour_end = hour_start + timedelta(hours=1)
-
-    terminate_statuses = [NOTIFICATION_DELIVERED, NOTIFICATION_TEMPORARY_FAILURE, NOTIFICATION_PERMANENT_FAILURE]
-    delete_query = Notification.query.filter(
-        Notification.notification_type == notification_type,
-        Notification.service_id == service_id,
-        Notification.created_at >= hour_start,
-        Notification.created_at <= hour_end,
-        Notification.status.in_(terminate_statuses)
-    )
-    # Iterate hour by hour
-    del_count = 0
-    while hour_start < end_date:
-        del_count += delete_query.delete(synchronize_session=False)
-        db.session.commit()
-        # print(hour_start, hour_end)
-        # increment hour
-        hour_end = hour_end + timedelta(hours=1)
-        hour_start = hour_start + timedelta(hours=1)
-        delete_query = Notification.query.filter(
-            Notification.notification_type == notification_type,
-            Notification.service_id == service_id,
-            Notification.created_at >= hour_start,
-            Notification.created_at <= hour_end,
-            Notification.status.in_(terminate_statuses)
-        )
-
-    print(f"""Completed deleting {del_count} from notifications
-              for {notification_type} notifications for
-              service: {service_id}, starting at {start_date} and {end_date}
-             """)
-
-
-def _create_notification_bkup_table(bkup_tble_name, end_date, notification_type, service_id, start_date):
-    try:
-        create_tbl_sql = f"""
-            CREATE TABLE {bkup_tble_name} AS 
-                SELECT *
-                  FROM notifications
-                 WHERE service_id = :service_id
-                   AND notification_type = :notification_type
-                   AND created_at >= :start_date
-                   AND created_at <= :end_date
-                   AND key_type = 'normal'
-                   AND notification_status in ('delivered', 'permanent-failure', 'temporary-failure')
-        """
-        input_params = {
-            "service_id": service_id,
-            "notification_type": notification_type,
-            "start_date": start_date,
-            "end_date": end_date
-        }
-        db.session.execute(create_tbl_sql, input_params)
-        db.session.commit()
-    except SQLAlchemyError as e:
-        db.session.commit()  # terminate previous transaction
-        # This query isn't quite right yet, if the notifications are already deleted still get 0.
-        # however it doesn't cause any harm, because there is nothing to delete.
-        # But it will also return 0 if the rows exist in notifications
-        qry = f""" Select count(*) from notifications 
-          WHERE service_id = :service_id
-          AND notification_type = :notification_type
-          AND created_at >= :start_date
-          AND created_at <= :end_date
-          AND key_type = 'normal'
-          UNION 
-          SELECT count(*) 
-            FROM {bkup_tble_name} 
-        """
-        result = db.session.execute(qry, input_params).fetchall()
-        if result[0][0] == result[1][0]:
-            print("Table and data already exists, keep going")
-            return
-        else:
-            # This will throw the exception if the data is already deleted or partically deleted...
-            # but gives us a chance to see what's happending.
-            print(f"Table already exists but row counts are inconsistent is missing bail out. "
-                  f"There are {result[0][0]} rows in notifications and {result[1][0]} rows in {bkup_tble_name}")
-            raise e

app/dao/notifications_dao.py

@@ -307,14 +307,19 @@ def delete_notifications_older_than_retention_by_type(notification_type, qry_lim
     ).all()
     deleted = 0
     for f in flexible_data_retention:
-        current_app.logger.info(
-            "Deleting {} notifications for service id: {}".format(notification_type, f.service_id))
-
-        day_to_delete_backwards_from = get_london_midnight_in_utc(
+        days_of_retention = get_london_midnight_in_utc(
             convert_utc_to_bst(datetime.utcnow()).date()) - timedelta(days=f.days_of_retention)
 
-        deleted += _move_notifications_to_notification_history(
-            notification_type, f.service_id, day_to_delete_backwards_from, qry_limit)
+        if notification_type == LETTER_TYPE:
+            _delete_letters_from_s3(
+                notification_type, f.service_id, days_of_retention, qry_limit
+            )
+
+        insert_update_notification_history(notification_type, days_of_retention, f.service_id)
+
+        current_app.logger.info(
+            "Deleting {} notifications for service id: {}".format(notification_type, f.service_id))
+        deleted += _delete_notifications(notification_type, days_of_retention, f.service_id, qry_limit)
 
     current_app.logger.info(
         'Deleting {} notifications for services without flexible data retention'.format(notification_type))
@@ -324,54 +329,18 @@ def delete_notifications_older_than_retention_by_type(notification_type, qry_lim
     service_ids_to_purge = db.session.query(Service.id).filter(Service.id.notin_(services_with_data_retention)).all()
 
     for service_id in service_ids_to_purge:
-        deleted += _move_notifications_to_notification_history(
-            notification_type, service_id, seven_days_ago, qry_limit)
+        if notification_type == LETTER_TYPE:
+            _delete_letters_from_s3(
+                notification_type, service_id, seven_days_ago, qry_limit
+            )
+        insert_update_notification_history(notification_type, seven_days_ago, service_id)
+        deleted += _delete_notifications(notification_type, seven_days_ago, service_id, qry_limit)
 
     current_app.logger.info('Finished deleting {} notifications'.format(notification_type))
 
     return deleted
 
 
-def _move_notifications_to_notification_history(notification_type, service_id, day_to_delete_backwards_from, qry_limit):
-    deleted = 0
-    if notification_type == LETTER_TYPE:
-        _delete_letters_from_s3(
-            notification_type, service_id, day_to_delete_backwards_from, qry_limit
-        )
-
-    stop = -1  # exclusive, we want to include 0
-    step = -1
-    for hour_delta in range(23, stop, step):
-        # We find the timestamp we want to delete all notifications backwards from
-        # We then start 23 hours ago, and do an insert notification history before deleting all notifications older
-        # We then look 22 hours ago, do an insert notifications history before deleting all notifications older
-        # We continue this until we reach the original timestamp we wanted to delete notifications backwardsfrom
-        # This enables us to break this into smaller database queries
-        timestamp_to_delete_backwards_from = day_to_delete_backwards_from - timedelta(hours=hour_delta)
-
-        if service_id == '539d63a1-701d-400d-ab11-f3ee2319d4d4':
-            current_app.logger.info(
-                "Beginning insert_update_notification_history for GOV.UK Email from {} backwards".format(
-                    timestamp_to_delete_backwards_from
-                )
-            )
-
-        insert_update_notification_history(notification_type, timestamp_to_delete_backwards_from, service_id, qry_limit)
-
-        if service_id == '539d63a1-701d-400d-ab11-f3ee2319d4d4':
-            current_app.logger.info(
-                "Beginning _delete_notifications for GOV.UK Email {} backwards".format(
-                    timestamp_to_delete_backwards_from
-                )
-            )
-
-        deleted += _delete_notifications(
-            notification_type, timestamp_to_delete_backwards_from, service_id, qry_limit
-        )
-
-    return deleted
-
-
 def _delete_notifications(notification_type, date_to_delete_from, service_id, query_limit):
     subquery = db.session.query(
         Notification.id
@@ -420,8 +389,6 @@ def insert_update_notification_history(notification_type, date_to_delete_from, s
         Notification.service_id == service_id,
         Notification.created_at < date_to_delete_from,
         Notification.key_type != KEY_TYPE_TEST
-    ).order_by(
-        Notification.created_at
     )
     notifications_count = notification_query.count()
 

app/models.py

@@ -2148,3 +2148,10 @@ class ServiceContactList(db.Model):
             "created_at": created_at_in_bst.strftime("%Y-%m-%d %H:%M:%S"),
         }
         return contact_list
+
+
+class HighVolumeService(db.Model):
+    # Service that we want to purge data for hourly
+    __tablename__ = 'high_volume_service'
+
+    service_id = db.Column(UUID(as_uuid=True), primary_key=True, unique=True, index=True, nullable=False)

migrations/versions/0319_high_volume_service.py

@@ -0,0 +1,27 @@
+"""
+
+Revision ID: 0319_high_volume_service
+Revises: 0318_service_contact_list
+Create Date: 2020-03-20 08:53:22.624516
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision = '0319_high_volume_service'
+down_revision = '0318_service_contact_list'
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('high_volume_service',
+                    sa.Column('service_id', postgresql.UUID(as_uuid=True), nullable=False),
+                    sa.PrimaryKeyConstraint('service_id')
+                    )
+    op.create_index(op.f('ix_high_volume_service_service_id'), 'high_volume_service', ['service_id'], unique=True)
+
+
+def downgrade():
+    op.drop_index(op.f('ix_high_volume_service_service_id'), table_name='high_volume_service')
+    op.drop_table('high_volume_service')

tests/app/celery/test_scheduled_tasks.py

@@ -6,6 +6,7 @@ from collections import namedtuple
 from freezegun import freeze_time
 from mock import mock
 
+from app import db
 from app.celery import scheduled_tasks
 from app.celery.scheduled_tasks import (
     check_job_status,
@@ -19,7 +20,7 @@ from app.celery.scheduled_tasks import (
     check_for_missing_rows_in_completed_jobs,
     check_for_services_with_high_failure_rates_or_sending_to_tv_numbers,
     switch_current_sms_provider_on_slow_delivery,
-)
+    purge_high_volume_notifications)
 from app.config import QueueNames, TaskNames, Config
 from app.dao.jobs_dao import dao_get_job_by_id
 from app.dao.notifications_dao import dao_get_scheduled_notifications
@@ -30,7 +31,7 @@ from app.models import (
     JOB_STATUS_FINISHED,
     NOTIFICATION_DELIVERED,
     NOTIFICATION_PENDING_VIRUS_CHECK,
-)
+    HighVolumeService, Notification, FactBilling, FactNotificationStatus)
 from app.v2.errors import JobIncompleteError
 from tests.app import load_example_csv
 
@@ -314,7 +315,7 @@ def test_replay_created_notifications(notify_db_session, sample_service, mocker)
 
 
 def test_replay_created_notifications_create_letters_pdf_tasks_for_letters_not_ready_to_send(
-        sample_letter_template, mocker
+    sample_letter_template, mocker
 ):
     mock_task = mocker.patch('app.celery.scheduled_tasks.create_letters_pdf.apply_async')
     create_notification(template=sample_letter_template, billable_units=0,
@@ -556,3 +557,56 @@ def test_check_for_services_with_high_failure_rates_or_sending_to_tv_numbers(
         subject="[test] High failure rates for sms spotted for services",
         ticket_type='incident'
     )
+
+
+@freeze_time('2020-03-19 13:30')
+def test_purge_high_volume_notifications(sample_email_template, notify_db_session):
+    # should be deleted
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(days=4), status='delivered')
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(hours=2), status='permanent-failure')
+
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(hours=1, minutes=1), status='temporary-failure')
+    # should NOT be deleted
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(minutes=59), status='temporary-failure')
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(hours=1), status='temporary-failure')
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(hours=1), status='delivered')
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(hours=1), status='created')
+
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(days=1), status='sending')
+
+    create_notification(template=sample_email_template,
+                        created_at=datetime.utcnow() - timedelta(days=1), status='technical-failure')
+
+    high_volume_service = HighVolumeService(service_id=sample_email_template.service_id)
+    db.session.add(high_volume_service)
+    db.session.commit()
+    purge_high_volume_notifications()
+
+    notifications = Notification.query.all()
+    assert len(notifications) == 6
+
+    ft_billing = FactBilling.query.all()
+    assert len(ft_billing) == 1
+    assert ft_billing[0].service_id == sample_email_template.service_id
+    assert ft_billing[0].notifications_sent == 3
+    assert str(ft_billing[0].bst_date) == '2020-03-19'
+
+    ft_status = FactNotificationStatus.query.order_by(FactNotificationStatus.notification_status).all()
+    assert len(ft_status) == 3
+    assert str(ft_status[0].bst_date) == '2020-03-19'
+    assert ft_status[0].notification_status == 'delivered'
+    assert ft_status[0].notification_count == 1
+    assert str(ft_status[1].bst_date) == '2020-03-19'
+    assert ft_status[1].notification_status == 'permanent-failure'
+    assert ft_status[1].notification_count == 1
+    assert str(ft_status[2].bst_date) == '2020-03-19'
+    assert ft_status[2].notification_status == 'temporary-failure'
+    assert ft_status[2].notification_count == 1