Add afterCatchup callback to StreamingSourcingStrategy to execute post-catchup actions

Author: helightdev

krescent-core/src/main/kotlin/dev/helight/krescent/source/strategy/StreamingSourcingStrategy.kt

@@ -20,7 +20,9 @@ import dev.helight.krescent.source.StreamingToken
  * - [SystemHintBeginTransactionEvent] at the start of the catch-up phase.
  * - [SystemHintEndTransactionEvent] at the end of the catch-up phase, after the live stream has started.
  */
-class StreamingSourcingStrategy: EventSourcingStrategy {
+class StreamingSourcingStrategy(
+    val afterCatchup: suspend () -> Unit = {},
+) : EventSourcingStrategy {
     override suspend fun source(
         source: StreamingEventSource,
         startToken: StreamingToken<*>?,
@@ -37,9 +39,9 @@ class StreamingSourcingStrategy: EventSourcingStrategy {
             consumer.forwardSystemEvent(SystemStreamCaughtUpEvent)
         } finally {
             consumer.forwardSystemEvent(SystemHintEndTransactionEvent)
+            afterCatchup()
         }
 
-
         // Begin streaming events until interrupted
         source.streamEvents(lastToken).collect {
             it.forwardTo(consumer)