feat: Add Entity Fetcher Service for Event Sourcing

core/nhcore.cabal

@@ -127,6 +127,8 @@ library
     Path
     Record
     Result
+    Service.EntityFetcher
+    Service.EntityFetcher.Core
     Service.Event
     Service.Event.EntityName
     Service.Event.EventMetadata
@@ -150,6 +152,10 @@ library
     Test.AppSpec.AppSpec
     Test.AppSpec.Scenario
     Test.AppSpec.Verify
+    Test.Service.EntityFetcher
+    Test.Service.EntityFetcher.Core
+    Test.Service.EntityFetcher.Fetch.Context
+    Test.Service.EntityFetcher.Fetch.Spec
     Test.Service.EventStore
     Test.Service.EventStore.BatchValidation.Context
     Test.Service.EventStore.BatchValidation.Spec

core/service/Service/EntityFetcher.hs

@@ -0,0 +1,5 @@
+module Service.EntityFetcher (
+  module Reexported,
+) where
+
+import Service.EntityFetcher.Core as Reexported

core/service/Service/EntityFetcher/Core.hs

@@ -0,0 +1,82 @@
+module Service.EntityFetcher.Core (
+  EntityFetcher (..),
+  Error (..),
+  new,
+) where
+
+import Core
+import Service.Event (EntityName, Event (..), StreamId)
+import Service.EventStore.Core (EventStore)
+import Service.EventStore.Core qualified as EventStore
+import Stream qualified
+import Task qualified
+
+
+-- | Errors that can occur during entity operations
+data Error
+  = EventStoreError EventStore.Error
+  | ReductionError Text
+  deriving (Eq, Show)
+
+
+-- | An entity fetcher is responsible for fetching an entity's current state
+--   by reading all events from the event store and applying a reduction function.
+--
+--   The fetcher takes an initial state and a reduce function that applies events
+--   to update the state.
+data EntityFetcher state event = EntityFetcher
+  { -- | Fetch an entity's current state by entity name and stream ID.
+    --   Reads all events from the event store and applies the reduction function
+    --   to compute the current state.
+    fetch :: EntityName -> StreamId -> Task Error state
+  }
+
+
+-- | Create a new entity fetcher with the given event store, initial state, and reduce function.
+--
+--   The reduce function takes an event and the current state, and returns the new state
+--   after applying the event.
+--
+--   Example:
+--
+--   @
+--   fetcher <- EntityFetcher.new eventStore initialState applyEvent
+--   state <- fetcher.fetch entityName streamId
+--   @
+new ::
+  forall state event.
+  EventStore event ->
+  state ->
+  (event -> state -> state) ->
+  Task Error (EntityFetcher state event)
+new eventStore initialState reduceFunction = do
+  let fetchImpl entityName streamId = do
+        -- Read all events for this entity stream
+        streamMessages <-
+          eventStore.readAllStreamEvents entityName streamId
+            |> Task.mapError EventStoreError
+
+        -- Consume the stream, applying the reduction function to each event
+        -- This processes events one-by-one without loading everything into memory
+        finalState <-
+          streamMessages
+            |> Stream.consume
+              ( \state message -> do
+                  case message of
+                    -- Only process actual events, ignore other message types
+                    EventStore.StreamEvent event -> do
+                      let eventData = event.event
+                      let newState = reduceFunction eventData state
+                      Task.yield newState
+                    -- For all other message types, keep state unchanged
+                    _ -> Task.yield state
+              )
+              initialState
+            |> Task.mapError (\errorText -> EventStoreError (EventStore.StorageFailure errorText))
+
+        Task.yield finalState
+
+  Task.yield
+    EntityFetcher
+      { fetch = fetchImpl
+      }

core/service/Service/EventStore/Postgres/Internal.hs

@@ -762,9 +762,21 @@ performReadStreamEvents
               Nothing -> 0
 
       positionRef <- Var.new streamPosition
+      -- Track if this is the first batch to use relative position correctly
+      isFirstBatchRef <- Var.new True
 
       Task.while (shouldKeepReading) do
-        selectSession <- Sessions.selectStreamEventBatch positionRef entityName streamId relative readDirection
+        isFirstBatch <- Var.get isFirstBatchRef
+        -- Use the original relative position for the first batch, then switch to position-based
+        currentRelative <-
+          if isFirstBatch
+            then Task.yield relative
+            else do
+              -- For subsequent batches, use FromAndAfter with current position
+              -- The position has already been incremented, so this avoids duplicates
+              currentPos <- Var.get positionRef
+              Task.yield (Just (FromAndAfter (StreamPosition currentPos)))
+        selectSession <- Sessions.selectStreamEventBatch positionRef entityName streamId currentRelative readDirection
         records <-
           selectSession
             |> Sessions.run conn
@@ -775,6 +787,10 @@ performReadStreamEvents
           stream |> Stream.writeItem StreamReadingStarted
           notifiedReadingRef |> Var.set True
 
+        -- Mark that we've completed the first batch
+        Task.when (isFirstBatch) do
+          isFirstBatchRef |> Var.set False
+
         Task.when (records |> Array.isEmpty) do
           breakLoopRef |> Var.set True
 

core/test/Service/EventStore/InMemorySpec.hs

@@ -4,11 +4,20 @@ import Core
 import Service.EventStore.InMemory qualified as InMemory
 import Task qualified
 import Test
+import Test.Service.EntityFetcher qualified as EntityFetcher
+import Test.Service.EntityFetcher.Core qualified as EntityFetcherCore
 import Test.Service.EventStore qualified as EventStore
 
 
 spec :: Spec Unit
 spec = do
   describe "InMemoryEventStore" do
-    let newStore = InMemory.new |> Task.mapError toText
+    let newStore = InMemory.new @EntityFetcherCore.BankAccountEvent |> Task.mapError toText
     EventStore.spec newStore
+
+    let newStoreAndFetcher = do
+          store <- newStore
+          fetcher <- EntityFetcherCore.newFetcher store |> Task.mapError toText
+          Task.yield (store, fetcher)
+
+    EntityFetcher.spec newStoreAndFetcher

core/test/Service/EventStore/Postgres/Internal/SubscriptionStoreSpec.hs

@@ -13,7 +13,7 @@ import Service.EventStore.Postgres.Internal.SubscriptionStore (Error (..), Subsc
 import Service.EventStore.Postgres.Internal.SubscriptionStore qualified as SubscriptionStore
 import Task qualified
 import Test
-import Test.Service.EventStore.Core (MyEvent (..))
+import Test.Service.EventStore.Core (BankAccountEvent (..))
 
 
 spec :: Spec Unit
@@ -312,12 +312,12 @@ spec = do
 
 
 -- Helper function to create a test event
-createTestEvent :: Task Error (Event MyEvent)
+createTestEvent :: Task Error (Event BankAccountEvent)
 createTestEvent = do
   metadata <- EventMetadata.new
   streamId <- StreamId.new
   let entityName = Event.EntityName "TestEntity"
-  let event = MyEvent
+  let event = MoneyDeposited {amount = 100}
   Task.yield
     Event
       { entityName = entityName,

core/test/Service/EventStore/PostgresSpec.hs

@@ -7,8 +7,10 @@ import Service.EventStore.Postgres.Internal.Core qualified as PostgresCore
 import Service.EventStore.Postgres.Internal.Sessions qualified as Sessions
 import Task qualified
 import Test
+import Test.Service.EntityFetcher qualified as EntityFetcher
+import Test.Service.EntityFetcher.Core qualified as EntityFetcherCore
 import Test.Service.EventStore qualified as EventStore
-import Test.Service.EventStore.Core (MyEvent)
+import Test.Service.EventStore.Core (BankAccountEvent)
 import Var qualified
 
 
@@ -26,34 +28,40 @@ spec = do
     describe "new method" do
       it "acquires the connection" \_ -> do
         (ops, observe) <- mockNewOps
-        Internal.new @MyEvent ops config
+        Internal.new @BankAccountEvent ops config
           |> Task.mapError toText
           |> discard
         observe.acquireCalls
           |> varContents shouldBe 1
 
       it "initializes the table" \_ -> do
         (ops, observe) <- mockNewOps
-        Internal.new @MyEvent ops config
+        Internal.new @BankAccountEvent ops config
           |> Task.mapError toText
           |> discard
         observe.initializeTableCalls
           |> varContents shouldBe 1
 
       it "initializes the subscriptions" \_ -> do
         (ops, observe) <- mockNewOps
-        Internal.new @MyEvent ops config
+        Internal.new @BankAccountEvent ops config
           |> Task.mapError toText
           |> discard
         observe.initializeSubscriptionsCalls
           |> varContents shouldBe 1
 
     let newStore = do
-          let ops = Internal.defaultOps @MyEvent
+          let ops = Internal.defaultOps @BankAccountEvent
           dropPostgres ops config
           Postgres.new config |> Task.mapError toText
     EventStore.spec newStore
 
+    let newStoreAndFetcher = do
+          store <- newStore
+          fetcher <- EntityFetcherCore.newFetcher store |> Task.mapError toText
+          Task.yield (store, fetcher)
+    EntityFetcher.spec newStoreAndFetcher
+
 
 data NewObserve = NewObserve
   { acquireCalls :: Var Int,

core/testlib/Test/Service/EntityFetcher.hs

@@ -0,0 +1,15 @@
+module Test.Service.EntityFetcher where
+
+import Core
+import Service.EntityFetcher.Core (EntityFetcher)
+import Service.EventStore.Core (EventStore)
+import Test
+import Test.Service.EntityFetcher.Core (BankAccountEvent, BankAccountState)
+import Test.Service.EntityFetcher.Fetch.Spec qualified as Fetch
+
+
+spec ::
+  Task Text (EventStore BankAccountEvent, EntityFetcher BankAccountState BankAccountEvent) -> Spec Unit
+spec newStoreAndFetcher = do
+  describe "Entity Fetcher Specification Tests" do
+    Fetch.spec newStoreAndFetcher

core/testlib/Test/Service/EntityFetcher/Core.hs

@@ -0,0 +1,73 @@
+module Test.Service.EntityFetcher.Core (
+  BankAccountEvent (..),
+  BankAccountState (..),
+  initialState,
+  applyEvent,
+  newFetcher,
+) where
+
+import Core
+import Json qualified
+import Service.EntityFetcher.Core (EntityFetcher)
+import Service.EntityFetcher.Core qualified as EntityFetcher
+import Service.EventStore.Core (EventStore)
+import Test.Service.EventStore.Core (BankAccountEvent (..))
+
+
+-- | Example entity state for a bank account
+data BankAccountState = BankAccountState
+  { balance :: Int,
+    isOpen :: Bool,
+    version :: Int
+  }
+  deriving (Eq, Show, Ord, Generic)
+
+
+instance Json.ToJSON BankAccountState
+
+
+instance Json.FromJSON BankAccountState
+
+
+-- | Initial state for a new bank account
+initialState :: BankAccountState
+initialState =
+  BankAccountState
+    { balance = 0,
+      isOpen = False,
+      version = 0
+    }
+
+
+-- | Apply a bank account event to the current state
+applyEvent :: BankAccountEvent -> BankAccountState -> BankAccountState
+applyEvent event state = do
+  let newVersion = state.version + 1
+  case event of
+    AccountOpened {initialBalance} ->
+      BankAccountState
+        { balance = initialBalance,
+          isOpen = True,
+          version = newVersion
+        }
+    MoneyDeposited {amount} ->
+      state
+        { balance = state.balance + amount,
+          version = newVersion
+        }
+    MoneyWithdrawn {amount} ->
+      state
+        { balance = state.balance - amount,
+          version = newVersion
+        }
+    AccountClosed ->
+      state
+        { isOpen = False,
+          version = newVersion
+        }
+
+
+-- | Create a new entity fetcher for bank accounts
+newFetcher :: EventStore BankAccountEvent -> Task EntityFetcher.Error (EntityFetcher BankAccountState BankAccountEvent)
+newFetcher eventStore = do
+  EntityFetcher.new eventStore initialState applyEvent

core/testlib/Test/Service/EntityFetcher/Fetch/Context.hs

@@ -0,0 +1,29 @@
+module Test.Service.EntityFetcher.Fetch.Context (
+  Context (..),
+  initialize,
+) where
+
+import Core
+import Service.EntityFetcher.Core (EntityFetcher)
+import Service.Event qualified as Event
+import Service.Event.StreamId qualified as StreamId
+import Service.EventStore (EventStore)
+import Test.Service.EntityFetcher.Core (BankAccountEvent, BankAccountState)
+
+
+data Context = Context
+  { store :: EventStore BankAccountEvent,
+    fetcher :: EntityFetcher BankAccountState BankAccountEvent,
+    streamId :: Event.StreamId,
+    entityName :: Event.EntityName
+  }
+
+
+initialize ::
+  Task Text (EventStore BankAccountEvent, EntityFetcher BankAccountState BankAccountEvent) ->
+  Task Text Context
+initialize newStoreAndFetcher = do
+  (store, fetcher) <- newStoreAndFetcher
+  streamId <- StreamId.new
+  let entityName = Event.EntityName "BankAccount"
+  pure Context {store, fetcher, streamId, entityName}