{
  "eventId": "evt_01HZX...",
  "eventType": "regulatory.case.accepted.v1",
  "eventVersion": 1,
  "occurredAt": "2026-07-03T09:15:30Z",
  "producer": "case-command-service",
  "correlationId": "corr_...",
  "causationId": "cmd_...",
  "aggregateType": "case",
  "aggregateId": "case_123",
  "partitionKey": "case_123",
  "payload": {}
}

consumer_name + event_id

create schema if not exists integration;

create table integration.consumer_inbox (
    inbox_id              uuid primary key default gen_random_uuid(),

    consumer_name         text not null,
    event_id              text not null,
    event_type            text not null,
    event_version         integer not null,

    aggregate_type        text not null,
    aggregate_id          text not null,

    topic_name            text not null,
    partition_no          integer not null,
    offset_no             bigint not null,

    correlation_id        text null,
    causation_id          text null,
    producer              text not null,

    occurred_at           timestamptz not null,
    received_at           timestamptz not null default now(),

    status                text not null,
    attempt_count         integer not null default 0,
    next_attempt_at       timestamptz null,

    locked_by             text null,
    locked_at             timestamptz null,

    processed_at          timestamptz null,
    quarantined_at        timestamptz null,

    error_code            text null,
    error_message         text null,
    error_detail          jsonb null,

    payload               jsonb not null,
    headers               jsonb not null default '{}'::jsonb,

    created_at            timestamptz not null default now(),
    updated_at            timestamptz not null default now(),

    constraint consumer_inbox_status_ck check (
        status in (
            'RECEIVED',
            'PROCESSING',
            'PROCESSED',
            'RETRY_PENDING',
            'QUARANTINED',
            'IGNORED'
        )
    ),

    constraint consumer_inbox_attempt_ck check (attempt_count >= 0),
    constraint consumer_inbox_event_version_ck check (event_version > 0),

    constraint consumer_inbox_unique_event_per_consumer_uk
        unique (consumer_name, event_id)
);

create index consumer_inbox_retry_idx
on integration.consumer_inbox (consumer_name, status, next_attempt_at, received_at)
where status in ('RECEIVED', 'RETRY_PENDING');

create index consumer_inbox_aggregate_idx
on integration.consumer_inbox (aggregate_type, aggregate_id, received_at desc);

create index consumer_inbox_topic_offset_idx
on integration.consumer_inbox (topic_name, partition_no, offset_no);

create index consumer_inbox_quarantine_idx
on integration.consumer_inbox (consumer_name, quarantined_at desc)
where status = 'QUARANTINED';

for each Kafka record:
    parse envelope
    validate event contract
    begin DB transaction
        inserted = insert inbox row if absent
        if not inserted:
            if existing status is PROCESSED or IGNORED:
                commit DB
                commit Kafka offset
                continue
            if existing status is QUARANTINED:
                commit DB
                do not process automatically
                commit Kafka offset only if policy allows quarantine progress
                continue

        apply domain side effect idempotently
        insert domain audit
        optionally insert outbox events
        mark inbox PROCESSED
    commit DB transaction
    commit Kafka offset

commit Kafka offset
apply DB update

apply DB update
commit DB
commit Kafka offset

for each record:
    parse envelope
    validate envelope minimally
    begin DB transaction
        insert inbox if absent
    commit DB transaction
    commit Kafka offset

loop:
    begin DB transaction
        claim due inbox rows with FOR UPDATE SKIP LOCKED
    commit

    for each row:
        begin DB transaction
            reload row for update
            if status not PROCESSING: skip
            apply side effect
            mark PROCESSED
        commit

        on transient failure:
            mark RETRY_PENDING with next_attempt_at

        on non-retryable failure:
            mark QUARANTINED

with candidate as (
    select inbox_id
    from integration.consumer_inbox
    where consumer_name = #{consumerName}
      and status in ('RECEIVED', 'RETRY_PENDING')
      and (next_attempt_at is null or next_attempt_at <= now())
    order by received_at asc
    limit #{limit}
    for update skip locked
)
update integration.consumer_inbox i
set status = 'PROCESSING',
    locked_by = #{workerId},
    locked_at = now(),
    attempt_count = attempt_count + 1,
    updated_at = now()
from candidate c
where i.inbox_id = c.inbox_id
returning i.*;

update integration.consumer_inbox
set status = 'RETRY_PENDING',
    next_attempt_at = now(),
    locked_by = null,
    locked_at = null,
    error_code = 'STALE_PROCESSING_LOCK',
    error_message = 'Processing lock exceeded timeout',
    updated_at = now()
where status = 'PROCESSING'
  and locked_at < now() - interval '10 minutes';

public interface ConsumerInboxMapper {

    int insertIfAbsent(ConsumerInboxInsertRow row);

    ConsumerInboxRow findByConsumerAndEventId(
        @Param("consumerName") String consumerName,
        @Param("eventId") String eventId
    );

    List<ConsumerInboxRow> claimDue(
        @Param("consumerName") String consumerName,
        @Param("workerId") String workerId,
        @Param("limit") int limit
    );

    int markProcessed(
        @Param("inboxId") UUID inboxId,
        @Param("processedAt") OffsetDateTime processedAt
    );

    int markRetryPending(
        @Param("inboxId") UUID inboxId,
        @Param("errorCode") String errorCode,
        @Param("errorMessage") String errorMessage,
        @Param("errorDetail") String errorDetailJson,
        @Param("nextAttemptAt") OffsetDateTime nextAttemptAt
    );

    int markQuarantined(
        @Param("inboxId") UUID inboxId,
        @Param("errorCode") String errorCode,
        @Param("errorMessage") String errorMessage,
        @Param("errorDetail") String errorDetailJson
    );
}

<mapper namespace="com.acme.integration.inbox.ConsumerInboxMapper">

  <insert id="insertIfAbsent" parameterType="ConsumerInboxInsertRow">
    insert into integration.consumer_inbox (
      consumer_name,
      event_id,
      event_type,
      event_version,
      aggregate_type,
      aggregate_id,
      topic_name,
      partition_no,
      offset_no,
      correlation_id,
      causation_id,
      producer,
      occurred_at,
      status,
      payload,
      headers
    )
    values (
      #{consumerName},
      #{eventId},
      #{eventType},
      #{eventVersion},
      #{aggregateType},
      #{aggregateId},
      #{topicName},
      #{partitionNo},
      #{offsetNo},
      #{correlationId},
      #{causationId},
      #{producer},
      #{occurredAt},
      'RECEIVED',
      #{payload,jdbcType=OTHER,typeHandler=com.acme.mybatis.JsonbTypeHandler},
      #{headers,jdbcType=OTHER,typeHandler=com.acme.mybatis.JsonbTypeHandler}
    )
    on conflict (consumer_name, event_id) do nothing
  </insert>

  <select id="findByConsumerAndEventId" resultMap="ConsumerInboxRowMap">
    select *
    from integration.consumer_inbox
    where consumer_name = #{consumerName}
      and event_id = #{eventId}
  </select>

  <select id="claimDue" resultMap="ConsumerInboxRowMap">
    with candidate as (
      select inbox_id
      from integration.consumer_inbox
      where consumer_name = #{consumerName}
        and status in ('RECEIVED', 'RETRY_PENDING')
        and (next_attempt_at is null or next_attempt_at &lt;= now())
      order by received_at asc
      limit #{limit}
      for update skip locked
    )
    update integration.consumer_inbox i
    set status = 'PROCESSING',
        locked_by = #{workerId},
        locked_at = now(),
        attempt_count = attempt_count + 1,
        updated_at = now()
    from candidate c
    where i.inbox_id = c.inbox_id
    returning i.*
  </select>

  <update id="markProcessed">
    update integration.consumer_inbox
    set status = 'PROCESSED',
        processed_at = #{processedAt},
        locked_by = null,
        locked_at = null,
        error_code = null,
        error_message = null,
        error_detail = null,
        updated_at = now()
    where inbox_id = #{inboxId}
      and status = 'PROCESSING'
  </update>

  <update id="markRetryPending">
    update integration.consumer_inbox
    set status = 'RETRY_PENDING',
        next_attempt_at = #{nextAttemptAt},
        locked_by = null,
        locked_at = null,
        error_code = #{errorCode},
        error_message = #{errorMessage},
        error_detail = #{errorDetail,jdbcType=OTHER,typeHandler=com.acme.mybatis.JsonbTypeHandler},
        updated_at = now()
    where inbox_id = #{inboxId}
      and status = 'PROCESSING'
  </update>

  <update id="markQuarantined">
    update integration.consumer_inbox
    set status = 'QUARANTINED',
        quarantined_at = now(),
        locked_by = null,
        locked_at = null,
        error_code = #{errorCode},
        error_message = #{errorMessage},
        error_detail = #{errorDetail,jdbcType=OTHER,typeHandler=com.acme.mybatis.JsonbTypeHandler},
        updated_at = now()
    where inbox_id = #{inboxId}
      and status = 'PROCESSING'
  </update>

</mapper>

public final class KafkaInboxConsumer implements Runnable {

    private final KafkaConsumer<String, String> consumer;
    private final EventEnvelopeParser parser;
    private final TransactionRunner tx;
    private final ConsumerInboxMapper inboxMapper;
    private final String consumerName;

    private volatile boolean running = true;

    @Override
    public void run() {
        consumer.subscribe(List.of("regulatory.case.events.v1"));

        while (running) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

            for (ConsumerRecord<String, String> record : records) {
                handleRecord(record);
            }

            consumer.commitSync();
        }
    }

    private void handleRecord(ConsumerRecord<String, String> record) {
        EventEnvelope envelope = parser.parse(record.value());

        tx.required(() -> {
            ConsumerInboxInsertRow row = ConsumerInboxInsertRow.from(envelope, record, consumerName);
            inboxMapper.insertIfAbsent(row);
            return null;
        });
    }

    public void shutdown() {
        running = false;
        consumer.wakeup();
    }
}

create table integration.consumer_raw_poison (
    poison_id      uuid primary key default gen_random_uuid(),
    consumer_name  text not null,
    topic_name     text not null,
    partition_no   integer not null,
    offset_no      bigint not null,
    record_key     text null,
    raw_value      text null,
    headers        jsonb not null default '{}'::jsonb,
    error_message  text not null,
    created_at     timestamptz not null default now(),

    constraint consumer_raw_poison_offset_uk
        unique (consumer_name, topic_name, partition_no, offset_no)
);

public final class InboxWorker implements Runnable {

    private final TransactionRunner tx;
    private final ConsumerInboxMapper inboxMapper;
    private final DomainEventDispatcher dispatcher;
    private final RetryPolicy retryPolicy;
    private final String consumerName;
    private final String workerId;
    private final int batchSize;

    private volatile boolean running = true;

    @Override
    public void run() {
        while (running) {
            List<ConsumerInboxRow> rows = tx.required(() ->
                inboxMapper.claimDue(consumerName, workerId, batchSize)
            );

            if (rows.isEmpty()) {
                sleepQuietly(Duration.ofMillis(250));
                continue;
            }

            for (ConsumerInboxRow row : rows) {
                processOne(row);
            }
        }
    }

    private void processOne(ConsumerInboxRow row) {
        try {
            tx.required(() -> {
                ConsumerInboxRow locked = inboxMapper.findByIdForUpdate(row.inboxId());

                if (!locked.isProcessingBy(workerId)) {
                    return null;
                }

                dispatcher.dispatch(locked);
                inboxMapper.markProcessed(locked.inboxId(), OffsetDateTime.now());
                return null;
            });
        } catch (TransientProcessingException ex) {
            tx.requiresNew(() -> {
                OffsetDateTime next = retryPolicy.nextAttempt(row.attemptCount());
                inboxMapper.markRetryPending(row.inboxId(), ex.code(), ex.getMessage(), ex.detailJson(), next);
                return null;
            });
        } catch (NonRetryableProcessingException ex) {
            tx.requiresNew(() -> {
                inboxMapper.markQuarantined(row.inboxId(), ex.code(), ex.getMessage(), ex.detailJson());
                return null;
            });
        }
    }
}

public final class DomainEventDispatcher {

    private final Map<String, DomainEventHandler> handlers;

    public void dispatch(ConsumerInboxRow row) {
        DomainEventHandler handler = handlers.get(row.eventType());

        if (handler == null) {
            throw new NonRetryableProcessingException(
                "UNSUPPORTED_EVENT_TYPE",
                "No handler for event type " + row.eventType()
            );
        }

        handler.handle(row);
    }
}

public final class CaseAcceptedProjectionHandler implements DomainEventHandler {

    private final CaseProjectionMapper projectionMapper;

    @Override
    public void handle(ConsumerInboxRow row) {
        CaseAcceptedPayload payload = row.payloadAs(CaseAcceptedPayload.class);

        int updated = projectionMapper.upsertAcceptedState(
            payload.caseId(),
            payload.acceptedAt(),
            row.eventId()
        );

        if (updated != 1) {
            throw new TransientProcessingException(
                "PROJECTION_UPSERT_FAILED",
                "Expected one projection row to be upserted"
            );
        }
    }
}

insert into read_model.case_summary (
    case_id,
    case_status,
    accepted_at,
    last_event_id,
    updated_at
)
values (
    #{caseId},
    'ACCEPTED',
    #{acceptedAt},
    #{eventId},
    now()
)
on conflict (case_id) do update
set case_status = excluded.case_status,
    accepted_at = excluded.accepted_at,
    last_event_id = excluded.last_event_id,
    updated_at = now()
where read_model.case_summary.last_event_id is distinct from excluded.last_event_id;

create unique index sla_obligation_case_type_uk
on case_core.sla_obligation (case_id, obligation_type)
where cancelled_at is null;

public void handle(ConsumerInboxRow row) {
    CaseAcceptedPayload payload = row.payloadAs(CaseAcceptedPayload.class);

    slaMapper.insertIfAbsent(
        payload.caseId(),
        "INITIAL_ASSESSMENT_DUE",
        payload.acceptedAt().plusDays(5),
        row.eventId()
    );
}

insert into case_core.sla_obligation (
    case_id,
    obligation_type,
    due_at,
    source_event_id,
    status
)
values (
    #{caseId},
    #{obligationType},
    #{dueAt},
    #{sourceEventId},
    'OPEN'
)
on conflict do nothing;

public final class RetryPolicy {

    public OffsetDateTime nextAttempt(int attemptCount) {
        long seconds = switch (attemptCount) {
            case 0, 1 -> 5;
            case 2 -> 30;
            case 3 -> 120;
            case 4 -> 600;
            default -> 1800;
        };

        long jitter = ThreadLocalRandom.current().nextLong(0, Math.min(60, seconds));
        return OffsetDateTime.now().plusSeconds(seconds + jitter);
    }

    public boolean shouldQuarantine(int attemptCount) {
        return attemptCount >= 10;
    }
}

select pg_advisory_xact_lock(hashtext(#{aggregateType} || ':' || #{aggregateId}));

{
  "aggregateId": "case_123",
  "aggregateVersion": 17
}

alter table read_model.case_summary
add column last_aggregate_version bigint not null default 0;

update read_model.case_summary
set case_status = #{newStatus},
    last_aggregate_version = #{aggregateVersion},
    updated_at = now()
where case_id = #{caseId}
  and last_aggregate_version = #{aggregateVersion} - 1;

SIGTERM received
-> readiness = false
-> stop accepting new work
-> stop poll loop
-> finish current record/batch
-> commit offsets
-> close consumer
-> exit before terminationGracePeriodSeconds

{
  "event": "consumer_inbox_processed",
  "consumerName": "case-camunda-correlation-consumer",
  "eventId": "evt_01HZX...",
  "eventType": "regulatory.case.accepted.v1",
  "aggregateId": "case_123",
  "topic": "regulatory.case.events.v1",
  "partition": 4,
  "offset": 120932,
  "correlationId": "corr_...",
  "attemptCount": 1,
  "durationMs": 37
}

select consumer_name, event_type, count(*)
from integration.consumer_inbox
where received_at > now() - interval '1 hour'
group by consumer_name, event_type
order by count(*) desc;

select consumer_name,
       event_type,
       error_code,
       count(*) as total,
       min(quarantined_at) as first_seen,
       max(quarantined_at) as last_seen
from integration.consumer_inbox
where status = 'QUARANTINED'
group by consumer_name, event_type, error_code
order by total desc;

update integration.consumer_inbox
set status = 'RETRY_PENDING',
    next_attempt_at = now(),
    quarantined_at = null,
    error_code = null,
    error_message = null,
    error_detail = null,
    updated_at = now()
where inbox_id = #{inboxId}
  and status = 'QUARANTINED';