Message DB

Microservice Native Event Store and Message Store for Postgres

A fully-featured event store and message store implemented in PostgreSQL for Pub/Sub, Event Sourcing, Messaging, and Evented Microservices applications.

• Pub/Sub
• JSON message data
• Event streams
• Stream categories
• Metadata
• Message queues
• Message storage
• Consumer groups
• Service host
• Administration tools
• Reports

An event sourcing and Pub/Sub message store built on Postgres for simple cloud or local hosting. A minimalist implementation of the essential features of tools like Event Store or Kafka, with built-in support for messaging patterns like Pub/Sub, and consumer patterns like consumer groups.

Message DB was extracted from the Eventide Project to make it easier for users to write clients in the language of their choosing.

A complete user guide is available on the Eventide Project docs site:

http://docs.eventide-project.org/user-guide/message-db/

Message DB can be installed either as a Ruby Gem, an NPM package, or can simply be cloned from this repository.

git clone git@github.com:message-db/message-db.git

gem install message-db

npm install @eventide/message-db

Running the database installation script creates the database, schema, table, indexes, functions, views, types, a user role, and limit the user's privileges to the message store's public interface.

Make sure that your default Postgres user has administrative privileges.

The installation script is in the database directory of the cloned repo. Change directory to the message-db directory where you cloned the repo, and run the script:

database/install.sh

If you installed Message DB via RubyGems, a database installation Ruby executable will be installed with the message-db gem.

The executable will be in the gem executable search path and may also be executed through bundler:

bundle exec mdb-create-db

For more information about Ruby executables installed with the message-db Ruby Gem, see the Eventide docs on the administration tools that are bundled with the gem:

http://docs.eventide-project.org/user-guide/message-db/tools.html

The message-db NPM module doesn't ship with any special tooling other than the bundled scripts.

To execute the installation script, navigate to the directory where the message-db module is installed and run the script:

install.sh

By default, the database creation tool will create a database named message_store.

If you prefer either a different database name, you can override the name using the DATABASE_NAME environment variable.

DATABASE_NAME=some_other_database database/install.sh

If you need to drop the database (for example, on a local dev machine):

database/uninstall.sh

If you're upgrading a previous version of the database:

database/update.sh

The message store provides an interface of Postgres server functions that can be used with any programming language or through the psql command line tool.

Interaction with the underlying store through the Postgres server functions ensures correct writing and reading messages, streams, and categories.

Write a JSON-formatted message to a named stream, optionally specifying JSON-formatted metadata and an expected version number.

write_message(
  id varchar,
  stream_name varchar,
  type varchar,
  data jsonb,
  metadata jsonb DEFAULT NULL,
  expected_version bigint DEFAULT NULL
)

Position of the message written.

SELECT write_message('a11e9022-e741-4450-bf9c-c4cc5ddb6ea3', 'someStream-123', 'SomeMessageType', '{"someAttribute": "some value"}', '{"metadataAttribute": "some meta data value"}');

-[ RECORD 1 ]-+--
write_message | 0

Example: https://github.com/message-db/message-db/blob/master/database/write-test-message.sh

Retrieve messages from a single stream, optionally specifying the starting position, the number of messages to retrieve, and an additional condition that will be appended to the SQL command's WHERE clause.

get_stream_messages(
  stream_name varchar,
  position bigint DEFAULT 0,
  batch_size bigint DEFAULT 1000,
  condition varchar DEFAULT NULL
)

SELECT * FROM get_stream_messages('someStream-123', 0, 1000, condition => 'messages.time >= current_time');

-[ RECORD 1 ]---+---------------------------------------------------------
id              | 4b96f09e-104a-4b1f-b198-5b3b46cf1d06
stream_name     | someStream-123
type            | SomeType
position        | 0
global_position | 1
data            | {"attribute": "some value"}
metadata        | {"metaAttribute": "some meta value"}
time            | 2019-11-24 17:56:09.71594
-[ RECORD 2 ]---+---------------------------------------------------------
id              | d94e79e3-cdda-49a3-9aad-ce5d70a5edd7
stream_name     | someStream-123
type            | SomeType
position        | 1
global_position | 2
data            | {"attribute": "some value"}
metadata        | {"metaAttribute": "some meta value"}
time            | 2019-11-24 17:56:09.75969

Example: https://github.com/message-db/message-db/blob/master/test/get-stream-messages/get-stream-messages.sh

Retrieve messages from a category of streams, optionally specifying the starting position, the number of messages to retrieve, the correlation category for Pub/Sub, consumer group parameters, and an additional condition that will be appended to the SQL command's WHERE clause.

CREATE OR REPLACE FUNCTION get_category_messages(
  category_name varchar,
  position bigint DEFAULT 0,
  batch_size bigint DEFAULT 1000,
  correlation varchar DEFAULT NULL,
  consumer_group_member bigint DEFAULT NULL,
  consumer_group_size bigint DEFAULT NULL,
  condition varchar DEFAULT NULL
)

SELECT * FROM get_category_messages('someCategory', 1, 1000, correlation => 'someCorrelationCategory', consumer_group_member => 1, consumer_group_size => 2, condition => 'messages.time >= current_time');

-[ RECORD 1 ]---+---------------------------------------------------------
id              | 28d8347f-677e-4738-b6b9-954f1b15463b
stream_name     | someCategory-123
type            | SomeType
position        | 0
global_position | 111
data            | {"attribute": "some value"}
metadata        | {"correlationStreamName": "someCorrelationCategory-123"}
time            | 2019-11-24 17:51:49.836341
-[ RECORD 2 ]---+---------------------------------------------------------
id              | 57894da7-680b-4483-825c-732dcf873e93
stream_name     | someCategory-456
type            | SomeType
position        | 1
global_position | 1111
data            | {"attribute": "some value"}
metadata        | {"correlationStreamName": "someCorrelationCategory-123"}
time            | 2019-11-24 17:51:49.879011

Note: Where someStream-123 is a stream name, someStream is a category. Reading the someStream category retrieves messages from all streams whose names start with someStream and are followed by an ID, or where someStream is the whole stream name.

Example: https://github.com/message-db/message-db/blob/master/test/get-category-messages/get-category-messages.sh

• write_message
• get_stream_messages
• get_category_messages
• get_last_stream_message
• stream_version
• id
• cardinal_id
• category
• is_category
• acquire_lock
• hash_64
• message_store_version

The message store is a single table named messages.

By default, the message store database is named message_store.

All message store database objects are contained within a schema named message_store.

A role named message_store is created. The message_store role is given the LOGIN attribute, but no password is assigned. A password can be assigned to the role, or the message_store role can be granted to another Postgres user.

View complete source code at:

https://github.com/message-db/message-db/tree/master/database

The Postgres Message Store is released under the MIT License.