@xeno-js/postgresql provides native PostgreSQL integration within the Xeno.JS ecosystem. It leverages Drizzle ORM and node-postgres to manage the database context, atomic transactions, and the transactional outbox pattern with a background worker.

Installation

npm install @xeno-js/postgresql pg drizzle-orm

The package requires Node.js 20 or higher and @xeno-js/core as a peer dependency.

1. Plugin Configuration and Database Connection

Problem. Enterprise applications require centralized and secure management of database connection pools, with specific parameters for development and production environments.

Principle. Connection configuration must be explicit, encapsulated in a dedicated factory, and injected into the framework's DI container.

Mechanism. The withPostgresql function exposes a configuration plugin based on the Xeno.JS builder, using DbClientFactory to instantiate the Drizzle client via a pg connection pool.

Implementation:

import { XenoAppBuilder } from '@xeno-js/core'
import { withPostgresql } from '@xeno-js/postgresql'

const app = await new XenoAppBuilder()
  .addDatabase(
    withPostgresql((options, config) => {
      options.connectionString = config.get('DATABASE_URL')
      options.max = 20
      options.idleTimeoutMillis = 30000
    }),
  )
  .build()

Behavior: During bootstrap, the client is configured by applying timeouts, idle pool error handling, and automatic SSL flags in production environments.

Boundary: The plugin handles exclusively database client configuration and persistence module initialization. It does not define application business schemas.

2. Transactional Outbox Pattern and Unit of Work

Problem. During write operations, saving data to the database and publishing domain events to a message broker must occur atomically to prevent state inconsistencies in case of failures.

Principle: Uncommitted events generated by Aggregate Roots must be saved in the same database transaction where the business state is persisted.

Mechanism: PostgresqlUnitOfWork intercepts Drizzle transactions, extracts events from aggregates, and persists them atomically inside the dedicated outbox table (_outbox_messages) dynamically generated via createOutboxTable.

Implementation:

Enable Outbox directly in the withPostgresql plugin configuration:

withPostgresql((options) => {
  options.connectionString = process.env.DATABASE_URL
  options.outbox = {
    enable: true,
    worker: true,
    name: ['Order', 'Payment'], // Monitored Aggregate Root names
  }
})

Behavior: When a command is executed within a transaction, the generated events are serialized in JSON format and inserted into the outbox table with a PENDING status. If the transaction fails, a full rollback is executed.

Boundary: The Unit of Work ensures atomic event persistence, but does not perform network dispatch to external consumers, delegating this task to the asynchronous worker.

3. Background Outbox Worker

Problem. Transactionally saved outbox messages must be processed and published asynchronously to the event system without overloading the main thread or causing duplications in distributed environments.

Principle: Pending message processing must use secure row-level locking mechanisms to allow concurrent execution of multiple workers.

Mechanism: OutboxWorker performs periodic polling cycles using the SQL query SELECT ... FOR UPDATE SKIP LOCKED to securely lock up to 50 pending messages per table without generating blocking conflicts.

Implementation:

The worker starts automatically when worker: true is configured in the outbox block. Specific handlers for event management can also be registered:

withPostgresql((options) => {
  options.outbox = {
    enable: true,
    worker: true,
    name: 'Order',
    handlers: [
      (scope) => new OrderCreatedHandler(scope.resolve(TOKENS.LOGGER)),
    ],
  }
})

Behavior: The worker extracts messages with a PENDING status, creates an isolated synthetic execution context via AsyncLocalStorage, and publishes the event via the IEventDispatcher. Upon success, the status changes to PUBLISHED; upon error, the retryCount counter is incremented (up to a maximum of 3 attempts before moving to FAILED).

Boundary: The worker operates as a background process independent of the HTTP request lifecycle, focusing on reliable message delivery.

License

Released under the MIT License.