Summer Data

Summer Data — Audit, Outbox & R2DBC

Gate: Verify summer-core loaded and io.f8a.summer:summer-platform in build.gradle.

Modules: summer-data-autoconfigure | summer-data-audit-autoconfigure | summer-data-outbox-autoconfigure | summer-data-r2dbc | summer-kafka-consumer (0.3.1+) | summer-kafka-consumer-autoconfigure (0.3.1+)

Tracks LATEST stable schema (0.3.5). Older versions: load matching overlay from references/versions/.

summer-kafka-consumer listener wiring (idempotency, retry, DLT) now in summer-kafka. This skill covers producer-side outbox + audit + R2DBC.

AuditService

Auto-activates with R2DBC. Requires audit_log table (see references/ddl-scripts.md). AuditTableValidator validates schema on startup (disable: f8a.audit.validate-on-startup=false).

Primary: audit(AuditLog) builder — auto-fills null fields from context

// Minimal (actor, request info, timestamps auto-filled)
auditService.audit(AuditLog.builder()
    .action("LOGIN").intent("USER_REQUEST").build());

// With entity + before/after payloads
auditService.audit(AuditLog.builder()
    .action("UPDATE").intent("SYSTEM_SYNC")
    .entityType("ExchangeRate").entityId(pair)
    .oldValues(mapper.valueToTree(old))
    .newValues(mapper.valueToTree(updated)).build());

// Override actor (skips security context when actorId set)
auditService.audit(AuditLog.builder()
    .action("CLEANUP").intent("SCHEDULED_JOB")
    .actorId("scheduler").actorUsername("cleanup-job").build());

Convenience methods

auditService.auditCreate(entity, "USER_REQUEST", "Created user");
auditService.auditUpdate(oldEntity, newEntity, "USER_REQUEST", "Updated");
auditService.auditDelete(entity, "USER_REQUEST", "Deleted user");
auditService.auditNonEntity("LOGIN", "USER_REQUEST", "User logged in");

Annotation-based (Mono/Flux return types only)

@Audit defaults:

@Audit(action = "UPDATE", comment = "Updated config")
public Mono<Void> updateConfig(ConfigRequest req) { ... }

@AuditField String name;  // marks field for diff tracking in diffValues

Config

f8a:
  audit:
    validate-on-startup: true

OutboxService (current — 0.3.x canon)

Transactional outbox with two publish modes — CDC (stream WAL via Debezium, preferred) and scheduler (poll table, multi-instance fallback). Both modes share retry+backoff, monitoring, and cleanup.

Recommended default: mode: cdc. core-ledger-ms and payment-orchestrator-ms ship CDC as production default — sub-second latency, no polling load, native ordering by PostgreSQL LSN. Summer internal default is scheduler (safer for first boot without WAL); set mode: cdc once PG prereqs in references/ddl-scripts.md are in place. Use scheduler only for multi-instance deployments (one CDC instance per replication slot) or without WAL access.

// 1. Save in your business transaction (R2DBC).
outboxService.saveEvent("ORDER_CREATED", orderId, payloadJson);

// 2. (Optional) override the built-in publisher.
@Bean
OutboxEventPublisher customPublisher() { ... }

Built-in KafkaOutboxPublisher (0.3.1+)

Auto-wired when KafkaTemplate on classpath and f8a.outbox.publisher.queue=kafka (default). Removes ~50 lines of boilerplate per service. Disable by setting queue to any non-kafka value (or provide own OutboxEventPublisher bean — auto-config backs off).

Recommended CDC-mode config (production default — mirrors core-ledger-ms)

f8a:
  outbox:
    enabled: true                        # default: true; false disables module entirely
    publisher:
      queue: kafka                       # selects KafkaOutboxPublisher (default)
      mode: ${OUTBOX_PUBLISHER_MODE:cdc} # cdc | scheduler — default CDC in this service
      topic-prefix: ${OUTBOX_PUBLISHER_TOPIC_PREFIX:<svc>.}   # prepended to outbound Kafka topic
                                                              # (DIFFERENT from cdc.topic-prefix below)
      cdc:
        # --- Debezium source DB connection (the connector reads PostgreSQL WAL) ---
        connector-name: <svc>-outbox-connector
        url:      ${SPRING_FLYWAY_URL}                       # jdbc:postgresql://host:5432/<db>
        username: ${SPRING_FLYWAY_USERNAME}
        password: ${SPRING_FLYWAY_PASSWORD}
        # --- WAL / replication wiring ---
        slot-name: <svc>_cdc_slot                            # Debezium replication slot (pgoutput)
        plugin-name: pgoutput                                # default; rarely changed
        table-include-list: "<schema>\\.outbox_events"       # regex; e.g. "ledger\\.outbox_events"
        # schema-include-list:                               # optional regex filter
        # --- Kafka storage for Debezium internal state (moved JDBC → Kafka in 0.3.1) ---
        offset-storage-topic:    "<svc>.outbox.offsets"
        offset-storage-partitions: 1
        offset-storage-replication-factor: 3                 # raise to ≥3 for production
        schema-history-topic:    "<svc>.schema.history"
        # --- Source-side server name (Debezium "database.server.name"; default "outbox-server") ---
        # topic-prefix: <svc>                                # optional; default is fine for most setups
        # --- Publish behavior ---
        publish-timeout: 30s
        skip-already-published: true
      scheduler:                         # retained as retry fallback even under mode=cdc
        cron: "${OUTBOX_PUBLISHER_CRON:*/5 * * * * *}"
        batch-size: 100
    retry:
      max-attempts: 5
      initial-interval: 1s
      multiplier: 2.0
      max-interval: 60s
      poll-interval: 10s                 # retry-task cadence
      batch-size: 100
    cleanup:
      cron: "0 0 3 * * ?"                # off-peak
      retention: 14d                     # Duration (was retention-days int pre-0.2.8)
    monitoring:
      cron: "0 0 * * * ?"
    circuit-breaker:
      enabled: true
      failure-rate-threshold: 50
      minimum-events: 10
      wait-duration: 60s                 # Duration (was wait-duration-seconds pre-0.2.8)
      sliding-window-size: 100

Reference: core-ledger-ms/src/main/resources/application.yml ships exactly this shape (mode: ${OUTBOX_PUBLISHER_MODE:cdc}, slot-name: ledger_cdc_slot, table-include-list: "ledger\\.outbox_events"). Copy as starting point for new services owning their own PostgreSQL.

Scheduler-mode variant (multi-instance / no WAL access)

Override at deploy with OUTBOX_PUBLISHER_MODE=scheduler — same YAML, no other changes. scheduler.* block already present; CDC block becomes inert. Use when running more than one app instance per database (one CDC instance per replication slot is hard limit) or in test environments without WAL.

f8a.outbox.publisher.cdc.bootstrap-servers dropped in 0.3.2 — CDC storage uses cluster at spring.kafka.bootstrap-servers. Configure Kafka once at spring.kafka.* (SASL/SSL included); Debezium inherits it since 0.3.2.

CDC prereqs (PostgreSQL) — required before first boot with mode: cdc: wal_level=logical, connector DB user has REPLICATION, publication exists for outbox_events, table has REPLICA IDENTITY DEFAULT (uses PK) or FULL. See references/ddl-scripts.md §"PostgreSQL prereqs for CDC mode" for exact SQL and slot-plugin recovery. One CDC instance per replication slot — fall back to mode: scheduler for multi-instance deployments.

Mode override. mode: ${OUTBOX_PUBLISHER_MODE:cdc} makes CDC the default, any env can downgrade to scheduler via env var. CI/local-dev without make cdc-setup-full sets OUTBOX_PUBLISHER_MODE=scheduler; production keeps default.

publisher.topic-prefix ≠ publisher.cdc.topic-prefix. Publisher prefix prepended to outbound Kafka topic (e.g. ledger. + order.created → ledger.order.created). CDC prefix is Debezium's database.server.name — source identifier for replication-slot offsets and schema history. Independent strings; same value is fine but they serve different purposes.

Schema columns (0.3.x)

outbox_events adds next_retry_at TIMESTAMPTZ (0.2.8) and lsn BIGINT (0.3.1):

ALTER TABLE outbox_events ADD COLUMN IF NOT EXISTS next_retry_at TIMESTAMPTZ;
ALTER TABLE outbox_events ADD COLUMN IF NOT EXISTS lsn BIGINT;

OutboxRetryTask re-emits same ob.lsn Kafka header on retry (CDC-mode only). Pre-0.3.1 events with NULL lsn fall back to epoch-nanos.

Kafka header ob.lsn is binary (0.3.1+)

Header is 8-byte big-endian binary. Consumers bind @Header Long lsn via Spring Messaging's default byte[] → Long converter. Pre-0.3.1 emitted UTF-8 decimal strings → wrong values under @Header Long — recompile against 0.3.1 to fix.

OutboxEventPublisher.publish(event, headers) takes Map<String, byte[]> (was Map<String, String> pre-0.3.1). Custom publishers: drop .getBytes(UTF_8) calls — values already bytes.

Kafka Consumer Idempotency (0.3.1+)

summer-kafka-consumer introduces LSN-watermark-based dedup. Replaces per-event dedup tables with single row per (consumer_group, topic, partition) — matches Kafka's offset-tracking granularity, scales with partition count. Schema: outbox_consumer_watermark (see docs/HOW_WE_IDEMPOTENCY_ON_CONSUMER.md).

@KafkaListener(topics = "orders.created")
Mono<Void> onOrderCreated(OrderCreatedEvent event,
                           @Header Long lsn,
                           ConsumerRecord<?, ?> record) {
    var ctx = consumerCtx.from(record, lsn);
    return idempotency.isProcessed(ctx)
        .flatMap(seen -> seen
            ? Mono.empty()
            : process(event)
                .then(idempotency.recordProcessed(ctx)))
        .as(transactionalOperator::transactional);
}

Config block f8a.kafka.consumer.{idempotency.{enabled, validate-on-startup}, retry.{max-attempts, initial-interval, multiplier, max-interval, dlt-suffix}}.

R2DBC Converters

Auto-configured (SummerR2dbcAutoConfiguration). Registers R2DBC converters for Password, PhoneNumber, Ufid, and (0.3.5+) Txid (see summer-core Shared Types).

Txid converters (0.3.5+) — write side is exclusive

Two converters ship; writer side gated by property:

• TxidConverter — Long ↔ Txid for BIGINT columns. Preferred for new schemas (4× smaller index entries, native int comparison).
• TxidUuidConverter — UUID ↔ Txid for UUID columns. For legacy schemas on UUID PKs.

Reads unambiguous — R2DBC driver yields Long for BIGINT and UUID for UUID; both Reading converters always registered, selected by source type.

Writes exclusive. MappingR2dbcConverter resolves writers by Java type (no SQL-type hint) — registering both leaves choice to registration order and silently writes wrong SQL type. Exactly one writer registered, selected by SummerR2dbcProperties.txidColumnType (@ConfigurationProperties("summer.r2dbc"), enum TxidColumnType — UUID or BIGINT):

summer:
  r2dbc:
    txid-column-type: uuid       # default — registers TxidUuidConverter.Writing only
    # txid-column-type: bigint   # new greenfield schemas — registers TxidConverter.Writing only

Spring binds enum case-insensitively. Value outside UUID / BIGINT → BindException at context start naming the property and offending value.

Default uuid — most eWallet services standardize on UUID PKs (saga ids, account ids, hold ids, va ids). Greenfield schemas wanting BIGINT space-savings opt in by setting the property. Mixing BIGINT and UUID Txid columns in same app: read side OK, write side not — standardize one per app.

@Table("transactions")
public record Transaction(
    @Id Txid txid,             // BIGINT or UUID, matches summer.r2dbc.txid-column-type
    Ufid customerId,           // UUID (canonical key)
    BigDecimal amount,
    Instant createdAt) {}

Source layout:

• io.f8a.summer.data.r2dbc.config.SummerR2dbcProperties — @ConfigurationProperties bean (summer-data-r2dbc).
• io.f8a.summer.data.r2dbc.config.TxidColumnType — the enum.
• SummerR2dbcAutoConfiguration enables via @EnableConfigurationProperties(SummerR2dbcProperties.class), switches on properties.getTxidColumnType() to pick writer.

Version Notes

Headline only — full detail: load matching overlay:

• 0.3.5 — R2DBC TxidConverter / TxidUuidConverter auto-registered; writer gated by summer.r2dbc.txid-column-type (uuid default). →
• 0.3.2 — Debezium Kafka storage inherits SASL/SSL from spring.kafka.*; Swagger BOM aligned. →
• 0.3.1 — Outbox config reshape (BREAKING): scheduler/CDC nested; Debezium storage JDBC→Kafka (BREAKING); KafkaOutboxPublisher auto-wired; headers Map<String, byte[]> (BREAKING); lsn BIGINT preserved on retry; summer-kafka-consumer LSN-watermark idempotency. →
• 0.2.8 — Outbox id UUID→Ufid; CDC mode added; unified retry+backoff; OutboxProperties redesign (BREAKING); next_retry_at column. →
• 0.2.1 — auditNonEntity arg order (BREAKING); audit(AuditLog) builder; auditCustom() deprecated; @Audit Flux fix; double-subscribe fix; AbstractTableValidator; embedded Flyway removed; BusinessChange deleted; validate-schema removed. →

See references/ddl-scripts.md for audit_log, outbox_events, and (0.3.1+) outbox_consumer_watermark DDL.

For full feature × version table see summer-core/references/version-matrix.md.

Rules

• Audit entity changes with AuditService — builder for complex audits, convenience methods for simple CRUD.
• Include intent in audit entries — distinguishes USER_REQUEST from SYSTEM_SYNC/SCHEDULED_JOB.
• Configure OutboxService scheduler and circuit breaker for production — defaults may be too aggressive.
• Never skip audit_log DDL migration — AuditTableValidator fails on startup.

Related Skills

• summer-core — Shared types in audit entries (Member for actor)
• database-patterns — R2DBC repository patterns, Flyway migrations
• messaging-patterns — OutboxService publishes to Kafka/RabbitMQ