Implement indexing logic using Domain Event

Using NXP-32507 indexing domain event, create computations that perform the indexing.

There are 2 consumers for this stream:

• A foreground indexer that takes care only of the synchronous part of the commands, it takes care of refreshing the index.
• A background indexer that is skipping all synchronous commands and processes the rest.

In addition to these two computations, we have the index bulk action.

All commands that concern children or direct children result in submitting bulk commands.

Waiting for foreground indexing

We want to be able to wait for an indexing domain event to be processed in order to reproduce the sync indexing that we have today and that is essential for UI navigation.

It’s possible to wait for a consumer to process an offset using LogAppender#waitFor

The idea is to be able to wait for consumers on the produced records at the transaction level. This could be done with TransactionHelper#onCompletionWaitFor(stream, consumer, timeout) that can be activated when creating the Indexing DEP if the context is coming from the UI.

Note that the LogAppender#waitFor needs to be checked under load (with high concurrency) because it creates a temporary KafkaConsumer even if it doesn’t join the group (no poll or rebalance in play) there is a roundtrip with Kafka brokers on an API that could be bottleneck. In this case we could mark a computation to report progress and publish the checkpointed offset using the PubSub service, this will also rely on Kafka roundtrip but in a scalable way.

Background indexing

We could imagine that a Rest API that trigger some indexing could return a sequence number that would be just the offset of the indexing DEP.
This way we could provide an API to wait or know when the indexing of a particular REST action is completed.

Waiting for the foreground part will be just using LogAppender#waitFor. We could do the same for the background api in this case it will only assert for the light indexing being done not the related bulk command.

We could use the KV to store the list of bulk commands matching a record to be able to wait for completion of async indexing if needed.