Event State Transfer – Base Listener Framework (wallet-service)

Branch: CM-11233_outbox_listener_base (local, uncommitted) · Base: origin/master @ afa5dc37d2

1. Why this exists

Card issuance (PAE / NAS) needs its own copy of the wallet-service state it uses for auth decisions (wallet lifecycle, limits, block rules, fees). It gets that copy from state events. wallet-service publishes a full snapshot event whenever that state changes.

Each event must be written atomically with the business change. If the DB commit succeeds but the publish fails, the event is lost. The transactional outbox handles this: the event row is inserted into outbox_events in the same DB transaction, and a separate drainer sends it later.

Before this change, every event needed its own listener and its own tracker (~150 lines of copied plumbing per event). This branch adds a shared base. A new event is now one handler class.

2. The flow at a glance

[Diagram]

In one sentence: Hibernate reports every flushed change, the tracker keeps only the ones some event cares about, and just before commit each event builds its snapshots and writes them to the outbox in the same transaction.

3. Components

[Diagram]

Class	Responsibility
`StateEventHibernateListener`	One Hibernate `PostInsert`/`PostUpdate`/`PostDelete` listener for every entity, registered once at startup. No entity needs `@EntityListeners`. For updates it passes the names of the changed columns.
`StateChangeTracker`	Matches each change to the handlers whose triggers fit, computes each handler's key, collects the keys per transaction (deduplicated), and calls `handler.publish(keys)` once per handler just before commit. Also the manual entry point for bulk/native SQL writes.
`StateEventHandler<K>`	Base class every event extends: triggers in the constructor, `publish(keys)` to build events and write them to the outbox.
`Triggers<K>`	Immutable `record` (built with `Triggers.builder()`) declaring which entity, which change types, optionally which columns, and how an entity maps to the event key. Each entry is a `Triggers.Trigger` record.
`ChangeType`	`INSERT`, `UPDATE`, `DELETE`.
`TableChangeRecorder`	For writers that only know a table name (the support API): maps table → entity and columns → properties through Hibernate's model, asks the tracker if an enabled handler reacts, and loads/refreshes/records the affected rows.
`WalletEventOutboxConnectorConfig`	Faizan's EventBridge connector for the drainer. Only active when `outbox.drainer.enabled=true`.

3.1 Class-by-class reference (what each one does, with an example)

The base lives in com.nium.spend.cards.wallet.eventtransfer. In call order:

[Diagram]

`ChangeType` (enum)

What it is: the three kinds of row change: INSERT, UPDATE, DELETE. Triggers filter on them, and every recorded change carries one.

.on(WalletBalance.class, b -> b.getWallet().getId(), ChangeType.INSERT, ChangeType.DELETE) // currency added/removed only
tracker.recordChange(wallet, ChangeType.UPDATE);                                            // manual record after bulk SQL

`Triggers<K>` (record) + `Triggers.Trigger` (record) + `Triggers.Builder`

What it is: the declaration of what an event listens to. It's an immutable record holding a list of Trigger entries. Each entry says:

Field of `Trigger`	Meaning	Example
`entityType`	Which entity class	`WalletBalance.class`
`keyFunction`	How a changed row maps to the event's key (the entity the event describes)	`b -> b.getWallet().getId()` (a balance change concerns its wallet)
`changeTypes`	Which change kinds count	`{INSERT, DELETE}`
`updatedColumns`	For UPDATE only: count it only if one of these entity properties changed (empty = any column)	`{"status", "name"}`

It's built with the builder, never by hand:

Triggers.<Long>builder()
        .on(Wallet.class, Wallet::getId, ChangeType.INSERT, ChangeType.DELETE)   // wallet created / deleted
        .onUpdateOf(Wallet.class, Wallet::getId, "status", "name")              // wallet updated, only these columns
        .on(WalletBalance.class, b -> b.getWallet().getId(), ChangeType.INSERT, ChangeType.DELETE)
        .build();

Example: what this declaration decides

Change	Matches?	Key
`INSERT wallet` id 42	yes	`42`
`UPDATE wallet SET status` (id 42)	yes	`42`
`UPDATE wallet SET comments` (id 42)	no, `comments` not in the list
`INSERT wallet_balance` (wallet 42)	yes	`42`
`UPDATE wallet_balance SET balance` (money movement)	no, UPDATE is not a listed change type
`UPDATE currency ...`	no, entity not listed

Safety checks: build() with no entries throws IllegalStateException; onUpdateOf(...) with no columns throws IllegalArgumentException. The record makes defensive copies, so a declared trigger cannot be changed afterwards.

`StateEventHandler<K>` (abstract class)

What it is: the base class of every event, and the only class an event author extends. It has two parts:

Constructor: pass the Triggers<K> (what to listen to).
publish(Set<K> keys): called once per transaction, inside that transaction, after Hibernate's final flush and just before commit, with every key that changed. Load the current state for those keys, build one full-snapshot event per key, and append them to the outbox. Throwing rolls the business change back.

K is the key type, e.g. Long for a wallet ID or String for clientId:currencyCode. It is also the natural outbox aggregate key.

@Component
@ConditionalOnProperty(name = "event-transfer.wallet-lifecycle.enabled", havingValue = "true")
class WalletLifecycleEventHandler extends StateEventHandler<Long> {
    WalletLifecycleEventHandler(...) {
        super(Triggers.<Long>builder().on(Wallet.class, Wallet::getId).build());
    }
    @Override
    protected void publish(Set<Long> walletIds) {   // e.g. {42, 57}
        // load wallets 42 and 57 as they are being committed → 2 events → outbox.appendAll(...)
    }
}

The handler is a normal Spring bean, so @ConditionalOnProperty is its feature flag: no bean means no tracking at all.

`StateEventHibernateListener`

What it is: one Hibernate listener for every entity, registered once at startup (@PostConstruct → Hibernate's EventListenerRegistry). It replaces per-entity @EntityListeners.

onPostInsert → tracker.onFlushedChange(entity, INSERT, ...)
onPostUpdate → the same with UPDATE, plus the names of the properties that changed (Hibernate's dirty properties), which is what makes onUpdateOf(...) work
onPostDelete → the same with DELETE

Example: walletRepository.save(wallet) with only status changed → at flush Hibernate fires PostUpdate → the listener forwards (wallet, UPDATE, {"status"}, session). It decides nothing itself; the tracker does.

`StateChangeTracker`

What it is: the core. It matches changes to handlers, collects keys per transaction, and calls the handlers before commit. It's Faizan's WalletChangeTracker approach (per-session pending set plus a Hibernate before-completion hook), generalized to any entity and any number of handlers.

Public methods:

Method	Used by	What it does
`isTracking(Class)`	bulk-update call sites	"Does any enabled handler listen to this entity?" Lets callers skip extra queries
`isAffectedBy(Class, ChangeType, Set<String> properties)`	`TableChangeRecorder`	Like `isTracking`, but for a specific change type and columns
`recordChange(entity, ChangeType)`	bulk JPQL / native SQL call sites	Records a change Hibernate couldn't see; for UPDATE the changed columns count as unknown (matches column-filtered triggers too)
`recordChange(entity, ChangeType, Set<String> properties)`	`TableChangeRecorder`	The same, when the changed columns are known

Inside, for each change:

Routes: on first use it reads every handler's Triggers once and caches "entity class → (handler, trigger)" routes. With no handlers, every change ends here after one lookup.
Match: change type and (for UPDATE) columns are checked against each route's trigger.
Key: the trigger's keyFunction runs immediately, so the key reflects the entity at that moment (important for a row that is later moved or deleted).
Pending: the key is added to Map<handler, Set<key>>, bound to the current Hibernate session. The first match in a transaction registers the before-completion hook.
Before commit: the hook calls handler.publish(keys) once per handler. If a handler throws, the transaction rolls back. If tracked entities changed while publishing, it fails on purpose rather than losing them.

Example: one transaction, traced

Step	What happens	Pending after the step
1	`save(new Wallet)` → flush → INSERT wallet 42	`{Lifecycle: [42]}`
2	`save(balance USD of wallet 42)` → INSERT	`{Lifecycle: [42]}` (deduplicated)
3	`save(balance EUR of wallet 42)` → INSERT	`{Lifecycle: [42]}`
4	`wallet.setComments(...)` → UPDATE `comments`	unchanged (column not watched)
5	commit → hook → `Lifecycle.publish([42])`	one event for wallet 42, with USD and EUR

Isolation: pending keys belong to one Hibernate session, so a nested REQUIRES_NEW transaction keeps and publishes its own keys.

`TableChangeRecorder` (+ `TrackedTable` record)

What it is: the bridge for writers that only know a table name and run raw SQL. Today that's the ops support-table API. It uses Hibernate's own mapping to translate table to entity and columns to properties, so no table names are hard-coded.

Method	What it does	Example
`trackedTable(schema, table, changeType, columns)`	Returns a `TrackedTable` if an enabled handler reacts, otherwise empty	`("wallet_service", "wallet_balance", UPDATE, ["wallet_id"])` → `TrackedTable(WalletBalance, idColumn "id", Long, UPDATE, {"wallet"})`
`load(trackedTable, ids)`	Loads rows as managed entities; skips IDs with no row	`load(t, [88])` → `[WalletBalance#88]`
`record(trackedTable, entity)`	Calls `tracker.recordChange(entity, type, properties)`	the key is computed now
`refresh(entity)`	Re-reads the row after raw SQL changed it	after `UPDATE ... SET wallet_id = 12`
`detach(entity)`	Drops a raw-deleted row from the session	after `DELETE`

TrackedTable carries entityType, idColumn, idType, changeType, changedProperties.

Example: ops moves balance 88 from wallet 11 to wallet 12

trackedTable(..., "wallet_balance", UPDATE, ["wallet_id"])  → tracked (handler watches property "wallet")
load(ids [88]) → balance#88 (wallet 11) → record → key 11
UPDATE wallet_balance SET wallet_id = 12 WHERE id = 88
refresh(balance#88) (wallet 12) → record → key 12
commit → Lifecycle.publish([11, 12])   (both wallets' currency lists changed)

`WalletEventOutboxConnectorConfig` (Faizan's, unchanged)

What it is: the drainer's destination, an EventBridge OutboxConnector that sends committed outbox rows on. It only loads when outbox.drainer.enabled=true, which is false until the transport decision (SNS FIFO vs EventBridge) is made. Until then, rows wait in outbox_events with status NEW.

4. Runtime sequence (one transaction)

[Diagram]

What happens in each outcome:

[Diagram]

Repeated saves of one wallet in one transaction → one key → one event with the final state.
Business failure / rollback → nothing is published.
Handler failure (e.g. outbox insert fails) → the business change rolls back too. No change can be committed without its event.
Nested REQUIRES_NEW transaction → keeps its own pending set and publishes on its own commit.
No handler registered → a single cached lookup per entity flush. No hook is registered and nothing else happens.

Writes Hibernate cannot see (bulk JPQL / native SQL)

[Diagram]

The branch wires two such paths (both from Faizan's commits):

Path	What it does
`WalletServiceImpl.walletStatusUpdate`	Bulk `UPDATE Wallet ... WHERE customerId` → loads the customer's wallets → `recordChange(wallet, UPDATE)`
`WalletSupportOperationsServiceImpl.logTableMutationRequest`	Ops' raw SQL fixes on any table → generic path through `TableChangeRecorder` (see below), for every table an enabled event tracks

Ops support-table API (dynamic SQL, any table)

The support API builds SQL at runtime for any table (allowed-tables=all). TableChangeRecorder makes it generic: there is no code specific to wallets, and a table counts only while an enabled event tracks it.

[Diagram]

Case	Result
Table/column no enabled event watches (e.g. `SET comments = ...` when events watch `status`)	SQL runs exactly as before, no lookups
`UPDATE wallet SET status = 'Blocked'`	wallet recorded before + after → one event (same key)
`UPDATE wallet_balance SET wallet_id = 12` (balance moved)	before = old wallet, after = new wallet → both publish
`DELETE` of a tracked row	recorded before the row disappears → parent's event publishes
`INSERT` into a tracked table	new id via `RETURNING`, row loaded and recorded
Row can't be resolved	SQL still runs; the response tells ops to re-publish through the event's bootstrap API
Dry run (`validateTableMutationRequest`)	untouched: it rolls back and records nothing

A future event (auth limits, block rules, fees) is covered by the support API automatically: its handler's Triggers are all it needs.

5. What changed on the branch

Picked from Faizan's commits (`911e0d3762`, `44c377986a`, `b8010fd8b0`, `345802a248`)

Area	Change	Status
`pom.xml`	`outbox-postgres-spring-boot-starter` + `outbox-aws` 1.1.1	kept as-is
`liquibase_changelog.xml`	`outbox_table` property + library changelog include (creates `outbox_events`, `outbox_events_dead_letter`)	kept as-is (placed after master's `v17/25`, `v17/26`)
`application.properties`	`outbox.` (drainer off), `event-transfer.eventbridge.`	kept; removed the `event-transfer.wallet-lifecycle.enabled` flag (belongs to the event PR)
`WalletEventOutboxConnectorConfig`	EventBridge connector for the drainer	kept, byte-identical
`WalletServiceImpl.walletStatusUpdate`	`@Transactional` + record wallets after the bulk update	kept; now guarded by `isTracking(Wallet.class)` so it adds no query when no event exists
`WalletSupportOperationsServiceImpl`	ops support-table tracking	made generic: Faizan's wallet-only lookups replaced by `TableChangeRecorder`, so it covers every table an enabled event tracks

Replaced by the generic base

Faizan's class (wallet-only)	Replaced by
`WalletEntityListener` + `@EntityListeners` on `Wallet` / `WalletBalance`	`StateEventHibernateListener` (global) + the handler's `Triggers`
`WalletChangeTracker` (hard-wired to `WalletLifecycleEventPublisher`)	`StateChangeTracker`: same core (per-session pending set + Hibernate before-completion hook), keyed by handler instead of wallet ID
`WalletChangeTrackerJpaTest`	`StateChangeTrackerJpaTest` (11 tests on real Hibernate) + `StateChangeTrackerTest`

Left out (belongs to the WalletLifecycleEvent PR, not the base)

WalletLifecycleEventFactory, WalletLifecycleEventPublisher, model/WalletLifecycleEvent, WalletLifecycleBootstrapService, WalletLifecycleEventController, the lifecycle schema JSON and their tests, the CM-11549 hash-column commit, the whole CustomerFeeStateEvent.

6. Example – adding Faizan's WalletLifecycleEvent on top of the base

Before vs after

[Diagram]

Step 1 – the handler (the only new class)

@Component
@ConditionalOnProperty(name = "event-transfer.wallet-lifecycle.enabled", havingValue = "true")
public class WalletLifecycleEventHandler extends StateEventHandler<Long> {

    private final WalletRepository walletRepository;
    private final WalletLifecycleEventPublisher walletLifecycleEventPublisher; // Faizan's class, unchanged

    public WalletLifecycleEventHandler(final WalletRepository walletRepository,
                                       final WalletLifecycleEventPublisher walletLifecycleEventPublisher) {
        super(Triggers.<Long>builder()
                // any insert / update / delete of a wallet
                .on(Wallet.class, Wallet::getId)
                // a currency added to / removed from the wallet; balance UPDATES (money movements) are NOT triggers
                .on(WalletBalance.class, balance -> balance.getWallet().getId(), ChangeType.INSERT, ChangeType.DELETE)
                .build());
        this.walletRepository = walletRepository;
        this.walletLifecycleEventPublisher = walletLifecycleEventPublisher;
    }

    @Override
    protected void publish(final Set<Long> walletIds) {
        // Runs inside the committing transaction, after the final flush: these are the rows being committed.
        walletLifecycleEventPublisher.publishAll(walletRepository.findAllById(walletIds));
    }
}

The triggers are the exact rules of Faizan's WalletEntityListener: Wallet persist/update/remove, WalletBalance persist/remove only.

Step 2 – bring Faizan's event classes over unchanged

WalletLifecycleEventFactory, WalletLifecycleEventPublisher, model/WalletLifecycleEvent, the schema JSON, WalletLifecycleBootstrapService, WalletLifecycleEventController.

Step 3 – flags

# application.properties
event-transfer.wallet-lifecycle.enabled=true

# config/application-test.properties  (H2 has no outbox table)
event-transfer.wallet-lifecycle.enabled=false

Step 4 – bulk / native writes

Nothing to add. The two wallet paths (walletStatusUpdate, support-table API) already call recordChange(wallet, UPDATE) in the base, and they activate automatically once a handler tracks Wallet.

Step 5 – tests

Faizan's WalletLifecycleEventFactoryTest / WalletLifecycleEventSchemaTest as-is, plus a @DataJpaTest that imports StateChangeTracker, StateEventHibernateListener and the handler with a mocked Outbox (same pattern as StateChangeTrackerJpaTest).

What it looks like at runtime

[Diagram]

7. How to add a new event – full walkthrough

Example: WalletAuthLimitStateEvent, the client- and currency-level spend limits in wallet_auth_limits (entity WalletAuthLimits). Its key is clientId:currencyCode, which is also the SQS MessageGroupId in the contract doc.

[Diagram]

Files you create, all in eventtransfer/ (except the schema):

File	Purpose
`model/WalletAuthLimitStateEvent.java`	The payload (full snapshot of one row)
`resources/schema/async/public-wallet-service-wallet-auth-limit-state-event-value.json`	The contract, copied from the doc
`WalletAuthLimitEventHandler.java`	Triggers + `publish()`: the only class that touches the base
`WalletAuthLimitBootstrapService.java` + controller (optional)	One-time backfill per environment

Not needed: listener class, @EntityListeners, tracker copy, support-API changes.

Step 1 – the event model

@Data
public class WalletAuthLimitStateEvent {
    public static final String OUTBOX_EVENT_TYPE = "WalletAuthLimitStateEvent.v1";

    private String schemaVersion = "1.0";
    private String eventId;
    private String updatedAt;
    private long clientId;
    private String currencyCode;
    private BigDecimal dailyLimit;
    private long dailyCount;
    private BigDecimal monthlyLimit;
    private long monthlyCount;
    private BigDecimal perTransactionLimit;
    // ...remaining fields exactly as the contract doc (section 10.5) lists them
}

Step 2 – the schema

Copy the JSON schema from the contract doc into resources/schema/async/, and add a schema test like Faizan's WalletLifecycleEventSchemaTest.

Step 3 – the handler (the only class that uses the base)

@Slf4j
@Component
@ConditionalOnProperty(name = "event-transfer.wallet-auth-limit.enabled", havingValue = "true")
public class WalletAuthLimitEventHandler extends StateEventHandler<String> {

    private final WalletAuthLimitsRepository walletAuthLimitsRepository;
    private final Outbox outbox;

    public WalletAuthLimitEventHandler(final WalletAuthLimitsRepository walletAuthLimitsRepository, final Outbox outbox) {
        super(Triggers.<String>builder()
                // any insert / update / delete of a limits row, keyed like the SQS MessageGroupId
                .on(WalletAuthLimits.class, limit -> limit.getClientId() + ":" + limit.getCurrencyCode())
                .build());
        this.walletAuthLimitsRepository = walletAuthLimitsRepository;
        this.outbox = outbox;
    }

    @Override
    protected void publish(final Set<String> keys) {          // e.g. {"1001:USD", "1001:EUR"}
        final List<PendingEvent> events = new ArrayList<>();
        for (final String key : keys) {
            final long clientId = Long.parseLong(StringUtils.substringBefore(key, ":"));
            final String currencyCode = StringUtils.substringAfter(key, ":");
            final WalletAuthLimits row = walletAuthLimitsRepository.getByClientIdAndCurrencyCode(clientId, currencyCode);
            if (row == null) {
                log.warn("WalletAuthLimits {} no longer exists; delete handling is an open contract question", key);
                continue;
            }
            final UUID eventId = UUID.randomUUID();
            events.add(new PendingEvent(eventId, key, toEvent(row, eventId)));
        }
        if (!events.isEmpty()) {
            // Same outbox call as Faizan's WalletLifecycleEventPublisher; runs in the committing transaction.
            outbox.appendAll(events, PendingEvent::eventId, PendingEvent::key, event -> EventMode.REALTIME,
                    event -> null, event -> WalletAuthLimitStateEvent.OUTBOX_EVENT_TYPE, PendingEvent::payload);
        }
    }

    private static WalletAuthLimitStateEvent toEvent(final WalletAuthLimits row, final UUID eventId) {
        final WalletAuthLimitStateEvent event = new WalletAuthLimitStateEvent();
        event.setEventId(eventId.toString());
        event.setUpdatedAt(Instant.now().toString());
        event.setClientId(row.getClientId());
        event.setCurrencyCode(row.getCurrencyCode());
        event.setDailyLimit(row.getDailyLimit());
        event.setDailyCount(row.getDailyCount());
        event.setMonthlyLimit(row.getMonthlyLimit());
        event.setMonthlyCount(row.getMonthlyCount());
        event.setPerTransactionLimit(row.getPerTransactionLimit());
        return event;
    }

    private record PendingEvent(UUID eventId, String key, WalletAuthLimitStateEvent payload) { }
}

If only some columns matter to the event, use onUpdateOf(...) for updates instead of a bare .on(...):

.on(WalletAuthLimits.class, key, ChangeType.INSERT, ChangeType.DELETE)
.onUpdateOf(WalletAuthLimits.class, key, "dailyLimit", "dailyCount", "monthlyLimit", "monthlyCount", "perTransactionLimit")

Step 4 – flags

# application.properties (per environment)
event-transfer.wallet-auth-limit.enabled=true

# config/application-test.properties (H2 has no outbox table)
event-transfer.wallet-auth-limit.enabled=false

Step 5 – writes Hibernate can't see

Bulk JPQL / native SQL on wallet_auth_limits: none exist today (checked). If someone adds one later, they call tracker.recordChange(row, ChangeType.UPDATE) for the affected rows.
Ops support-table API: nothing to do. TableChangeRecorder maps wallet_auth_limits → WalletAuthLimits and records the row automatically once this handler is enabled.

Step 6 – bootstrap (one-time backfill)

Publish one SNAPSHOT event per existing row with outbox.appendSnapshot(...) / appendAll(...) and EventMode.SNAPSHOT, following Faizan's WalletLifecycleBootstrapService. Run it once per environment before consumers rely on the stream.

Step 7 – tests

@DataJpaTest
@ActiveProfiles("test")
@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
@TestPropertySource(properties = "event-transfer.wallet-auth-limit.enabled=true")
@Import({StateChangeTracker.class, StateEventHibernateListener.class, WalletAuthLimitEventHandler.class})
@Transactional(propagation = Propagation.NOT_SUPPORTED)
class WalletAuthLimitEventHandlerJpaTest {

    @MockitoBean private Outbox outbox;
    @Autowired private WalletAuthLimitsRepository repository;
    @Autowired private TransactionTemplate transactionTemplate;

    @Test
    void changingALimit_appendsOneEventKeyedByClientAndCurrency() {
        final WalletAuthLimits row = repository.save(limits(1001L, "USD"));
        reset(outbox);

        transactionTemplate.executeWithoutResult(status -> {
            final WalletAuthLimits loaded = repository.findById(row.getId()).orElseThrow();
            loaded.setDailyLimit(new BigDecimal("500.00"));
        });

        // capture outbox.appendAll(...) and assert: one item, key "1001:USD", dailyLimit 500.00
    }
}

Also add a model/factory unit test and the schema test.

What happens at runtime

[Diagram]

Checklist

StateEventHandler subclass with Triggers + publish(), guarded by @ConditionalOnProperty
Event model + JSON schema (from the contract doc)
Flag true per environment, false in config/application-test.properties
Bulk JPQL / native writes to the table → tracker.recordChange(...) (support API is automatic)
Bootstrap endpoint for the one-time backfill
Tests: model/factory, schema, @DataJpaTest with the handler and a mocked Outbox

Trigger cheat sheet:

Need	Trigger
Any change of an entity	`.on(Entity.class, keyFn)`
Only some change types	`.on(Entity.class, keyFn, ChangeType.INSERT, ChangeType.DELETE)`
Only when specific columns change	`.onUpdateOf(Entity.class, keyFn, "status", "name")`
Child row changes the parent's event	`.on(Child.class, child -> child.getParent().getId())`
Child row moved to another parent	`.onUpdateOf(Child.class, child -> child.getParent().getId(), "parent")`
Several tables feed one event	chain several `.on(...)` calls with the same key type

8. Rules every event author must know

Write events with the outbox (or plain JDBC) inside publish(). It runs after Hibernate's last flush, so JPA save() calls made there are never written.
Do not change tracked entities inside publish(). The tracker fails the transaction on purpose rather than silently losing those changes.
A handler exception rolls back the business change. This is the atomicity guarantee.
Bulk / native SQL bypasses Hibernate → call recordChange yourself.
Turn the event flag off in the test profile (no outbox table on H2).
Keys are deduplicated per transaction, so publish() gets each key once, with the final state.

9. Verification

Check	Result
Unit suite (`mvn test`, incl. PMD + Checkstyle)	5251 tests, 0 failures, 61 skipped
`StateChangeTrackerJpaTest` (real Hibernate)	11/11: routing, balance movements ignored, add/remove currency, dedup, column filter, rollback, handler failure rolls back, `REQUIRES_NEW` isolation, manual record, outside-transaction no-op
`TableChangeRecorderJpaTest` (real Hibernate)	6/6: table/column mapping incl. case and schema, untracked tables/columns ignored, raw balance move publishes both wallets, raw status update publishes once, raw delete publishes the parent, unknown IDs skipped
`WalletSupportOperationsServiceImplTest`	24/24: untracked tables skipped, UPDATE recorded before + after, DELETE recorded then detached, INSERT by returned ID, unresolved rows still execute, dry run untouched
Full application context (`ApplicationIT`)	starts with the global listener, the outbox starter and `TableChangeRecorder`
All integration tests (1164)	4 PDF failures, caused by a stale font in local `target/`. They pass after `mvn clean` and are not related to this change

10. Open items

Transport not decided. The doc recommends SNS FIFO, and the EventBridge connector cannot give per-entity ordering. The drainer stays off until this is decided.
Outbox tables are created wherever the DB-change job runs; needs DBA sign-off.
OutboxBacklogMetrics warning: the starter's metrics poller logs a WARN wherever outbox_events doesn't exist yet (H2, or an environment before Liquibase). The library has no off switch, so it's harmless but noisy.
Stale ops message: Faizan's ops message points to a "wallet lifecycle event bootstrap api", which arrives with the event PR.
Merge master first: origin/master is 7 commits ahead of the branch base. They don't overlap with this change, but merge before raising the PR.

Event State Transfer – Base Listener Framework (wallet-service)

1. Why this exists

2. The flow at a glance

3. Components

3.1 Class-by-class reference (what each one does, with an example)

ChangeType (enum)

Triggers<K> (record) + Triggers.Trigger (record) + Triggers.Builder

StateEventHandler<K> (abstract class)

StateEventHibernateListener

StateChangeTracker

TableChangeRecorder (+ TrackedTable record)

WalletEventOutboxConnectorConfig (Faizan's, unchanged)

4. Runtime sequence (one transaction)

Writes Hibernate cannot see (bulk JPQL / native SQL)

Ops support-table API (dynamic SQL, any table)

5. What changed on the branch

Picked from Faizan's commits (911e0d3762, 44c377986a, b8010fd8b0, 345802a248)

Replaced by the generic base

Left out (belongs to the WalletLifecycleEvent PR, not the base)

6. Example – adding Faizan's WalletLifecycleEvent on top of the base

Before vs after

Step 1 – the handler (the only new class)

Step 2 – bring Faizan's event classes over unchanged

Step 3 – flags

Step 4 – bulk / native writes

Step 5 – tests

What it looks like at runtime

7. How to add a new event – full walkthrough

Step 1 – the event model

Step 2 – the schema

Step 3 – the handler (the only class that uses the base)

Step 4 – flags

Step 5 – writes Hibernate can't see

Step 6 – bootstrap (one-time backfill)

Step 7 – tests

What happens at runtime

Checklist

8. Rules every event author must know

9. Verification

10. Open items

`ChangeType` (enum)

`Triggers<K>` (record) + `Triggers.Trigger` (record) + `Triggers.Builder`

`StateEventHandler<K>` (abstract class)

`StateEventHibernateListener`

`StateChangeTracker`

`TableChangeRecorder` (+ `TrackedTable` record)

`WalletEventOutboxConnectorConfig` (Faizan's, unchanged)

Picked from Faizan's commits (`911e0d3762`, `44c377986a`, `b8010fd8b0`, `345802a248`)