track message costs

app/celery/scheduled_tasks.py

@@ -266,7 +266,7 @@ def process_delivery_receipts(self):
 
         cloudwatch = AwsCloudwatchClient()
         cloudwatch.init_app(current_app)
-        start_time = aware_utcnow() - timedelta(minutes=3)
+        start_time = aware_utcnow() - timedelta(minutes=30)
         end_time = aware_utcnow()
         delivered_receipts, failed_receipts = cloudwatch.check_delivery_receipts(
             start_time, end_time

app/clients/cloudwatch/aws_cloudwatch.py

@@ -7,6 +7,7 @@ from flask import current_app
 
 from app.clients import AWS_CLIENT_CONFIG, Client
 from app.cloudfoundry_config import cloud_config
+from app.utils import hilite
 
 
 class AwsCloudwatchClient(Client):
@@ -107,6 +108,13 @@ class AwsCloudwatchClient(Client):
         provider_response = self._aws_value_or_default(
             event, "delivery", "providerResponse"
         )
+        message_cost = self._aws_value_or_default(event, "delivery", "priceInUSD")
+        if message_cost is None or message_cost == "":
+            message_cost = 0.0
+        else:
+            message_cost = float(message_cost)
+        current_app.logger.info(hilite(f"EVENT {event} message_cost = {message_cost}"))
+
         my_timestamp = self._aws_value_or_default(event, "notification", "timestamp")
         return {
             "notification.messageId": event["notification"]["messageId"],
@@ -114,6 +122,7 @@ class AwsCloudwatchClient(Client):
             "delivery.phoneCarrier": phone_carrier,
             "delivery.providerResponse": provider_response,
             "@timestamp": my_timestamp,
+            "delivery.priceInUSD": message_cost,
         }
 
     # Here is an example of how to get the events with log insights

app/dao/notifications_dao.py

@@ -507,7 +507,7 @@ def insert_notification_history_delete_notifications(
          SELECT id, job_id, job_row_number, service_id, template_id, template_version, api_key_id,
              key_type, notification_type, created_at, sent_at, sent_by, updated_at, reference, billable_units,
              client_reference, international, phone_prefix, rate_multiplier, notification_status,
-              created_by_id, document_download_count
+              created_by_id, document_download_count, message_cost
           FROM notifications
         WHERE service_id = :service_id
           AND notification_type = :notification_type
@@ -842,7 +842,6 @@ def dao_update_delivery_receipts(receipts, delivered):
         new_receipts.append(r)
 
     receipts = new_receipts
-
     id_to_carrier = {
         r["notification.messageId"]: r["delivery.phoneCarrier"] for r in receipts
     }
@@ -851,9 +850,13 @@ def dao_update_delivery_receipts(receipts, delivered):
     }
     id_to_timestamp = {r["notification.messageId"]: r["@timestamp"] for r in receipts}
 
+    id_to_message_cost = {
+        r["notification.messageId"]: r["delivery.priceInUSD"] for r in receipts
+    }
     status_to_update_with = NotificationStatus.DELIVERED
     if not delivered:
         status_to_update_with = NotificationStatus.FAILED
+
     stmt = (
         update(Notification)
         .where(Notification.message_id.in_(id_to_carrier.keys()))
@@ -877,6 +880,12 @@ def dao_update_delivery_receipts(receipts, delivered):
                     for key, value in id_to_provider_response.items()
                 ]
             ),
+            message_cost=case(
+                *[
+                    (Notification.message_id == key, value)
+                    for key, value in id_to_message_cost.items()
+                ]
+            ),
         )
     )
     db.session.execute(stmt)
@@ -908,7 +917,7 @@ def dao_close_out_delivery_receipts():
 
 
 def dao_batch_insert_notifications(batch):
-
+    current_app.logger.info(f"ENTER DAO_BATCH_INSERT with batch {batch}")
     db.session.bulk_save_objects(batch)
     db.session.commit()
     current_app.logger.info(f"Batch inserted notifications: {len(batch)}")

app/dao/services_dao.py

@@ -534,7 +534,9 @@ def dao_fetch_stats_for_service_from_hours(service_id, start_date, end_date):
     # Update to group by HOUR instead of DAY
     total_substmt = (
         select(
-            func.date_trunc("hour", NotificationAllTimeView.created_at).label("hour"),  # UPDATED
+            func.date_trunc("hour", NotificationAllTimeView.created_at).label(
+                "hour"
+            ),  # UPDATED
             Job.notification_count.label("notification_count"),
         )
         .join(Job, NotificationAllTimeView.job_id == Job.id)
@@ -556,11 +558,14 @@ def dao_fetch_stats_for_service_from_hours(service_id, start_date, end_date):
     total_stmt = select(
         total_substmt.c.hour,  # UPDATED
         func.sum(total_substmt.c.notification_count).label("total_notifications"),
-    ).group_by(total_substmt.c.hour)  # UPDATED
+    ).group_by(
+        total_substmt.c.hour
+    )  # UPDATED
 
     # Ensure we're using hourly timestamps in the response
     total_notifications = {
-        row.hour: row.total_notifications for row in db.session.execute(total_stmt).all()
+        row.hour: row.total_notifications
+        for row in db.session.execute(total_stmt).all()
     }
 
     # Update the second query to also use "hour"
@@ -568,7 +573,9 @@ def dao_fetch_stats_for_service_from_hours(service_id, start_date, end_date):
         select(
             NotificationAllTimeView.notification_type,
             NotificationAllTimeView.status,
-            func.date_trunc("hour", NotificationAllTimeView.created_at).label("hour"),  # UPDATED
+            func.date_trunc("hour", NotificationAllTimeView.created_at).label(
+                "hour"
+            ),  # UPDATED
             func.count(NotificationAllTimeView.id).label("count"),
         )
         .where(
@@ -895,7 +902,9 @@ def get_specific_days_stats(
     return stats
 
 
-def get_specific_hours_stats(data, start_date, hours=None, end_date=None, total_notifications=None):
+def get_specific_hours_stats(
+    data, start_date, hours=None, end_date=None, total_notifications=None
+):
     if hours is not None and end_date is not None:
         raise ValueError("Only set hours OR set end_date, not both.")
     elif hours is not None:
@@ -919,10 +928,10 @@ def get_specific_hours_stats(data, start_date, hours=None, end_date=None, total_
     # Format statistics, returning only hours with results
     stats = {
         hour.strftime("%Y-%m-%dT%H:00:00Z"): statistics.format_statistics(
-            rows,
-            total_notifications.get(hour, 0) if total_notifications else None
+            rows, total_notifications.get(hour, 0) if total_notifications else None
         )
-        for hour, rows in grouped_data.items() if rows
+        for hour, rows in grouped_data.items()
+        if rows
     }
 
     return stats

app/models.py

@@ -1508,6 +1508,7 @@ class Notification(db.Model):
     created_at = db.Column(db.DateTime, index=True, unique=False, nullable=False)
     sent_at = db.Column(db.DateTime, index=False, unique=False, nullable=True)
     sent_by = db.Column(db.String, nullable=True)
+    message_cost = db.Column(db.Float, nullable=True, default=0.0)
     updated_at = db.Column(
         db.DateTime,
         index=False,
@@ -1813,6 +1814,7 @@ class NotificationHistory(db.Model, HistoryModel):
     created_at = db.Column(db.DateTime, unique=False, nullable=False)
     sent_at = db.Column(db.DateTime, index=False, unique=False, nullable=True)
     sent_by = db.Column(db.String, nullable=True)
+    message_cost = db.Column(db.Float, nullable=True, default=0.0)
     updated_at = db.Column(
         db.DateTime,
         index=False,