Aidbox Topic-Based Subscriptions

Overview

This feature enables dynamic subscriptions to changes in FHIR resources, allowing users/systems to receive notifications through multiple channels.

For an application example, refer to Aidbox Subscriptions & Kafka AidboxTopicDestination

Key Components

• AidboxSubscriptionTopic is a custom Aidbox resource modeled after the FHIR R6 SubscriptionTopic resource. The resource allows defining a set of events that clients can subscribe to, such as changes in specific resources.
• AidboxTopicDestination is a custom Aidbox resource that defines where and how the notifications triggered by an AidboxSubscriptionTopic should be sent. This resource offers flexibility in specifying various types of destinations. And is considered as a system configuration resource.
• AidboxSubscriptionStatus is a custom Aidbox resource which describes the notifications: what messages stored in the bundle, source and destination.

Currently supported channels

Webhook

Send FHIR resource events to HTTP webhooks with retry logic and batch support.

GCP Pub/Sub

Stream FHIR resource events to Google Cloud Pub/Sub with guaranteed delivery.

Kafka

Stream FHIR resource events to Apache Kafka with best-effort or at-least-once delivery.

ClickHouse

Export FHIR resources to ClickHouse with SQL-on-FHIR ViewDefinitions for real-time reporting.

BigQuery

Export FHIR resources to Google BigQuery with SQL-on-FHIR ViewDefinitions.

NATS

Integrate Aidbox topic-based subscriptions with NATS and NATS JetStream.

RabbitMQ

Connect Aidbox subscriptions to RabbitMQ over AMQP for real-time FHIR events.

ActiveMQ

Integrate Aidbox subscriptions with ActiveMQ using AMQP 1.0 for event streaming.

AWS EventBridge

Route FHIR resource events to AWS EventBridge for serverless processing.

AWS SNS

Stream FHIR resource events to AWS SNS for pub/sub messaging and fan-out notifications.

Data Lakehouse (Databricks)

Export FHIR resources to Databricks Unity Catalog managed Delta tables.

AidboxSubscriptionTopic

The AidboxSubscriptionTopic resource describes the data sources for subscriptions. It allows clients to subscribe to events in Aidbox and filter them using user-defined triggers, which are specified by the trigger element. Supported properties:

* required.

Create AidboxSubscriptionTopic resource

POST /fhir/AidboxSubscriptionTopic
content-type: application/json
accept: application/json

{
  "resourceType": "AidboxSubscriptionTopic",
  "url": "http://example.org/FHIR/R5/SubscriptionTopic/QuestionnaireResponse-topic",
  "status": "active",
  "trigger": [
    {
      "resource": "QuestionnaireResponse",
      "fhirPathCriteria": "status = 'completed' or status = 'amended'"
    }
  ]
}

Built-in topics

Aidbox ships topics that exist without a stored AidboxSubscriptionTopic resource. Reference them by URL in AidboxTopicDestination.topic. Aidbox rejects a stored topic that reuses a built-in URL.

AidboxTopicDestination

The AidboxTopicDestination resource is used to define channel configurations for processing subscription data.

Create a TopicDestination

To start processing subscription data, create a AidboxTopicDestination resource with a reference to the AidboxSubscriptionTopic. Examples of AidboxTopicDestination resources can be found in sub-sections.

Stop subscription data processing

To stop processing subscription data, delete the AidboxTopicDestination resource.

Updating a TopicDestination

Each AidboxTopicDestination runs a live sender: a Kafka producer, a webhook delivery worker, a GCP Pub/Sub session. Aidbox builds the sender when you create the destination and cannot rebuild it from an edited resource. For this reason PUT and PATCH on a destination return 405 Method Not Allowed, in direct requests and inside transaction and batch bundles, including init bundles. (AidboxSubscriptionTopic carries no runtime state and supports PUT as usual.)

Some destination kinds exempt specific parameters from this rule and accept a PUT that changes them alone. Each kind's page documents its exceptions, for example the webhook endpoint.

To change anything else, recreate the destination: DELETE the existing resource and POST a new one with the same id. Undelivered events are dropped together with the old destination, so drain the queue first if they matter: watch messagesQueued in $status reach zero before the DELETE.

In an init bundle, use a batch bundle for the recreate pattern, so the DELETE succeeds when the resource does not yet exist (returns 204):

{
  "resourceType": "Bundle",
  "type": "batch",
  "entry": [
    {
      "request": { "method": "DELETE", "url": "AidboxTopicDestination/<id>" }
    },
    {
      "resource": { "resourceType": "AidboxTopicDestination", "id": "<id>", "...": "..." },
      "request": { "method": "POST", "url": "AidboxTopicDestination" }
    }
  ]
}

Status ($status) and HA deployments

Each kind-specific tutorial (e.g. webhook, Kafka) describes the $status operation. One detail applies to every destination kind:

• messagesDelivered, messageBatchesDelivered, messagesDeliveryAttempts, messageBatchesDeliveryAttempts, messagesInProcess, lastErrorDetail, and startTimestamp are per-replica and held in memory. They reset on Aidbox restart and are not summed across replicas.
• messagesQueued is computed from the database and is consistent across replicas.

In a highly available deployment, calling $status through a load balancer returns whichever replica answered. To get cluster-wide totals, query each pod directly and sum the in-memory counters — see Counter semantics for the full breakdown.

AidboxTopicDestination Profile

Ensure that the resource metadata contains the kind-specific AidboxTopicDestination profile.

Elements

* required.

Filter events with filterBy

A destination receives every event of its topic unless you add filterBy. The topic declares the parameters you can filter by in trigger.canFilterBy. Each parameter has a FHIRPath expression that extracts values from the resource of the event.

Define a filter parameter on the topic:

POST /fhir/AidboxSubscriptionTopic
content-type: application/json
accept: application/json

{
  "resourceType": "AidboxSubscriptionTopic",
  "url": "http://example.org/SubscriptionTopic/patient-topic",
  "status": "active",
  "trigger": [
    {
      "resource": "Patient",
      "canFilterBy": [
        {
          "filterParameter": "gender",
          "filterDefinitionFhirPathExpression": "gender",
          "comparator": ["eq"],
          "description": "Patient gender"
        }
      ]
    }
  ]
}

Filter by it in a destination:

POST /fhir/AidboxTopicDestination
content-type: application/json
accept: application/json

{
  "resourceType": "AidboxTopicDestination",
  "id": "female-patients",
  "meta": {
    "profile": [
      "http://health-samurai.io/fhir/core/StructureDefinition/aidboxtopicdestination-webhookAtLeastOnceProfile"
    ]
  },
  "kind": "webhook-at-least-once",
  "topic": "http://example.org/SubscriptionTopic/patient-topic",
  "filterBy": [
    {
      "filterParameter": "gender",
      "comparator": "eq",
      "value": "female"
    }
  ],
  "parameter": [
    {
      "name": "endpoint",
      "valueUrl": "https://example.com/webhook"
    }
  ]
}

The destination receives events of female patients only. With "comparator": "ne", it receives events of every patient except female ones, including patients without gender.

Aidbox matches filters by these rules:

• Aidbox compares value with the values that filterDefinitionFhirPathExpression returns. References compare by their reference string, booleans and numbers by their string form. Other complex values, such as a Coding, match nothing, so point the expression at a primitive, for example code.coding.code.
• comparator accepts eq and ne. Without comparator, Aidbox uses eq. Aidbox rejects other comparators with 422.
• eq matches when any value of the event equals value. ne matches when no value of the event equals value.
• Comma-separated values in one filter match eq when any of them matches, and ne when none of them matches: "value": "male,female".
• Several filters match when all of them match.
• resource limits a filter to the trigger of that resource, for topics with several triggers.
• An event with no value for the filter parameter does not match eq and matches ne. A filter on a parameter that the topic does not define matches no event with eq and every event with ne.
• You cannot change filterBy after creation: PUT returns 405. Delete the destination and create it again.

filterBy works with every destination kind. For the fhir-native-topic-based-subscription kind, Aidbox applies the destination filters first and the filterBy of each FHIR Subscription after them. Built-in topics define their own filter parameters, see Filter audit events.

Organization-based hierarchical filtering

Both AidboxSubscriptionTopic and AidboxTopicDestination support organization-based hierarchical access control.

How it works

The filtering mechanism uses organization extensions and works as consecutive filters:

1. Use organization extension:

   In organizational-based access control, both events (resources) and subscription resources (AidboxSubscriptionTopic and AidboxTopicDestination) are automatically marked with organization extensions when created through organization-specific APIs (/Organization/<org-id>/fhir/...).

2. Consecutive filtering: The filters work in sequence - first the topic filter is applied, then the destination filter.

3. Resources without organization extension: If topic/destination resources are created without organization extension, they completely ignore organization information and work as before. All events will be caught by the topic if triggered (regardless of whether the event has organization extension or not).

4. Resources with organization extension:

   • Events without organization extension: Will always be skipped
   • Events with organization extension:
     • ✅ Caught: If the event organization is under or equal to the topic/destination organization in the hierarchy
     • ❌ Skipped: If the event organization is above or unrelated to the topic/destination organization

Examples

Example 1: Organization hierarchy with event filtering

Consider an organization structure where Organization B is part of Organization A. The diagram below shows how events flow through the subscription system with hierarchical filtering:

Filtering behavior:

• ✅ Event from Org A: Processed by the topic (same organization) but filtered out at the aidbox topic destination level
• ✅ Event from Org B: Processed by the topic (child organization) and forwarded to destination at the aidbox topic destination level
• ❌ Event from unrelated organization: Would be filtered out at the subscription topic level

Example 2: Filtering at the destination level only

Consider an organization structure which two organizations Organization A and Organization B. The diagram below shows how events flow through the subscription system when filtering is applied at the destination level only:

Filtering behavior:

• ✅ Event from Org A: Processed by the topic (no filtering) and forwarded to aidbox topic destination
• ✅ Event from Org B: Processed by the topic (no filtering) but filtered out at the aidbox topic destination level
• ❌ Event from unrelated organization: Would be filtered out at the aidbox topic destination level

Notification Shape

Notification is a FHIR Bundle resource with history type, containing relevant resources in its entries. The first entry is a AidboxSubscriptionStatus resource, which describes the payload.
The other entries depend on AidboxTopicDestination.content element. This element has the binding to the FHIR subscription-payload-content value set: full-resource | id-only | empty

{
  "resourceType":"Bundle",
  "type":"history",
  "timestamp":"2024-10-03T10:07:55Z",
  "entry":[
    {
      "resource":{
        "resourceType":"AidboxSubscriptionStatus",
        "status":"active",
        "type":"event-notification",
        "notificationEvent":[
          {
            "eventNumber":1,
            "focus":{
              "reference":"QuestionnaireResponse/458e771c-0ff1-4858-ac07-93b7a10c8e3b"
            }
          }
        ],
        "topic":"http://example.org/FHIR/R5/SubscriptionTopic/QuestionnaireResponse-topic",
        "topic-destination":{
          "reference":"AidboxTopicDestination/kafka-destination"
        }
      }
    },
    {
      "resource":{
        "id":"458e771c-0ff1-4858-ac07-93b7a10c8e3b",
        "item":[
          {
            "text":"Leishmania sp 14kD IgG Ser Ql IB",
            "answer":[
              {
                "valueString":"123"
              }
            ],
            "linkId":"128852"
          },
          {
            "text":"Leishmania sp 16kD IgG Ser Ql IB",
            "answer":[
              {
                "valueString":"432"
              }
            ],
            "linkId":"128851"
          }
        ],
        "meta":{
          "lastUpdated":"2024-10-03T10:07:55.843374Z",
          "versionId":"1970",
          "extension":[
            {
              "url":"ex:createdAt",
              "valueInstant":"2024-10-03T10:07:42.342731Z"
            }
          ]
        },
        "author":{
          "identifier":{
            "type":{
              "coding":[
                {
                  "code":"UID",
                  "system":"urn:system:aidbox",
                  "display":"User ID"
                }
              ]
            },
            "value":"admin",
            "system":"http://localhost:8765"
          }
        },
        "status":"completed",
        "authored":"2024-10-03T10:07:55.664Z",
        "resourceType":"QuestionnaireResponse",
        "questionnaire":"http://loinc.org/q/100109-8"
      },
      "fullUrl": "http://aidbox-server/fhir/QuestionnaireResponse/458e771c-0ff1-4858-ac07-93b7a10c8e3b",
      "request":{
        "method":"POST",
        "url":"/fhir/QuestionnaireResponse"
      }
    }
  ]
}

{
  "resourceType":"Bundle",
  "type":"history",
  "timestamp":"2024-10-03T10:07:55Z",
  "entry":[
    {
      "resource":{
        "resourceType":"AidboxSubscriptionStatus",
        "status":"active",
        "type":"event-notification",
        "notificationEvent":[
          {
            "eventNumber":1,
            "focus":{
              "reference":"QuestionnaireResponse/458e771c-0ff1-4858-ac07-93b7a10c8e3b"
            }
          }
        ],
        "topic":"http://example.org/FHIR/R5/SubscriptionTopic/QuestionnaireResponse-topic",
        "topic-destination":{
          "reference":"AidboxTopicDestination/kafka-destination"
        }
      }
    },
    {
      "fullUrl": "http://aidbox-server/fhir/QuestionnaireResponse/458e771c-0ff1-4858-ac07-93b7a10c8e3b",
      "request":{
        "method":"POST",
        "url":"/fhir/QuestionnaireResponse"
      }
    }
  ]
}

{
  "resourceType":"Bundle",
  "type":"history",
  "timestamp":"2024-10-03T10:07:55Z",
  "entry":[
    {
      "resource":{
        "resourceType":"AidboxSubscriptionStatus",
        "status":"active",
        "type":"event-notification",
        "notificationEvent":[
          {
            "eventNumber":1,
            "focus":{
              "reference":"QuestionnaireResponse/458e771c-0ff1-4858-ac07-93b7a10c8e3b"
            }
          }
        ],
        "topic":"http://example.org/FHIR/R5/SubscriptionTopic/QuestionnaireResponse-topic",
        "topic-destination":{
          "reference":"AidboxTopicDestination/kafka-destination"
        }
      }
    }
  ]
}

Entry extensions

When includeEntryAction or includeVersionId is enabled on the AidboxTopicDestination, each resource Bundle.entry (except the AidboxSubscriptionStatus entry) will include an extension array with the corresponding values:

{
  "resource": { ... },
  "fullUrl": "...",
  "request": { ... },
  "extension": [
    {
      "url": "http://health-samurai.io/fhir/core/StructureDefinition/bundle-entryActionCode",
      "valueCode": "create"
    },
    {
      "url": "http://health-samurai.io/fhir/core/StructureDefinition/bundle-entryVersionId",
      "valueId": "1970"
    }
  ]
}

Only the extensions corresponding to the enabled flags are included. The extensions appear on entries for both full-resource and id-only content modes. In empty mode there are no resource entries, so extensions are not included.

Focus reference with version id

When includeVersionIdInFocusReference is enabled, the focus.reference in each notificationEvent includes the version history path:

{
  "notificationEvent": [
    {
      "eventNumber": 1,
      "focus": {
        "reference": "Patient/123/_history/5"
      }
    }
  ]
}

Without this flag, the reference would be Patient/123.

Correlation id

Send a correlation id on the request that creates, updates, or deletes a resource, and Aidbox copies it into the resulting notification. Use the module.topics.correlation-id-header setting to name the request header Aidbox reads (default: x-topic-correlation-id). Aidbox treats the value as an opaque string: it does not parse, validate, or generate one on your behalf.

{
  "notificationEvent": [
    {
      "eventNumber": 1,
      "focus": {
        "reference": "Patient/123"
      },
      "correlationId": "order-4711"
    }
  ]
}

If the request carries no correlation id header, Aidbox omits correlationId from the notification.

A single header covers every event the request produces. On a transaction bundle or a conditional delete that matches several resources, every resulting notificationEvent carries the same correlation id. The value survives redelivery: Aidbox persists it with the event before the first delivery attempt, so retries and catch-up deliveries after downtime carry the same id as the original.

All AidboxTopicDestination channels support this: webhook, Kafka, GCP Pub/Sub, and both NATS destination kinds. On NATS, Aidbox also sets the correlation id as a native NATS message header, alongside the copy in notificationEvent. See Message headers for details.

There is no per-AidboxTopicDestination toggle for this feature: whether a notification carries a correlation id depends only on the triggering request.

Delivery logging

When enableLogging is set to true on an AidboxTopicDestination, Aidbox logs the AidboxSubscriptionStatus resource after each delivery attempt. On successful delivery, the status is logged as-is. On failure, the status includes an error array.

The logLevel element controls the log level used for these messages. For example, setting logLevel to "error" emits logs at the error level, while the default "info" emits at the info level. The configured level must be equal to or higher than the system-wide log level setting — otherwise the messages will be filtered out and will not appear in the logs.

{
  "resourceType": "AidboxSubscriptionStatus",
  "id": "auto-generated-uuid",
  "status": "active",
  "type": "event-notification",
  "topic": "http://example.org/FHIR/R5/SubscriptionTopic/patient-topic",
  "topic-destination": {
    "reference": "AidboxTopicDestination/my-destination"
  },
  "error": ["Connection refused: localhost:1"]
}

The logs are emitted with the event key topic-destination/status.