PLAT-2789: Support Bulk Event Ingestion and Asynchronous Processing

Overview

This PR introduces support for bulk event ingestion to the metrics-api, significantly improving throughput for high-volume data providers. It also implements a robust fallback mechanism to ensure data integrity during asynchronous processing failures.

Key Changes

1. Optimized Event Ingestion (eventsctrl.py)

• Bulk Support: The POST /events endpoint now accepts a list of event objects.
• Hybrid Processing:
  • Single Events: Remain synchronous to support existing service expectations and provide immediate feedback.
  • Bulk Events: Processed asynchronously via a new Celery task (process-events-batch), allowing the API to return quickly while offloading heavy DB operations.
• Validation: Improved validation logic that pre-processes events before queuing, ensuring only well-formed data enters the task queue.

2. Asynchronous Ingestion Task (tasks/events_ingestion.py)

• Implemented process_events_batch which utilizes a new Event.bulk_save method.
• Added a retry policy (3 retries with a 60-second countdown) to handle transient database issues.

3. Fallback Storage Mechanism

• Added fallback_event_storage table to the schema.
• Implemented FallbackEventStorage model to capture payloads that fail after all retries are exhausted. This prevents data loss and allows for manual inspection and recovery.

4. Database Optimizations (models/event.py)

• Added bulk_save using psycopg2.extras.execute_values for high-performance batch inserts.
• Implemented Python-side deduplication within batches to prioritize the latest data for conflicting event_uri records.
• Enhanced Event.init to be robust against both datetime and string timestamp formats.

Testing Performed

• Unit Tests: New tests in src/tasks/tests/test_events_ingestion.py cover successful processing, retries, and the fallback trigger.
• Integration Tests: Updated src/tests/test_events.py to verify:
  • Single events are saved synchronously and immediately available.
  • Bulk events correctly trigger the asynchronous pipeline.
• Environment: All tests verified against a live Postgres/Redis environment using the standard test suite variables.

How to Review

1. Check src/eventsctrl.py to verify the logic branching between single and bulk POSTs.
2. Review the bulk_save implementation in src/models/event.py for correct conflict handling.
3. Verify the Celery task configuration in src/tasks/config.py and src/tasks/celery_app.py.

---

Card: https://ubiquitypress.atlassian.net/browse/PLAT-2789

---

MR Feedback code changes:

1. Remove conflict between task-level and controller-level logic for bulk saveing events.
2. Propagated useful error messages to add to quarantined rows
3. Change API response for bulk-save path to reflect a 202 accept, and return the invalid events, along with an indication that these were rejected due to invalid parameters (left more detailed error feedback as future work).
4. Delegated bulk-event ingestion to a dedicated celery worker.
5. Added a max-batch-size guard at the controller, as well as a lightweight, fail-fast content lenth check.
6. Removed web.ctx.status = '200 OK' from non-production code, and ignored this check in tests.