feat(eventsourcing): move aggregate id scanning to EventStore

documentation/docs/en/guide/eventstore.md

@@ -23,10 +23,10 @@ In traditional architectures, databases only store the current state, and histor
 
 ## Core Interface
 
-The `EventStore` interface defines the core operations for event storage:
+The `EventStore` interface defines the core operations for event storage and owns paginated aggregate ID scanning by named aggregate:
 
 ```kotlin
-interface EventStore {
+interface EventStore : AggregateIdScanner {
     fun append(eventStream: DomainEventStream): Mono<Void>
     fun load(
         aggregateId: AggregateId,
@@ -38,6 +38,12 @@ interface EventStore {
         headEventTime: Long,
         tailEventTime: Long
     ): Flux<DomainEventStream>
+    fun last(aggregateId: AggregateId): Mono<DomainEventStream>
+    fun scanAggregateId(
+        namedAggregate: NamedAggregate,
+        afterId: String = AggregateIdScanner.FIRST_ID,
+        limit: Int = 10
+    ): Flux<AggregateId>
 }
 ```
 
@@ -63,7 +69,7 @@ Key characteristics:
 |---|---|---|
 | `DomainEvent` | Immutable fact about a past business action within an aggregate | [DomainEvent.kt:52-95](https://github.com/Ahoo-Wang/Wow/blob/main/wow-api/src/main/kotlin/me/ahoo/wow/api/event/DomainEvent.kt#L52-L95) |
 | `DomainEventStream` | Ordered batch of domain events produced by a single command | [DomainEventStream.kt:51-125](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/event/DomainEventStream.kt#L51-L125) |
-| `EventStore` | Core interface for appending and loading event streams | [EventStore.kt:27-98](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/eventsourcing/EventStore.kt#L27-L98) |
+| `EventStore` | Core interface for appending, loading event streams, and scanning aggregate IDs | [EventStore.kt](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/eventsourcing/EventStore.kt) |
 | `SnapshotStore` | Optimizes aggregate loading with versioned state checkpoints | [SnapshotStore.kt:27-58](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/eventsourcing/snapshot/SnapshotStore.kt#L27-L58) |
 
 ## Aggregate State Reconstruction
@@ -136,6 +142,7 @@ classDiagram
         +append(DomainEventStream) Mono~Void~
         +load(AggregateId, headVersion, tailVersion) Flux~DomainEventStream~
         +load(AggregateId, headEventTime, tailEventTime) Flux~DomainEventStream~
+        +scanAggregateId(NamedAggregate, String, Int) Flux~AggregateId~
     }
     class AbstractEventStore {
         <<abstract>>

documentation/docs/en/guide/extensions/mongo.md

@@ -415,20 +415,19 @@ classDiagram
         +appendStream(DomainEventStream) Mono~Void~
         +loadStream(...) Flux~DomainEventStream~
         +last(AggregateId) Mono~DomainEventStream~
+        +scanAggregateId(...) Flux~AggregateId~
     }
 
     class SnapshotStore {
         <<interface>>
         +load(AggregateId) Mono~Snapshot~
         +save(Snapshot) Mono~Void~
-        +scanAggregateId(NamedAggregate, String, Int) Flux~AggregateId~
     }
 
     class MongoSnapshotStore {
         -database: MongoDatabase
         +load(AggregateId) Mono~Snapshot~
         +save(Snapshot) Mono~Void~
-        +scanAggregateId(...) Flux~AggregateId~
     }
 
     class PrepareKey~V~ {

documentation/docs/en/guide/snapshot.md

@@ -101,10 +101,10 @@ stateDiagram-v2
 
 ## Snapshot Store
 
-The snapshot store is responsible for storing and retrieving snapshots.
+The snapshot store is responsible for storing and retrieving snapshots. Batch aggregate ID scanning belongs to `EventStore.scanAggregateId(...)`, not to the snapshot store.
 
 ```kotlin
-interface SnapshotStore : Named, AggregateIdScanner {
+interface SnapshotStore : Named {
     fun <S : Any> load(aggregateId: AggregateId): Mono<Snapshot<S>>
     fun <S : Any> save(snapshot: Snapshot<S>): Mono<Void>
     fun getVersion(aggregateId: AggregateId): Mono<Int>

documentation/docs/zh/guide/eventstore.md

@@ -23,10 +23,10 @@ description: 事件存储是事件溯源架构的核心持久化引擎 -- 不可
 
 ## 核心接口
 
-`EventStore` 接口定义了事件存储的核心操作：
+`EventStore` 接口定义了事件存储的核心操作，并承担按命名聚合分页扫描聚合 ID 的职责：
 
 ```kotlin
-interface EventStore {
+interface EventStore : AggregateIdScanner {
     fun append(eventStream: DomainEventStream): Mono<Void>
     fun load(
         aggregateId: AggregateId,
@@ -38,6 +38,12 @@ interface EventStore {
         headEventTime: Long,
         tailEventTime: Long
     ): Flux<DomainEventStream>
+    fun last(aggregateId: AggregateId): Mono<DomainEventStream>
+    fun scanAggregateId(
+        namedAggregate: NamedAggregate,
+        afterId: String = AggregateIdScanner.FIRST_ID,
+        limit: Int = 10
+    ): Flux<AggregateId>
 }
 ```
 
@@ -63,7 +69,7 @@ interface DomainEventStream : EventMessage<DomainEventStream, List<DomainEvent<*
 |---|---|---|
 | `DomainEvent` | 关于聚合内过去业务行为的不可变事实 | [DomainEvent.kt:52-95](https://github.com/Ahoo-Wang/Wow/blob/main/wow-api/src/main/kotlin/me/ahoo/wow/api/event/DomainEvent.kt#L52-L95) |
 | `DomainEventStream` | 单个命令产生的有序领域事件批次 | [DomainEventStream.kt:51-125](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/event/DomainEventStream.kt#L51-L125) |
-| `EventStore` | 追加和加载事件流的核心接口 | [EventStore.kt:27-98](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/eventsourcing/EventStore.kt#L27-L98) |
+| `EventStore` | 追加、加载事件流并扫描聚合 ID 的核心接口 | [EventStore.kt](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/eventsourcing/EventStore.kt) |
 | `SnapshotStore` | 通过带版本的快照检查点优化聚合加载 | [SnapshotStore.kt:27-58](https://github.com/Ahoo-Wang/Wow/blob/main/wow-core/src/main/kotlin/me/ahoo/wow/eventsourcing/snapshot/SnapshotStore.kt#L27-L58) |
 
 ## 聚合状态重建
@@ -136,6 +142,7 @@ classDiagram
         +append(DomainEventStream) Mono~Void~
         +load(AggregateId, headVersion, tailVersion) Flux~DomainEventStream~
         +load(AggregateId, headEventTime, tailEventTime) Flux~DomainEventStream~
+        +scanAggregateId(NamedAggregate, String, Int) Flux~AggregateId~
     }
     class AbstractEventStore {
         <<abstract>>

documentation/docs/zh/guide/extensions/mongo.md

@@ -415,20 +415,19 @@ classDiagram
         +appendStream(DomainEventStream) Mono~Void~
         +loadStream(...) Flux~DomainEventStream~
         +last(AggregateId) Mono~DomainEventStream~
+        +scanAggregateId(...) Flux~AggregateId~
     }
 
     class SnapshotStore {
         <<interface>>
         +load(AggregateId) Mono~Snapshot~
         +save(Snapshot) Mono~Void~
-        +scanAggregateId(NamedAggregate, String, Int) Flux~AggregateId~
     }
 
     class MongoSnapshotStore {
         -database: MongoDatabase
         +load(AggregateId) Mono~Snapshot~
         +save(Snapshot) Mono~Void~
-        +scanAggregateId(...) Flux~AggregateId~
     }
 
     class PrepareKey~V~ {

documentation/docs/zh/guide/snapshot.md

@@ -101,10 +101,10 @@ stateDiagram-v2
 
 ## 快照存储
 
-快照存储负责存储和检索快照。
+快照存储负责存储和检索快照。批量扫描聚合 ID 属于 `EventStore.scanAggregateId(...)`，而不是快照存储职责。
 
 ```kotlin
-interface SnapshotStore : Named, AggregateIdScanner {
+interface SnapshotStore : Named {
     fun <S : Any> load(aggregateId: AggregateId): Mono<Snapshot<S>>
     fun <S : Any> save(snapshot: Snapshot<S>): Mono<Void>
     fun getVersion(aggregateId: AggregateId): Mono<Int>

test/wow-mock/src/main/kotlin/me/ahoo/wow/eventsourcing/mock/DelayEventStore.kt

@@ -14,6 +14,7 @@
 package me.ahoo.wow.eventsourcing.mock
 
 import me.ahoo.wow.api.modeling.AggregateId
+import me.ahoo.wow.api.modeling.NamedAggregate
 import me.ahoo.wow.event.DomainEventStream
 import me.ahoo.wow.eventsourcing.EventStore
 import me.ahoo.wow.eventsourcing.InMemoryEventStore
@@ -42,4 +43,12 @@ class DelayEventStore(
     override fun last(aggregateId: AggregateId): Mono<DomainEventStream> {
         return delegate.last(aggregateId).delaySubscription(delaySupplier())
     }
+
+    override fun scanAggregateId(
+        namedAggregate: NamedAggregate,
+        afterId: String,
+        limit: Int
+    ): Flux<AggregateId> {
+        return delegate.scanAggregateId(namedAggregate, afterId, limit).delaySubscription(delaySupplier())
+    }
 }

test/wow-mock/src/main/kotlin/me/ahoo/wow/eventsourcing/mock/DelaySnapshotStore.kt

@@ -14,12 +14,10 @@
 package me.ahoo.wow.eventsourcing.mock
 
 import me.ahoo.wow.api.modeling.AggregateId
-import me.ahoo.wow.api.modeling.NamedAggregate
 import me.ahoo.wow.eventsourcing.snapshot.InMemorySnapshotStore
 import me.ahoo.wow.eventsourcing.snapshot.Snapshot
 import me.ahoo.wow.eventsourcing.snapshot.SnapshotStore
 import me.ahoo.wow.infra.Decorator
-import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import java.time.Duration
 
@@ -42,8 +40,4 @@ class DelaySnapshotStore(
     override fun <S : Any> save(snapshot: Snapshot<S>): Mono<Void> {
         return delegate.save(snapshot).delaySubscription(delaySupplier())
     }
-
-    override fun scanAggregateId(namedAggregate: NamedAggregate, afterId: String, limit: Int): Flux<AggregateId> {
-        return delegate.scanAggregateId(namedAggregate, afterId, limit).delaySubscription(delaySupplier())
-    }
 }

test/wow-tck/src/main/kotlin/me/ahoo/wow/tck/eventsourcing/EventStoreSpec.kt

@@ -25,7 +25,7 @@ import me.ahoo.wow.eventsourcing.EventVersionConflictException
 import me.ahoo.wow.id.generateGlobalId
 import me.ahoo.wow.metrics.Metrics.metrizable
 import me.ahoo.wow.modeling.aggregateId
-import me.ahoo.wow.tck.event.MockDomainEventStreams.generateEventStream
+import me.ahoo.wow.modeling.toNamedAggregate
 import me.ahoo.wow.tck.metrics.LoggingMeterRegistryInitializer
 import me.ahoo.wow.tck.mock.MockAggregateCreated
 import me.ahoo.wow.test.aggregate.GivenInitializationCommand
@@ -36,6 +36,7 @@ import org.junit.jupiter.api.extension.ExtendWith
 import reactor.core.publisher.Flux
 import reactor.core.scheduler.Schedulers
 import reactor.kotlin.test.test
+import me.ahoo.wow.tck.event.MockDomainEventStreams.generateEventStream as generateMockEventStream
 
 /**
  * Provides tests for verifying `EventStore` specification rules.
@@ -55,7 +56,7 @@ abstract class EventStoreSpec {
     protected abstract fun createEventStore(): EventStore
 
     protected fun generateEventStream(aggregateId: AggregateId): DomainEventStream {
-        return generateEventStream(aggregateId, eventCount = 10)
+        return generateMockEventStream(aggregateId, eventCount = 10)
     }
 
     protected fun generateEventStream(): DomainEventStream {
@@ -371,6 +372,163 @@ abstract class EventStoreSpec {
         }
     }
 
+    @Test
+    open fun scanAggregateId() {
+        val eventStore = createEventStore().metrizable()
+        val cursorId = generateGlobalId()
+        val aggregateId = namedAggregate.aggregateId(generateGlobalId())
+        eventStore.append(generateEventStream(aggregateId))
+            .test()
+            .verifyComplete()
+
+        eventStore.scanAggregateId(namedAggregate, afterId = cursorId, limit = 1)
+            .test()
+            .expectNext(aggregateId)
+            .verifyComplete()
+        eventStore.scanAggregateId(namedAggregate, afterId = aggregateId.id, limit = 1)
+            .test()
+            .expectNextCount(0)
+            .verifyComplete()
+    }
+
+    @Test
+    open fun scanAggregateIdShouldFilterNamedAggregate() {
+        val eventStore = createEventStore().metrizable()
+        val cursorId = generateGlobalId()
+        val targetAggregateId = namedAggregate.aggregateId(generateGlobalId())
+        val otherAggregateId = "other_aggregate"
+            .toNamedAggregate(namedAggregate.contextName)
+            .aggregateId(generateGlobalId())
+        eventStore.append(generateEventStream(targetAggregateId))
+            .test()
+            .verifyComplete()
+        eventStore.append(generateEventStream(otherAggregateId))
+            .test()
+            .verifyComplete()
+
+        eventStore.scanAggregateId(namedAggregate, afterId = cursorId, limit = 10)
+            .collectList()
+            .test()
+            .consumeNextWith {
+                it.assert().containsExactly(targetAggregateId)
+            }
+            .verifyComplete()
+    }
+
+    @Test
+    open fun scanAggregateIdShouldFilterBoundedContext() {
+        val eventStore = createEventStore().metrizable()
+        val cursorId = generateGlobalId()
+        val targetAggregateId = namedAggregate.aggregateId(generateGlobalId())
+        val otherAggregateId = namedAggregate.aggregateName
+            .toNamedAggregate(contextName = "other_context")
+            .aggregateId(generateGlobalId())
+        eventStore.append(generateEventStream(targetAggregateId))
+            .test()
+            .verifyComplete()
+        eventStore.append(generateEventStream(otherAggregateId))
+            .test()
+            .verifyComplete()
+
+        eventStore.scanAggregateId(namedAggregate, afterId = cursorId, limit = 10)
+            .collectList()
+            .test()
+            .consumeNextWith {
+                it.assert().containsExactly(targetAggregateId)
+            }
+            .verifyComplete()
+    }
+
+    @Test
+    open fun scanAggregateIdShouldPreserveTenantId() {
+        val eventStore = createEventStore().metrizable()
+        val cursorId = generateGlobalId()
+        val aggregateId = namedAggregate.aggregateId(generateGlobalId(), tenantId = "tenant-1")
+        eventStore.append(generateEventStream(aggregateId))
+            .test()
+            .verifyComplete()
+
+        eventStore.scanAggregateId(namedAggregate, afterId = cursorId, limit = 10)
+            .collectList()
+            .test()
+            .consumeNextWith {
+                it.assert().containsExactly(aggregateId)
+            }
+            .verifyComplete()
+    }
+
+    @Test
+    open fun scanAggregateIdShouldLimitResult() {
+        val eventStore = createEventStore().metrizable()
+        val cursorId = generateGlobalId()
+        val aggregateIds = (1..20).map {
+            namedAggregate.aggregateId(generateGlobalId())
+        }
+        aggregateIds.forEach { aggregateId ->
+            eventStore.append(generateEventStream(aggregateId))
+                .test()
+                .verifyComplete()
+        }
+
+        eventStore.scanAggregateId(namedAggregate, afterId = cursorId, limit = 3)
+            .collectList()
+            .test()
+            .consumeNextWith {
+                it.assert().hasSize(3)
+            }
+            .verifyComplete()
+    }
+
+    @Test
+    open fun scanAggregateIdShouldReturnLexicographicalOrder() {
+        val eventStore = createEventStore().metrizable()
+        val idPrefix = generateGlobalId()
+        val aggregateIds = listOf("003", "001", "004", "002").map {
+            namedAggregate.aggregateId("$idPrefix-$it")
+        }
+        aggregateIds.forEach { aggregateId ->
+            eventStore.append(generateEventStream(aggregateId))
+                .test()
+                .verifyComplete()
+        }
+
+        eventStore.scanAggregateId(namedAggregate, afterId = "$idPrefix-001", limit = 2)
+            .collectList()
+            .test()
+            .consumeNextWith {
+                it.assert().containsExactly(
+                    namedAggregate.aggregateId("$idPrefix-002"),
+                    namedAggregate.aggregateId("$idPrefix-003"),
+                )
+            }
+            .verifyComplete()
+    }
+
+    @Test
+    open fun scanAggregateIdShouldReturnEachAggregateOnce() {
+        val eventStore = createEventStore().metrizable()
+        val cursorId = generateGlobalId()
+        val aggregateId = namedAggregate.aggregateId(generateGlobalId())
+        eventStore.append(generateEventStream(aggregateId))
+            .test()
+            .verifyComplete()
+        eventStore.append(
+            generateMockEventStream(
+                aggregateId = aggregateId,
+                aggregateVersion = Version.INITIAL_VERSION,
+            )
+        ).test()
+            .verifyComplete()
+
+        eventStore.scanAggregateId(namedAggregate, afterId = cursorId, limit = 10)
+            .collectList()
+            .test()
+            .consumeNextWith {
+                it.assert().containsExactly(aggregateId)
+            }
+            .verifyComplete()
+    }
+
     companion object {
         const val TIMES = 1000
         const val DEFAULT_PARALLELISM = 2