|
| 1 | +-- | A file-based event source and sink using JSON encoding. |
| 2 | +-- |
| 3 | +-- This serves as an example of how to create an 'EventSource' and 'EventSink'. |
| 4 | +module Hydra.Events.FileBased where |
| 5 | + |
| 6 | +import Hydra.Prelude |
| 7 | + |
| 8 | +import Control.Concurrent.Class.MonadSTM (newTVarIO, writeTVar) |
| 9 | +import Hydra.Chain (IsChainState) |
| 10 | +import Hydra.Events (EventSink (..), EventSource (..), StateEvent (..)) |
| 11 | +import Hydra.HeadLogic.Outcome (StateChanged) |
| 12 | +import Hydra.Persistence (PersistenceIncremental (..)) |
| 13 | + |
| 14 | +-- | A basic file based event source and sink defined using an |
| 15 | +-- 'PersistenceIncremental' handle. |
| 16 | +-- |
| 17 | +-- The complexity in this implementation mostly stems from the fact that we want |
| 18 | +-- to be backward-compatible with the old, plain format of storing |
| 19 | +-- 'StateChanged' items directly to disk using 'PersistenceIncremental'. |
| 20 | +-- |
| 21 | +-- If any 'Legacy StateChanged' items are discovered, a running index is used |
| 22 | +-- for the 'eventId', while the 'New StateEvent' values are just stored as is. |
| 23 | +-- |
| 24 | +-- A new implementation for an 'EventSource' with a compatible 'EventSink' could |
| 25 | +-- be defined more generically with constraints: |
| 26 | +-- |
| 27 | +-- (ToJSON e, FromJSON e, HasEventId) e => (EventSource e m, EventSink e m) |
| 28 | +eventPairFromPersistenceIncremental :: |
| 29 | + (IsChainState tx, MonadSTM m) => |
| 30 | + PersistenceIncremental (PersistedStateChange tx) m -> |
| 31 | + m (EventSource (StateEvent tx) m, EventSink (StateEvent tx) m) |
| 32 | +eventPairFromPersistenceIncremental PersistenceIncremental{append, loadAll} = do |
| 33 | + eventIdV <- newTVarIO Nothing |
| 34 | + let |
| 35 | + getLastSeenEventId = readTVar eventIdV |
| 36 | + |
| 37 | + setLastSeenEventId StateEvent{eventId} = do |
| 38 | + writeTVar eventIdV (Just eventId) |
| 39 | + |
| 40 | + getNextEventId = |
| 41 | + maybe 0 (+ 1) <$> readTVar eventIdV |
| 42 | + |
| 43 | + -- Keep track of the last seen event id when loading |
| 44 | + getEvents = do |
| 45 | + items <- loadAll |
| 46 | + atomically . forM items $ \i -> do |
| 47 | + event <- case i of |
| 48 | + New e -> pure e |
| 49 | + Legacy sc -> do |
| 50 | + eventId <- getNextEventId |
| 51 | + pure $ StateEvent eventId sc |
| 52 | + |
| 53 | + setLastSeenEventId event |
| 54 | + pure event |
| 55 | + |
| 56 | + -- Filter events that are already stored |
| 57 | + putEvent e@StateEvent{eventId} = do |
| 58 | + atomically getLastSeenEventId >>= \case |
| 59 | + Nothing -> store e |
| 60 | + Just lastSeenEventId |
| 61 | + | eventId > lastSeenEventId -> store e |
| 62 | + | otherwise -> pure () |
| 63 | + |
| 64 | + store e = do |
| 65 | + append (New e) |
| 66 | + atomically $ setLastSeenEventId e |
| 67 | + |
| 68 | + pure (EventSource{getEvents}, EventSink{putEvent}) |
| 69 | + |
| 70 | +-- | Internal data type used by 'createJSONFileEventSourceAndSink' to be |
| 71 | +-- compatible with plain usage of 'PersistenceIncrementa' using plain |
| 72 | +-- 'StateChanged' items to the new 'StateEvent' persisted items. |
| 73 | +data PersistedStateChange tx |
| 74 | + = Legacy (StateChanged tx) |
| 75 | + | New (StateEvent tx) |
| 76 | + deriving stock (Generic, Show, Eq) |
| 77 | + |
| 78 | +instance IsChainState tx => ToJSON (PersistedStateChange tx) where |
| 79 | + toJSON = \case |
| 80 | + Legacy sc -> toJSON sc |
| 81 | + New e -> toJSON e |
| 82 | + |
| 83 | +instance IsChainState tx => FromJSON (PersistedStateChange tx) where |
| 84 | + parseJSON v = |
| 85 | + New <$> parseJSON v |
| 86 | + <|> Legacy <$> parseJSON v |
0 commit comments