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/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