feat: FeedAggregationStore

packages/core/src/state-management/repository.ts

@@ -42,6 +42,7 @@ import { IReconApi, ReconEventFeedResponse } from '../recon.js'
 import { Utils } from '../utils.js'
 import { ModelInstanceDocument } from '@ceramicnetwork/stream-model-instance'
 import { Model } from '@ceramicnetwork/stream-model'
+import { FeedAggregationStore } from '../store/feed-aggregation-store.js'
 
 const DEFAULT_LOAD_OPTS = { sync: SyncOptions.PREFER_CACHE, syncTimeoutSeconds: 3 }
 const APPLY_ANCHOR_COMMIT_ATTEMPTS = 3
@@ -121,6 +122,7 @@ export class Repository {
   readonly inmemory: StateCache<RunningState>
 
   private readonly feed: Feed
+  private readonly feedAggregationStore: FeedAggregationStore
 
   private reconEventFeedSubscription: Subscription | undefined
 
@@ -163,6 +165,7 @@ export class Repository {
       state$.complete()
     })
     this.updates$ = this.updates$.bind(this)
+    this.feedAggregationStore = new FeedAggregationStore()
   }
 
   /**
@@ -177,6 +180,7 @@ export class Repository {
   }
 
   async init(): Promise<void> {
+    await this.feedAggregationStore.open(this.#deps.kvFactory)
     await this.pinStore.open(this.#deps.kvFactory)
     await this.anchorRequestStore.open(this.#deps.kvFactory) // Initialization hell
     await this.index.init()

packages/core/src/store/feed-aggregation-store.ts

@@ -0,0 +1,21 @@
+import { ObjectStore } from './object-store.js'
+import { StreamID } from '@ceramicnetwork/streamid'
+
+function serializeStreamID(streamID: StreamID): string {
+  return streamID.toString()
+}
+
+function deserializeStreamID(input: string): StreamID {
+  return StreamID.fromString(input)
+}
+
+/**
+ * A storage for feed aggregation queue: key is a timestamp, value is StreamID.
+ */
+export class FeedAggregationStore extends ObjectStore<number, StreamID> {
+  protected useCaseName = 'feed-aggregation'
+
+  constructor() {
+    super(String, serializeStreamID, deserializeStreamID)
+  }
+}