diff --git a/app/celery/reporting_tasks.py b/app/celery/reporting_tasks.py index 7ce1ff57a..ea5e50f2b 100644 --- a/app/celery/reporting_tasks.py +++ b/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() diff --git a/app/celery/scheduled_tasks.py b/app/celery/scheduled_tasks.py index e0c419534..6c56f6eb1 100644 --- a/app/celery/scheduled_tasks.py +++ b/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() diff --git a/app/models.py b/app/models.py index 802567046..d1c3d5c1d 100644 --- a/app/models.py +++ b/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) diff --git a/migrations/versions/0319_high_volume_service.py b/migrations/versions/0319_high_volume_service.py new file mode 100644 index 000000000..4589d32b6 --- /dev/null +++ b/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') diff --git a/tests/app/celery/test_scheduled_tasks.py b/tests/app/celery/test_scheduled_tasks.py index bb43251c6..e266b13a6 100644 --- a/tests/app/celery/test_scheduled_tasks.py +++ b/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