Safe Haskell	None
Language	GHC2021

Hindsight.Store.Memory.Internal

Contents

Core Types
Cursor that can be rebuilt from sequence numbers
Version Control
Event Processing
State Management
Common Operations

Synopsis

data StoreState backend = StoreState {
- nextSequence :: Integer
- events :: Map Integer StoredEvent
- streamEvents :: Map StreamId [Integer]
- streamVersions :: Map StreamId (Cursor backend)
- streamLocalVersions :: Map StreamId StreamVersion
- streamNotifications :: Map StreamId (TVar Integer)
- globalNotification :: TVar Integer
}
data StoredEvent = StoredEvent {
- seqNo :: Integer
- eventId :: EventId
- streamId :: StreamId
- correlationId :: Maybe CorrelationId
- createdAt :: UTCTime
- eventName :: Text
- eventVersion :: Integer
- payload :: Value
- streamVersion :: StreamVersion
}
class StoreCursor backend where
- makeCursor :: Integer -> Cursor backend
- makeSequenceNo :: Cursor backend -> Integer
checkVersionConstraint :: Eq (Cursor backend) => StoreState backend -> StreamId -> ExpectedVersion backend -> Either (VersionMismatch backend) ()
getCurrentVersion :: StoreState backend -> StreamId -> Maybe (Cursor backend)
getCurrentStreamVersion :: StoreState backend -> StreamId -> Maybe StreamVersion
processEvents :: forall (ts :: [Symbol]) m backend. (MonadUnliftIO m, StoreCursor backend, Show (Cursor backend)) => EventMatcher ts backend m -> [StoredEvent] -> m SubscriptionResult
processEventThroughMatchers :: forall m backend (event :: Symbol) (ts :: [Symbol]). (MonadUnliftIO m, Event event, StoreCursor backend, Show (Cursor backend)) => (Proxy event, EventHandler event m backend) -> EventMatcher ts backend m -> StoredEvent -> m SubscriptionResult
updateState :: StoreCursor backend => StoredEvent -> StoreState backend -> StoreState backend
makeStoredEvents :: forall (t :: Type -> Type) backend. Foldable t => StoreState backend -> Maybe CorrelationId -> UTCTime -> [EventId] -> StreamId -> StreamWrite t SomeLatestEvent backend -> ([StoredEvent], [Integer])
checkAllVersions :: forall (t :: Type -> Type) backend. Eq (Cursor backend) => StoreState backend -> Map StreamId (StreamWrite t SomeLatestEvent backend) -> Either (EventStoreError backend) ()
insertAllEvents :: forall backend (t :: Type -> Type). (StoreCursor backend, Foldable t) => StoreState backend -> Maybe CorrelationId -> UTCTime -> [EventId] -> Map StreamId (StreamWrite t SomeLatestEvent backend) -> (StoreState backend, Cursor backend, Map StreamId (Cursor backend))
subscribeToEvents :: forall m backend (ts :: [Symbol]). (MonadUnliftIO m, StoreCursor backend, Show (Cursor backend)) => TVar (StoreState backend) -> EventMatcher ts backend m -> EventSelector backend -> m (SubscriptionHandle backend)

Core Types

data StoreState backend Source #

Internal state maintained by the store.

Constructors

StoreState

Fields

nextSequence :: Integer
Next global sequence number to assign
events :: Map Integer StoredEvent
All events indexed by sequence number
streamEvents :: Map StreamId [Integer]
Stream-to-sequence-numbers index
streamVersions :: Map StreamId (Cursor backend)
Latest global cursor per stream
streamLocalVersions :: Map StreamId StreamVersion
Latest local version per stream
streamNotifications :: Map StreamId (TVar Integer)
Per-stream notification variables for subscriptions
globalNotification :: TVar Integer
Global notification variable for all-stream subscriptions

Instances

Instances details

Eq (Cursor backend) => Eq (StoreState backend) Source #
Instance details Defined in Hindsight.Store.Memory.Internal Methods (==) :: StoreState backend -> StoreState backend -> Bool # (/=) :: StoreState backend -> StoreState backend -> Bool #

data StoredEvent Source #

Raw stored event with minimal type information.

Constructors

StoredEvent

Fields

seqNo :: Integer
Global sequence number for total ordering
eventId :: EventId
Unique event identifier
streamId :: StreamId
Stream this event belongs to
correlationId :: Maybe CorrelationId
Optional correlation ID for tracking
createdAt :: UTCTime
Event creation timestamp
eventName :: Text
Event type name
eventVersion :: Integer
Event payload version number
payload :: Value
JSON payload
streamVersion :: StreamVersion
Local stream version (1, 2, 3...)

Instances

Instances details

FromJSON StoredEvent Source #

Instance details

Defined in Hindsight.Store.Memory.Internal

Methods

parseJSON :: Value -> Parser StoredEvent #

parseJSONList :: Value -> Parser [StoredEvent] #

omittedField :: Maybe StoredEvent #

ToJSON StoredEvent Source #

Instance details

Defined in Hindsight.Store.Memory.Internal

Methods

toJSON :: StoredEvent -> Value #

toEncoding :: StoredEvent -> Encoding #

toJSONList :: [StoredEvent] -> Value #

toEncodingList :: [StoredEvent] -> Encoding #

omitField :: StoredEvent -> Bool #

Generic StoredEvent Source #

Instance details

Defined in Hindsight.Store.Memory.Internal

Associated Types

type Rep StoredEvent

Instance details

Defined in Hindsight.Store.Memory.Internal

Methods

from :: StoredEvent -> Rep StoredEvent x #

to :: Rep StoredEvent x -> StoredEvent #

Show StoredEvent Source #

Instance details

Defined in Hindsight.Store.Memory.Internal

Methods

showsPrec :: Int -> StoredEvent -> ShowS #

show :: StoredEvent -> String #

showList :: [StoredEvent] -> ShowS #

Eq StoredEvent Source #

Instance details

Defined in Hindsight.Store.Memory.Internal

Methods

(==) :: StoredEvent -> StoredEvent -> Bool #

(/=) :: StoredEvent -> StoredEvent -> Bool #

type Rep StoredEvent Source #

Instance details

Defined in Hindsight.Store.Memory.Internal

Cursor that can be rebuilt from sequence numbers

class StoreCursor backend where Source #

Capability for creating cursors from sequence numbers

Methods

makeCursor :: Integer -> Cursor backend Source #

Create a cursor from a sequence number

makeSequenceNo :: Cursor backend -> Integer Source #

Get the current sequence number

Instances

Instances details

StoreCursor MemoryStore Source #
Instance details Defined in Hindsight.Store.Memory Methods makeCursor :: Integer -> Cursor MemoryStore Source # makeSequenceNo :: Cursor MemoryStore -> Integer Source #

Version Control

checkVersionConstraint :: Eq (Cursor backend) => StoreState backend -> StreamId -> ExpectedVersion backend -> Either (VersionMismatch backend) () Source #

Check version constraints for a stream

getCurrentVersion :: StoreState backend -> StreamId -> Maybe (Cursor backend) Source #

Get current version of a stream (global cursor)

getCurrentStreamVersion :: StoreState backend -> StreamId -> Maybe StreamVersion Source #

Get current stream version of a stream (local cursor)

Event Processing

processEvents :: forall (ts :: [Symbol]) m backend. (MonadUnliftIO m, StoreCursor backend, Show (Cursor backend)) => EventMatcher ts backend m -> [StoredEvent] -> m SubscriptionResult Source #

Process events through matcher chain

processEventThroughMatchers :: forall m backend (event :: Symbol) (ts :: [Symbol]). (MonadUnliftIO m, Event event, StoreCursor backend, Show (Cursor backend)) => (Proxy event, EventHandler event m backend) -> EventMatcher ts backend m -> StoredEvent -> m SubscriptionResult Source #

Process a single event through matchers

State Management

updateState :: StoreCursor backend => StoredEvent -> StoreState backend -> StoreState backend Source #

Update store state with new events

Common Operations

makeStoredEvents :: forall (t :: Type -> Type) backend. Foldable t => StoreState backend -> Maybe CorrelationId -> UTCTime -> [EventId] -> StreamId -> StreamWrite t SomeLatestEvent backend -> ([StoredEvent], [Integer]) Source #

Make stored events from raw data

checkAllVersions :: forall (t :: Type -> Type) backend. Eq (Cursor backend) => StoreState backend -> Map StreamId (StreamWrite t SomeLatestEvent backend) -> Either (EventStoreError backend) () Source #

Check version constraints for all streams

insertAllEvents :: forall backend (t :: Type -> Type). (StoreCursor backend, Foldable t) => StoreState backend -> Maybe CorrelationId -> UTCTime -> [EventId] -> Map StreamId (StreamWrite t SomeLatestEvent backend) -> (StoreState backend, Cursor backend, Map StreamId (Cursor backend)) Source #

Insert all events into state Returns: (new state, final global cursor, per-stream cursors)

subscribeToEvents :: forall m backend (ts :: [Symbol]). (MonadUnliftIO m, StoreCursor backend, Show (Cursor backend)) => TVar (StoreState backend) -> EventMatcher ts backend m -> EventSelector backend -> m (SubscriptionHandle backend) Source #

Common implementation of event subscription