Class JdbcDelivery

java.lang.Object
org.j7m.middleware.integrations.JdbcDelivery
public final class JdbcDelivery extends Object
JDBC transactional outbox/inbox for H2 and PostgreSQL using application-supplied drivers. Enqueue and consume methods never commit, roll back, close or replace the caller's connection. Callers must use one non-autocommit transaction for their application writes and these methods, and roll back that whole transaction on any exception. Relay owns its short transactions and sends at least once; there is deliberately no atomic DB-plus-remote-queue commit claim. Instances are immutable and thread-safe; connections must not be shared between threads.

  • Constructor Details

    • JdbcDelivery

      public JdbcDelivery(String scope, Observability telemetry)
      Binds DB identities to one authenticated queue/inbox scope.
      Parameters:
      scope - scope identifier, at most 128 ASCII identifier characters
      telemetry - optional observability integration; null disables span export
      Throws:
      IllegalArgumentException - for invalid scope

  • Method Details

    • initialize

      public static void initialize(Connection connection, JdbcDelivery.Dialect dialect) throws SQLException
      Creates additive integration tables; execute as an explicit migration on a dedicated connection. Never call during an application transaction: DDL transaction semantics differ by database. Tables and tombstones remain on rollback/uninstall to preserve delivery identity.
      Parameters:
      connection - dedicated migration connection
      dialect - supported database dialect
      Throws:
      SQLException - for schema creation failure

    • enqueue

      public long enqueue(Connection connection, String id, String aggregate, TracedQueue.Message message, Instant createdAt, Instant retainUntil) throws SQLException, IOException
      Appends an ordered message in the caller's application transaction. Existing IDs with identical envelope/aggregate return their original sequence; conflicting IDs fail. Concurrent first insertion of an aggregate can require retry of the whole transaction.
      Parameters:
      connection - exact non-autocommit application transaction
      id - stable message identity
      aggregate - aggregate ordering identity
      message - immutable application payload and retry-stable context
      createdAt - application-selected timestamp used only for retention metadata
      retainUntil - earliest time payload may be removed after successful publication
      Returns:
      aggregate-local sequence
      Throws:
      SQLException - for inactive transaction, conflicts or database errors
      IOException - for invalid message envelope

    • pending

      public List<JdbcDelivery.Pending> pending(Connection connection, int maximum) throws SQLException
      Selects only the first unsent message of each aggregate, preserving send order across relays. Sending/marking an older row more than once is safe only because the real queue retains the same message identity and content fingerprint. This method does not claim rows or commit.
      Parameters:
      connection - caller-owned connection
      maximum - bounded head count in [1,256]
      Returns:
      immutable ordered head rows
      Throws:
      SQLException - for database access failure

    • relayOnce

      public int relayOnce(JdbcDelivery.Connections connections, org.j7m.middleware.distributed.ReplicatedQueue queue, String name, int maximum, Instant deadline) throws SQLException, IOException
      Publishes head rows at least once and marks only confirmed queue publications in short DB transactions. Stable queue requests bind both DB scope and message ID. A timeout preserves the pending row. Concurrent relays may resend the same head but cannot publish a successor before confirmed send.
      Parameters:
      connections - independent relay connection factory
      queue - authenticated queue bound to this scope
      name - destination queue name
      maximum - at most 256 aggregate heads
      deadline - absolute queue deadline
      Returns:
      number of confirmed/marked rows
      Throws:
      SQLException - for DB failures; rows remain recoverable on retry
      IOException - for queue failure/conflict or unknown outcome

    • consume

      public boolean consume(Connection connection, String id, byte[] envelope, JdbcDelivery.Mutation mutation, Instant consumedAt) throws SQLException, IOException
      Inserts the inbox identity and applies consumer mutation in the caller's single transaction. The database uniqueness constraint serializes concurrent duplicates. After a uniqueness error, roll back and retry the whole transaction so the committed digest can be checked. A returned false means an identical previously committed consumption and performs no application mutation.
      Parameters:
      connection - exact non-autocommit consumer transaction
      id - queue message identity
      envelope - received traced message bytes
      mutation - application mutation using this connection
      consumedAt - audit timestamp, not an expiry authority
      Returns:
      true for a newly applied mutation, false for a verified replay
      Throws:
      SQLException - on conflict, application failure or DB failure
      IOException - on malformed envelope

    • pruneSentPayloads

      public int pruneSentPayloads(Connection connection, Instant now) throws SQLException
      Removes only payloads already confirmed sent and past their retention time. Hash/identity/order tombstones are retained indefinitely so expired payloads cannot permit conflicting ID reuse. Operators must provision/monitor tombstone storage explicitly.
      Parameters:
      connection - caller-managed retention transaction
      now - retention cutoff timestamp
      Returns:
      number of payloads removed
      Throws:
      SQLException - for inactive transaction or database error