Design a pluggable messaging backend architecture for `websubhub`

[ayeshLK]

Description

$subject

[ayeshLK]

Current structure of the websubhub codebase

Following diagram showcases the existing architecture of the websubhub code-base.

base module

The base ballerina module deals with the WebSub hub functionality, state management, and content delivery.

Following functionalities are available in the base module:

• WebSub hub API
• State management logic
• Message (received from the broker) serialization and de-serialization logic.
• Content delivery (to subscribers) logic

common module

The common types and utility functions are defined here.

Following public APIs are exposed from the common module.

// Generates a unique Id for a subscriber.
public isolated function generateSubscriberId(string topic, string callbackUrl) returns string;

// Generates a consumer group name for the kafka-consumer.
public isolated function generateGroupName(string topic, string callbackUrl) returns string;

// Generates a random `string` of 10 characters.
public isolated function generateRandomString() returns string;

// Logs errors with proper details.
public isolated function logError(string msg, error 'error, *log:KeyValues keyValues);

config module

The configurations for the WebSubHub deployment are defined here.

Following configurations are exposed via the config module.

// Common configurations used to configure the websubhub server
public configurable common:ServerConfig server = ?;

// Configurations related to websubhub server state
public configurable common:ServerStateConfig state = ?;

// Kafka connection related configurations
public configurable common:KafkaConfig kafka = ?;

// Message delivery related configurations
public configurable common:HttpClientConfig delivery = ?;

// Flag indicating whether security is enable or not. 
// This is derived by checking whether server authentication configuration is available or not
public final boolean securityOn = server.auth is common:JwtValidatorConfig;

connections module

The message broker connections (or clients) used within the WebSubHub are defined in this module.

Following public client are exposed from the connections module.

// Public client used to persist WebSubHub state changed events (topic-reg, topic-dereg etc).
public final kafka:Producer statePersistProducer = ...;

// Public client used to receive WebSubHub state change events.
public final kafka:Consumer websubEventsConsumer = ...;

Following public APIs are exposed from the connections module.

// API to create a Kafka consumer for the websub `subscriber`
public isolated function createMessageConsumer(string topicName, string groupName, int[]? partitions = ()) returns kafka:Consumer|error;

persistence module

Message persistence related APIs are exposed via this module. The persistence API uses the public client exposed from the connections module to complete the persistence action.

Following public APIs are exposed from the persistence module.

// Persist a topic registration.
public isolated function addRegsiteredTopic(websubhub:TopicRegistration message) returns error?;

// Persist a topic de-registration.
public isolated function removeRegsiteredTopic(websubhub:TopicDeregistration message) returns error?;

// Persist a subscription.
public isolated function addSubscription(websubhub:VerifiedSubscription message) returns error?;

// Persist a stale subscription.
public isolated function addStaleSubscription(common:StaleSubscription message) returns error?;

// Persist an unsubscription.
public isolated function removeSubscription(websubhub:VerifiedUnsubscription message) returns error?;

// Persist content notification.
public isolated function addUpdateMessage(string topicName, websubhub:UpdateMessage message, map<string|string[]> headers = {}) returns error?;

[ayeshLK]

Proposed structure for websubhub codebase

Changes required for common module

1. Update the structure of the common:KafkaConfig record

public type KafkaConfig record {|
    record {| 
        KafkaConnectionConfig connection;
        KafkaConsumerConfig consumer;
    |}  kafka;    
|};

// For other broker configurations also we will introduce a similar configuration type

2. Update logError API

The existing logError API was introduced earlier due to a limitation in the ballerina/log module, which couldn’t print stack traces at the time. Since that limitation has now been resolved, we should reconsider the need for this API.

Currently, logError is used throughout the codebase with a severity tag (e.g., FATAL or RECOVERABLE) to indicate the nature of the issue. Instead of a single API with a severity parameter, it would be cleaner to introduce two dedicated APIs such as logFatalError and logRecoverableError. This also gives us room to evolve these functions later with a more structured error format.

Establishing a unified error logging approach would also improve troubleshooting. As part of this, we could introduce a shared logging structure within the common module and use it consistently across the entire WebSubHub package.

3. Remove generateGroupName API

The generateGroupName API was originally introduced to create a unique consumer group for the Kafka consumer associated with a WebSub subscription. However, as the name suggests, this functionality is broker-specific. Therefore, it would be better to remove it from the common module and move it into a dedicated module that handles broker-specific behavior.

Changes required for config module

1. Rename the kafka configuration as broker

With this change the configuration would look like this.

// Message broker related configurations
public configurable common:KafkaConfig broker = ?;

Eventually broker configuration would be a union of multiple broker configurations (eg: common:KafkaConfig|common:SolaceConfig broker = ...;)

With this change the configuration structure in the Config.toml would be slightly shifted.

// Existing configuration structure 

[websubhub.config.kafka.connection]
bootstrapServers = "localhost:9092"

[websubhub.config.kafka.consumer]
maxPollRecords = 50
pollingInterval = 10.0
gracefulClosePeriod = 5.0

// Proposed configuration structure

[websubhub.config.broker.kafka.connection]
bootstrapServers = "localhost:9092"

[websubhub.config.broker.kafka.consumer]
maxPollRecords = 50
pollingInterval = 10.0
gracefulClosePeriod = 5.0

Changes required for connections module

1. Changing the module name

The connection module currently manages all broker-specific clients and APIs to create clients. It would be clearer and more accurate to rename it to something more descriptive, such as broker or messaging.

2. Introduce Message record.

The Message record will serve as an abstraction for broker-specific message formats. Initially, it will include two fields: payload and metadata.

public type Message record {|
    // The message payload
    byte[] payload;
    // The metadata associated with the message (e.g., Kafka message headers or JMS message properties)
    map<string|string[]> metadata?;
|};

3. Introduce Producer API

The Producer API will abstract the broker-specific producer interactions and provide a unified interface for publishing messages, making it usable consistently across the entire WebSubHub package.

public type Producer distinct client object {

    // The API name could be changed to `publish` or `produce`, but `send` is used here as this is a generic API (similar to JMS spec)
    function send(string topic, Message message) returns error?;

    // Closes the underlying broker connection.
    function close() returns error?;
};

4. Introduce Consumer API

The Consumer API abstract out the broker-specific consumer interaction and provides a unified interface for consuming and acknowledging messages.

public type Consumer distinct client object {

    function receive() returns Message|Message[]|error?;

    function ack(Message|Message[] message) returns error?;

    function nack(Message|Message[] message) returns error?;

    function close() returns error?;
};

5. Update createMessageConsumer API

public isolated function createConsumer(websubhub:VerifiedSubscription subscription) returns Consumer|error;

6. Update the public client types

public final Producer statePersistProducer = ...;

public final Consumer websubEventsConsumer = ...;

7. Introduce an API to retrieve a Producer per topic

This API is needed to retrieve the Producer associated with a given topic. Some brokers require a separate producer per topic (e.g., Azure Service Bus), so including this API would be beneficial.

public isolated function getProducer(string topic) returns Producer|error;

Changes required for persistence module

Since the current persistence module depends on the APIs provided by the broker/messaging module, the proposed updates—where message persistence logic is moved into the broker/messaging module itself—will require changes to the persistence module as well. It will need to be updated to use the new APIs introduced in the revised broker/messaging module.

Changes required for base module

With the proposed changes, the base module will no longer interact directly with broker-specific message structures or clients. Instead, it will be updated to use the APIs provided by the broker/messaging module.

Revised codebase structure

Following diagram showcases the revised architecture of the websubhub code-base.

[ayeshLK]

[25th Nov 2025] We had a proposal review and the following points were discussed:

Attendees: @shafreenAnfar @gayaldassanayake @RadCod3 @ayeshLK

Discussion points:

• For the messaging API, we can consider two possible approaches: a JMS-style API or an API similar to the Ballerina Message Store.
  • If we choose a JMS-style API, the proposed specification (Message, Producer, Consumer) is valid and can be used as-is.
  • If we opt for a Message Store–based API, we would need both Message and Store concepts, where the Store abstracts producer and consumer functionality.
  • When mapping WebSub publishers and subscribers to the messaging API, a publisher aligns with a producer, and a subscriber aligns with a consumer. Because of this, the Message Store model may not be appropriate—Message Store does not naturally support multiple consumers (one per WebSub subscriber). Even if we introduce consumer creation per WebSub subscriber within a Message Store–based API, the resulting design would essentially converge toward something similar to the proposed JMS-like API.
  • Therefore, the proposed JMS-style API is the more suitable option.
• Regarding the proposed API:
  • The receive function of the Consumer API should return either a Message or (). If we consider a Kafka-like implementation, we can internally iterate through the fetched messages and dispatch them one by one for each receive call.
  • The intention is to keep the API symmetrical for both the Producer and Consumer. Since general-purpose message brokers typically follow a pattern of publishing one message at a time and receiving one message at a time, this approach is appropriate.
  • Based on this reasoning, the proposed Consumer API would be restructured as follows:

    public type Consumer distinct client object {
    
        function receive() returns Message|error?;
    
        function ack(Message message) returns error?;
    
        function nack(Message message) returns error?;
    
        function close() returns error?;
    };

• Regarding the configuration structure:
  • In the proposed design, we currently have two configuration sets: connection, which contains the common configurations for both the producer and consumer, and consumer, which contains broker-specific configurations.
  • Instead of defining them at the same level, it would be more appropriate for connection to sit at a higher level, with the consumer-specific configurations nested beneath it.
  • Based on this, the restructured KafkaConfig record would look as follows:

    public type KafkaConfig record {|
        record {| 
            *KafkaConnectionConfig;
            KafkaConsumerConfig consumer;
        |}  kafka;    
    |};

  • For the configuration name, we can use store.

    // Message store related configurations
    public configurable common:KafkaConfig store = ?;

  • With this change the configuration structure in the Config.toml would like this.

     [websubhub.config.store.kafka]
     bootstrapServers = "localhost:9092"
    
     [websubhub.config.store.kafka.consumer]
     maxPollRecords = 50
     pollingInterval = 10.0
     gracefulClosePeriod = 5.0

• Based on the above discussion points, it would be appropriate to rename the messaging/broker module to store, with the APIs then referenced as store:Producer, store:Consumer, and store:Message.

[ayeshLK]

[27th Nov 2025] Update as of now

Since the messaging store APIs are shared by both websubhub and consolidator, I separated that logic into a standalone Ballerina package (message.store) instead of embedding it inside each codebase. Both services now depend on this package through the Ballerina workspace, so there’s no need to publish message.store to Ballerina Central.

In addition to the base messaging APIs (Producer, Consumer, Message), I’ve exposed a few public functions that allow initializing a broker-specific implementation of those base APIs. These APIs create the broker-specific implementation internally but only expose the base interfaces, ensuring the concrete implementation remains encapsulated within the message.store package.

# Initialize a producer for Kafka message store.
#
# + config - The Kafka connection configurations
# + clientId - The client Id
# + return - A `store:Producer` for Kafka message store, or else return an `error` if the operation fails
public isolated function createKafkaProducer(KafkaConfig config, string clientId) returns Producer|error;

# Initialize a consumer for Kafka message store.
#
# + config - The Kafka connection configurations
# + groupId - The Kafka consumer group to which this consumer should belong to
# + topic - The Kafka topic to which the consumer should received events for
# + autoCommit - The flag to enable auto-commit offsets  
# + offsetReset - The offset reset strategy if no initial offset
# + partitions - The Kafka topic partitions
# + return - A `store:Consumer` for Kafka message store, or else return an `error` if the operation fails
public isolated function createKafkaConsumer(store:KafkaConfig config, string groupId, string topic, boolean autoCommit = true, kafka:OffsetResetMethod? offsetReset = (), int[]? partitions = ()) returns Consumer|error;

The current APIs are specific for the Apache Kafka–based implementation. As we expand WebSubHub’s messaging backend support, we will need to introduce additional APIs for other brokers such as Solace, RabbitMQ etc.

The message.store package includes both the broker-specific implementation and its corresponding configuration model. Below is the configuration structure introduced for Apache Kafka. As we add support for additional messaging backends, each will expose its own configuration following a similar pattern.

# Defines the Kafka message store configurations
public type KafkaMessageStore record {|
    # Configurations related to Kafka message store connection
    KafkaConfig kafka;
|};

# Defines the complete set of Kafka configurations required for the application.
public type KafkaConfig record {|
    *KafkaConnectionConfig;
    # Kafka producer-specific configurations
    KafkaProducerConfig producer = {};
    # Kafka consumer-specific configurations
    KafkaConsumerConfig consumer;
|};

# Defines the configurations for establishing a connection to a Kafka cluster.
public type KafkaConnectionConfig record {|
    # A list of remote server endpoints for the Kafka brokers (e.g., "localhost:9092")
    string|string[] bootstrapServers;
    # SSL/TLS configurations for the Kafka connection
    kafka:SecureSocket secureSocket?;
    # Authentication-related configurations for the Kafka connection
    kafka:AuthenticationConfiguration auth?;
    # The security protocol to use for the broker connection (e.g., PLAINTEXT, SSL)
    kafka:SecurityProtocol securityProtocol = kafka:PROTOCOL_PLAINTEXT;
|};

# Defines configurations for the Kafka producer.
public type KafkaProducerConfig record {|
    # Number of acknowledgments
    kafka:ProducerAcks acks = kafka:ACKS_ALL;
    # Number of retries to resend a message
    int retryCount = 3;
|};

# Defines configurations for the Kafka consumer.
public type KafkaConsumerConfig record {|
    # The maximum number of records to return in a single call to `poll` function
    int maxPollRecords;
    # The polling interval in seconds for fetching new records
    decimal pollingInterval = 10;
    # The timeout duration (in seconds) for a graceful shutdown of the consumer
    decimal gracefulClosePeriod = 5;
|};

With the introduction of the message.store package, following changes will be introduced into the websubhub package.

• Configuration structure of the store will be changed like this.

  import wso2/message.store;
  
  # Messaging store connection related configurations
  public configurable store:KafkaMessageStore store = ?;

• The connections module will not be renamed.

• The connections module will expose following public clients.

  import wso2/message.store;
  
  public final store:Producer statePersistProducer = ...;
  
  public final store:Consumer websubEventsConsumer = ...;

• The connections module will expose following public APIs.

  # Initialize a `store:Consumer` for a WebSub subscriber.
  #
  # + subscription - The WebSub subscriber details
  # + return - A `store:Consumer` for the message store, or else return an `error` if the operation fails
  public isolated function createConsumer(websubhub:VerifiedSubscription subscription) returns store:Consumer|error;
  
  # Retrieves a message producer per topic.
  #
  # + topic - The message store topic
  # + return - A `store:Producer` for the message store, or else an `error` if the operation fails
  public isolated function getMessageProducer(string topic) returns store:Producer|error;

• With the configuration module introduced in message.store, I identified a few producer-specific settings that should also be exposed as configurables. As a result, the store configuration in Config.toml will now look like this:

  [websubhub.config.store.kafka]
  bootstrapServers = "localhost:9092"
  
  [websubhub.config.store.kafka.producer]
  acks = "all"
  retryCount = 3
  
  [websubhub.config.store.kafka.consumer]
  maxPollRecords = 50
  pollingInterval = 10.0
  gracefulClosePeriod = 5.0

Based on the above changes, I’ve submitted the initial PR [1] for the implementation.

[1] - #18

[ayeshLK]

[9th Dec 2025] Update as of now

Message brokers typically provide producer–consumer functionality, but many also include administrative APIs for creating topics and subscriptions. While this is optional for some brokers (e.g., Kafka, JMS-compliant brokers), others do not support dynamic creation of these entities. To address this, we are introducing a new Administrator API in the messagestore package to handle manual creation and management of topics and subscriptions.

Following is the API spec for the messagestore:Administrator:

public type Administrator distinct client object {

    function createTopic(string topic) returns TopicExists|error?;

    function deleteTopic(string topic) returns TopicNotFound|error?;

    function createSubscription(string topic, string subscriberId) returns SubscriptionExists|error?;

    function deleteSubscription(string topic, string subscriberId) returns SubscriptionNotFound|error?;

    function close() returns error?;
};

Following error types will also be introduced:

public type TopicExists distinct error;
public type TopicNotFound distinct error;
public type SubscriptionExists distinct error;
public type SubscriptionNotFound distinct error;

[ayeshLK]

[9th Jan 2026]

As of now we got into an error while introducing the store configuration as suggested in the proposal;

public configurable store:KafkaMessageStore|store:SolaceMessageStore store = ?;

Hence we had to introduce the configuration like below;

public configurable record {|store:KafkaConfig kafka?; store:SolaceConfig solace?;|} store = ?;

[ayeshLK]

[21st Jan 2026]

When closing a Consumer, JMS message brokers require the consumer to be closed before the associated subscriptions can be removed. In such cases, administrator-driven subscription deletion does not align well with the current WebSubHub architecture.

Ideally, the consumer.close() API should accept a flag that indicates the closure intent. This would allow the consumer implementation to decide whether it should trigger the appropriate subscription removal logic at the broker level.

This requirement is not limited to JMS brokers. Other broker implementations may also need additional handling based on the closure type, for example:

• Temporary closure: the consumer is expected to reuse the existing subscription once it becomes available again.
• Permanent closure: the existing subscription should be removed and not reused.

Introducing such a mechanism would make consumer shutdown behavior more explicit and extensible across different broker implementations.

Hence, it would be ideal to improve the Consumer API like below;

public type Consumer distinct client object {

    // other remove methods

    isolated remote function close(ClosureIntent intent = TEMPORARY) returns error?;
};

With this change, the following enum will be introduced in the message store. This allows the consumer.close() behavior to be expressed explicitly and makes the API extensible. If new closure semantics are required in the future, they can be added by extending the same enum without introducing breaking changes.

# Represents the intent of closing a `Consumer`. This is used to indicate how the underlying broker-side resources
# (such as subscriptions) should be handled when a consumer is closed.
public enum ClosureIntent {
    # Indicates a temporary shutdown where the consumer may reconnect and reuse the existing subscription
    TEMPORARY,
    # Indicates a permanent shutdown where the consumer will not be reused and any associated broker-side subscriptions should be removed
    PERMANENT
};

[ayeshLK]

[25th Feb 2026]

When delivering messages to WebSub subscribers, message serialization/deserialization failures may occur at the WebSubHub (e.g., due to corrupted messages). In such scenarios, the WebSubHub should handle the failure gracefully without blocking message processing. Therefore, we need a mechanism to move failed messages out of the processing queue, making a dead-letter mechanism an appropriate solution.

Regarding where this feature should be introduced, I believe it should be part of the Consumer API. This approach is inspired by the Azure Service Bus consumer model, where the deadLetter functionality is exposed through the consumer API [1]. Similarly, other broker implementations such as RabbitMQ [2] and Solace [3] support comparable behavior via negative acknowledgement (nack) mechanisms.

Based on this, the following is the proposed updated client API:

public type Consumer distinct client object {

    // other remote methods

    isolated remote function deadLetter(Message message) returns error?;
};

Configuration Considerations

The configuration model would depend on the underlying message store:

• Azure Service Bus: Since dead-lettering is integrated into the consumer API, additional configuration may not be required.
• Kafka / Solace: These systems typically require a dedicated dead-letter queue/topic configuration at the consumer level.

Therefore, dead-letter support can be introduced as an optional configuration, enabling or disabling the feature based on availability.

Proposed configuration priority:

1. Subscriber-level dead-letter configuration (highest priority)
   • A subscriber may specify a dead-letter queue during subscription if custom behavior is required.
2. System-level dead-letter configuration (fallback/default)
   • A global dead-letter queue configured for the platform.

[1] - https://learn.microsoft.com/en-us/java/api/com.azure.messaging.servicebus.servicebusreceiverclient?view=azure-java-stable#com-azure-messaging-servicebus-servicebusreceiverclient-deadletter(com-azure-messaging-servicebus-servicebusreceivedmessage)
[2] - https://central.ballerina.io/ballerinax/rabbitmq/3.3.2#Client-basicNack
[3] - https://central.ballerina.io/xlibb/solace/0.2.2#MessageConsumer-nack

[ayeshLK]

Closing this issue as we have covered all the bases required to introducing pluggable messaging backend architecture. PR list can be found below;

• Introduce pluggable messaging backend support for websubhub #18
• Introduce message-store admin functionality in the websubhub #20
• Introduce DLQ support for messagestore:Consumer API #51