Add aggregated Postgres writer for status change topic

DEVELOPER.md

@@ -164,6 +164,12 @@ EventGate uses a **direct invocation approach** for integration testing:
 - Docker running (Docker Desktop on macOS/Windows, or Docker Engine on Linux)
 - Python 3.13 with dependencies installed
 
+When using colima on macOS, set the following environment variables before running tests:
+```shell
+export DOCKER_HOST=unix://$HOME/.colima/default/docker.sock
+export TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=/var/run/docker.sock
+```
+
 ### Run Integration Tests
 
 Containers start and stop automatically:

adr/001-status-change-db/001-status-change-db.md

@@ -0,0 +1,185 @@
+# ADR 001. Event Bus -- Status change database
+
+## Decision
+Aggregate events and insert into newly created database table.
+
+## Background
+The status change topic receives job status events, i.e. it is an event log.
+For monitoring, a database table with the current status is needed, i.e. events need to be aggregated into the latest state
+So, the table stores the latest merged state per `job_id`, not the full event log.
+
+This ADR shall describe the database schema as well as the aggregation logic and some of the queries that the database should support.
+
+## Database schema
+
+```sql
+CREATE TABLE job (
+    job_id               UUID PRIMARY KEY,
+    job_group_id         UUID,
+    parent_job_id        UUID,
+    initial_job_id       UUID,
+
+    -- Identity / business attributes
+    job_ref              TEXT,
+    job_name             TEXT,
+    definition_id        TEXT,
+    definition_version   TEXT,
+    tenant_id            TEXT,
+    country              TEXT,
+    source_app           TEXT,
+    source_app_version   TEXT,
+    environment          TEXT,
+
+    -- Execution context
+    platform             TEXT,
+    platform_metadata    JSONB,
+    input_arguments      JSONB,
+    additional_context   JSONB,
+    attempt_number       INTEGER NOT NULL CHECK (attempt_number > 0),
+
+    -- Current lifecycle status
+    status_type          TEXT CHECK (status_type IN ('WAITING', 'RUNNING', 'SUCCEEDED', 'FAILED', 'KILLED')),
+    status_subtype       TEXT,
+    status_detail        TEXT,
+
+    -- Lifecycle timestamps derived from events
+    created_at           TIMESTAMPTZ,
+    started_at           TIMESTAMPTZ,
+    finished_at          TIMESTAMPTZ,
+    last_updated_at      TIMESTAMPTZ NOT NULL,
+
+   CONSTRAINT fk_job_parent
+       FOREIGN KEY (parent_job_id)
+           REFERENCES job (job_id)
+           ON DELETE RESTRICT
+
+   CONSTRAINT fk_job_initial
+       FOREIGN KEY (initial_job_id)
+           REFERENCES job (job_id)
+           ON DELETE RESTRICT
+);
+```
+
+## Event insertion and aggregation
+
+### Assumptions
+- Most events are received in order per `job_id`, but some events may be received out-of-order (during failures scenarios). The order of events is determined by the field `timestamp_event`. If an event is out-of-order, it means that an event with a higher `timestamp_event` has already been processed for the database insert / update.
+- Duplicate events can occur and must be handled idempotently, i.e. duplicate events must not lead to two rows in the database table.
+- The `job` table stores the latest snapshot of the job's status, not historical versions.
+
+### Field merge strategy
+The field merge strategy is defined in the SQL upsert query. Therefore the field names of the database table rather than the Kafka event are used below.
+
+The field merge strategy takes into account out-of-order updates and the idempotency of duplicates. The true event order is determined by the field `last_updated_at` (same as `timestamp_event` in Kafka record). There are three merge strategies:
+- Take latest non-null: If either the incoming or the current field value is non-null, then the non-null value is accepted. If both incoming and current value are non-null, then the newer value (determined by `last_updated_at`) is accepted. This means that a value that has been set, will never be set to null again. This merge strategy applies to most fields.
+
+- Take latest: The newer value is accepted, even if it is null. This applies to the `status_subtype` and `status_detail` fields, as they are coupled to the `status_type` field, which is mandatory for all events. If an initial status like `RUNNING` has a `status_detail` (for whatever reason), then it should not be kept when the status changes to `FINISHED`, but instead set to null. However, usually only terminal statuses should have `status_detail` and `status_subtype`
+
+- Take latest non-default: For fields with default values, default values are treated like null values in terms of the merge. This means that non-default values are preferred, even if they are not the latest value. Once a value different to the default has been set, it is not possible to set it back to the default value. This applies to the field `attempt_number`.
+
+- Cumulative merge: For `additional_context`, the merge strategy is a cumulative merge, i.e. fields from both the incoming and current record are retained. Note that this is not the case for `input_arguments` and `platform_metadata`
+
+### Timestamp conversion
+Timestamps in Kafka events are epoch milliseconds, which should be converted to `TIMESTAMPTZ` in Postgres. Using a dedicated timestamp type greatly simplifies querying and reading from the table, especially in BI-tools.
+
+## Sample query patterns
+**Get jobs by job name**
+```sql
+SELECT *
+FROM job j
+WHERE job_name = :my_job_name
+ORDER BY last_updated_at;
+```
+
+**Get all jobs grouped by job group id (including retries)**
+```sql
+SELECT
+    j.*,
+    MIN(started_at) OVER (PARTITION BY job_group_id) AS group_started_at
+FROM job j
+ORDER BY
+    group_started_at,
+    job_group_id,
+    started_at;
+```
+
+**Get job hierarchy roots, including only latest attempt**
+```sql
+WITH latest_root_attempts AS (
+    SELECT *
+    FROM (
+             SELECT
+                 j.*,
+                 ROW_NUMBER() OVER (
+                     PARTITION BY COALESCE(initial_job_id, job_id)
+                     ORDER BY attempt_number DESC, started_at DESC
+                     ) AS rn
+             FROM job j
+             WHERE job_name = 'Aqueduct Ingestion'
+         ) t
+    WHERE rn = 1
+)
+SELECT
+    *
+FROM latest_root_attempts
+ORDER BY
+    job_group_id,
+    started_at;
+```    
+
+**Get all jobs in a job hierarchy given a specific job group id, including only latest attempts**
+```sql
+SELECT *
+FROM (
+         SELECT
+             j.*,
+             ROW_NUMBER() OVER (
+                 PARTITION BY COALESCE(initial_job_id, job_id)
+                 ORDER BY attempt_number DESC, started_at DESC
+                 ) AS rn
+         FROM job j
+         WHERE job_group_id = '11111111-1111-4111-8111-111111111111'::uuid
+     ) t
+WHERE rn = 1;
+```
+
+**Get all jobs in all job hierarchies, including only latest attempts**
+```sql
+WITH RECURSIVE latest_attempts AS (
+    SELECT *
+    FROM (
+             SELECT
+                 j.*,
+                 ROW_NUMBER() OVER (
+                     PARTITION BY COALESCE(initial_job_id, job_id)
+                     ORDER BY attempt_number DESC, started_at DESC
+                     ) AS rn
+             FROM job j
+         ) t
+    WHERE rn = 1
+),
+latest_root AS (
+    SELECT job_id
+    FROM latest_attempts
+    WHERE parent_job_id IS NULL
+    ORDER BY attempt_number DESC, started_at DESC
+    LIMIT 1
+),
+tree AS (
+    -- start from latest root
+    SELECT la.*
+    FROM latest_attempts la
+            JOIN latest_root r ON la.job_id = r.job_id
+
+    UNION ALL
+
+    -- traverse hierarchy
+    SELECT child.*
+    FROM latest_attempts child
+            JOIN tree parent
+                    ON child.parent_job_id = parent.job_id
+)
+SELECT *
+FROM tree
+ORDER BY started_at;
+```

conf/access.json

@@ -10,5 +10,5 @@
     },
     "public.cps.za.dlchange": ["FooUser", "BarUser"],
     "public.cps.za.test": ["TestUser"],
-    "public.cps.za.status_change": ["TestUser"]
+    "public.cps.za.status_change": ["TestUser", "IntegrationTestUser"]
 }

src/utils/constants.py

@@ -39,5 +39,5 @@
 TOPIC_TEST = "public.cps.za.test"
 TOPIC_STATUS_CHANGE = "public.cps.za.status_change"
 
-POSTGRES_WRITE_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS, TOPIC_DLCHANGE, TOPIC_TEST})
+POSTGRES_WRITE_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS, TOPIC_DLCHANGE, TOPIC_TEST, TOPIC_STATUS_CHANGE})
 SUPPORTED_STATS_TOPICS: frozenset[str] = frozenset({TOPIC_RUNS})

src/writers/sql/inserts.sql

@@ -35,3 +35,141 @@ INSERT INTO public_cps_za_test
 VALUES
     (:event_id, :tenant_id, :source_app, :environment,
      :timestamp_event, :additional_info);
+
+-- name: upsert_status_change(job_id, job_group_id, parent_job_id, initial_job_id, job_ref, job_name, definition_id, definition_version, tenant_id, country, source_app, source_app_version, environment, platform, platform_metadata, input_arguments, additional_context, attempt_number, status_type, status_subtype, status_detail, created_at, started_at, finished_at, last_updated_at)!
+-- Upsert a status_change event into the aggregated job table using merge logic per ADR 001.
+INSERT INTO public_cps_za_status_change_aggregated_job AS t (
+    job_id, job_group_id, parent_job_id, initial_job_id,
+    job_ref, job_name, definition_id, definition_version,
+    tenant_id, country, source_app, source_app_version, environment,
+    platform, platform_metadata, input_arguments, additional_context,
+    attempt_number, status_type, status_subtype, status_detail,
+    created_at, started_at, finished_at, last_updated_at
+)
+VALUES (
+    :job_id, :job_group_id, :parent_job_id, :initial_job_id,
+    :job_ref, :job_name, :definition_id, :definition_version,
+    :tenant_id, :country, :source_app, :source_app_version, :environment,
+    :platform, :platform_metadata, :input_arguments, :additional_context,
+    :attempt_number, :status_type, :status_subtype, :status_detail,
+    :created_at, :started_at, :finished_at, :last_updated_at
+)
+ON CONFLICT (job_id) DO UPDATE SET
+    job_group_id       = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.job_group_id, t.job_group_id)
+                           ELSE COALESCE(t.job_group_id, EXCLUDED.job_group_id)
+                         END,
+    parent_job_id      = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.parent_job_id, t.parent_job_id)
+                           ELSE COALESCE(t.parent_job_id, EXCLUDED.parent_job_id)
+                         END,
+    initial_job_id     = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.initial_job_id, t.initial_job_id)
+                           ELSE COALESCE(t.initial_job_id, EXCLUDED.initial_job_id)
+                         END,
+    job_ref            = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.job_ref, t.job_ref)
+                           ELSE COALESCE(t.job_ref, EXCLUDED.job_ref)
+                         END,
+    job_name           = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.job_name, t.job_name)
+                           ELSE COALESCE(t.job_name, EXCLUDED.job_name)
+                         END,
+    definition_id      = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.definition_id, t.definition_id)
+                           ELSE COALESCE(t.definition_id, EXCLUDED.definition_id)
+                         END,
+    definition_version = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.definition_version, t.definition_version)
+                           ELSE COALESCE(t.definition_version, EXCLUDED.definition_version)
+                         END,
+    tenant_id          = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.tenant_id, t.tenant_id)
+                           ELSE COALESCE(t.tenant_id, EXCLUDED.tenant_id)
+                         END,
+    country            = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.country, t.country)
+                           ELSE COALESCE(t.country, EXCLUDED.country)
+                         END,
+    source_app         = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.source_app, t.source_app)
+                           ELSE COALESCE(t.source_app, EXCLUDED.source_app)
+                         END,
+    source_app_version = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.source_app_version, t.source_app_version)
+                           ELSE COALESCE(t.source_app_version, EXCLUDED.source_app_version)
+                         END,
+    environment        = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.environment, t.environment)
+                           ELSE COALESCE(t.environment, EXCLUDED.environment)
+                         END,
+    platform           = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.platform, t.platform)
+                           ELSE COALESCE(t.platform, EXCLUDED.platform)
+                         END,
+    platform_metadata  = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.platform_metadata, t.platform_metadata)
+                           ELSE COALESCE(t.platform_metadata, EXCLUDED.platform_metadata)
+                         END,
+    input_arguments    = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.input_arguments, t.input_arguments)
+                           ELSE COALESCE(t.input_arguments, EXCLUDED.input_arguments)
+                         END,
+    additional_context = CASE
+                           WHEN EXCLUDED.additional_context IS NULL THEN t.additional_context
+                           WHEN t.additional_context IS NULL        THEN EXCLUDED.additional_context
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN t.additional_context || EXCLUDED.additional_context
+                           ELSE EXCLUDED.additional_context || t.additional_context
+                         END,
+    attempt_number     = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(NULLIF(EXCLUDED.attempt_number, 1), t.attempt_number, 1)
+                           ELSE COALESCE(NULLIF(t.attempt_number, 1), EXCLUDED.attempt_number, 1)
+                         END,
+    status_type        = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN EXCLUDED.status_type
+                           ELSE t.status_type
+                         END,
+    status_subtype     = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN EXCLUDED.status_subtype
+                           ELSE t.status_subtype
+                         END,
+    status_detail      = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN EXCLUDED.status_detail
+                           ELSE t.status_detail
+                         END,
+    created_at         = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.created_at, t.created_at)
+                           ELSE COALESCE(t.created_at, EXCLUDED.created_at)
+                         END,
+    started_at         = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.started_at, t.started_at)
+                           ELSE COALESCE(t.started_at, EXCLUDED.started_at)
+                         END,
+    finished_at        = CASE
+                           WHEN EXCLUDED.last_updated_at >= t.last_updated_at
+                           THEN COALESCE(EXCLUDED.finished_at, t.finished_at)
+                           ELSE t.finished_at
+                         END,
+    last_updated_at    = GREATEST(EXCLUDED.last_updated_at, t.last_updated_at);