feat(eventsourcing): move aggregate id scanning to EventStore

documentation/docs/en/guide/eventstore.md

@@ -23,10 +23,12 @@ 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 :
+    RequestIdExistenceChecker,
+    AggregateIdScanner {
     fun append(eventStream: DomainEventStream): Mono<Void>
     fun load(
         aggregateId: AggregateId,
@@ -38,6 +40,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 +71,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 +144,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/extensions/redis.md

@@ -145,24 +145,34 @@ Each processor corresponds to a consumer group:
 
 ## Event Store
 
-Redis event store uses Hash structure for event stream storage:
+Redis event store uses bucketed Redis Cluster hash tags so event stream append, request idempotency, and aggregate ID scanning indexes can be updated atomically in one Lua script.
 
 ### Data Structure
 
 ```
-Key: {prefix}{contextName}.{aggregateName}:{aggregateId}:es
-Field: {version}
-Value: {eventStreamJson}
+Event stream ZSET key: {{contextAlias}.{aggregateName}:es:{bucket}}:{aggregateId}@{tenantId}
+Score: {version}
+Member: {eventStreamJson}
+
+Request id SET key: {{contextAlias}.{aggregateName}:es:{bucket}}:{aggregateId}@{tenantId}:req_idx
+Member: {requestId}
+
+Aggregate ID ZSET key: {{contextAlias}.{aggregateName}:es:{bucket}}:ids
+Score: 0
+Member: {aggregateId}
+
+Aggregate tenant HASH key: {{contextAlias}.{aggregateName}:es:{bucket}}:tenants
+Field: {aggregateId}
+Value: {tenantId}
 ```
 
 ### Request Idempotency
 
-A separate Key is used to store request IDs for idempotency:
+Request IDs are stored in the bucket-aligned SET key shown above.
 
-```
-Key: {prefix}{contextName}.{aggregateName}:{aggregateId}:req:{requestId}
-TTL: Configured expiration time
-```
+### Aggregate ID Scanning
+
+`EventStore.scanAggregateId` scans bucketed aggregate ID indexes and merges the results in lexicographical order. Aggregate IDs are globally unique, so the scanner stores one aggregate ID member and resolves its tenant from the bucket-aligned tenant HASH.
 
 ## Snapshot Storage
 
@@ -354,4 +364,4 @@ wow:
 2. **Use Cluster Mode**: Use Redis cluster in production for high availability and scalability
 3. **Configure Connection Pool Properly**: Configure appropriate connection pool size based on concurrency
 4. **Monitor Memory Usage**: Regularly monitor Redis memory usage to avoid OOM
-5. **Enable Persistence**: Configure RDB or AOF persistence to prevent data loss
+5. **Enable Persistence**: Configure RDB or AOF persistence to prevent data loss

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,12 @@ description: 事件存储是事件溯源架构的核心持久化引擎 -- 不可
 
 ## 核心接口
 
-`EventStore` 接口定义了事件存储的核心操作：
+`EventStore` 接口定义了事件存储的核心操作，并承担按命名聚合分页扫描聚合 ID 的职责：
 
 ```kotlin
-interface EventStore {
+interface EventStore :
+    RequestIdExistenceChecker,
+    AggregateIdScanner {
     fun append(eventStream: DomainEventStream): Mono<Void>
     fun load(
         aggregateId: AggregateId,
@@ -38,6 +40,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 +71,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 +144,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/extensions/redis.md

@@ -145,24 +145,34 @@ Redis 命令总线使用 Redis Streams 实现消息传递：
 
 ## 事件存储
 
-Redis 事件存储使用 Hash 结构存储事件流：
+Redis 事件存储使用分桶的 Redis Cluster hash tag，使事件流追加、请求幂等性和聚合 ID 扫描索引可以在同一个 Lua 脚本中原子更新。
 
 ### 数据结构
 
 ```
-Key: {prefix}{contextName}.{aggregateName}:{aggregateId}:es
-Field: {version}
-Value: {eventStreamJson}
+事件流 ZSET Key: {{contextAlias}.{aggregateName}:es:{bucket}}:{aggregateId}@{tenantId}
+Score: {version}
+Member: {eventStreamJson}
+
+请求 ID SET Key: {{contextAlias}.{aggregateName}:es:{bucket}}:{aggregateId}@{tenantId}:req_idx
+Member: {requestId}
+
+聚合 ID ZSET Key: {{contextAlias}.{aggregateName}:es:{bucket}}:ids
+Score: 0
+Member: {aggregateId}
+
+聚合租户 HASH Key: {{contextAlias}.{aggregateName}:es:{bucket}}:tenants
+Field: {aggregateId}
+Value: {tenantId}
 ```
 
 ### 请求幂等性
 
-使用单独的 Key 存储请求 ID 实现幂等性：
+请求 ID 存储在上面的同分桶 SET Key 中。
 
-```
-Key: {prefix}{contextName}.{aggregateName}:{aggregateId}:req:{requestId}
-TTL: 配置的过期时间
-```
+### 聚合 ID 扫描
+
+`EventStore.scanAggregateId` 会扫描分桶的聚合 ID 索引，并按字典序合并结果。由于 `aggregateId` 全局唯一，scanner 只存储一个聚合 ID 成员，并从同分桶的租户 HASH 中解析对应的租户。
 
 ## 快照存储
 
@@ -354,4 +364,4 @@ wow:
 2. **使用集群模式**：生产环境使用 Redis 集群保证高可用和扩展性
 3. **合理配置连接池**：根据并发量配置适当的连接池大小
 4. **监控内存使用**：定期监控 Redis 内存使用，避免 OOM
-5. **启用持久化**：配置 RDB 或 AOF 持久化防止数据丢失
+5. **启用持久化**：配置 RDB 或 AOF 持久化防止数据丢失

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())
-    }
 }