{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE OverloadedRecordDot #-}
{- | Event bus for acid-state applications
This module provides a type-safe event bus that works with any acid-state application.
The event bus:
- Publishes 'SomeUpdate' events to subscribers via 'TChan'
- Maintains a circular buffer of recent events for debugging
- Supports type-safe pattern matching on events via 'Typeable'
-}
module Data.Acid.Events (
-- * Core types
SomeUpdate (..),
EventBus (..),
-- * Initialization
newEventBus,
newEventBusWithSize,
-- * acid-state wrappers
update,
-- * Event bus operations
subscribe,
awaitBatched,
getRecentEvents,
-- * Pattern matching
matchUpdate,
) where
import Control.Concurrent (threadDelay)
import Control.Concurrent.STM (TChan, TVar, atomically, dupTChan, isEmptyTChan, modifyTVar', newBroadcastTChanIO, newTVarIO, readTChan, readTVarIO, tryReadTChan, writeTChan)
import Control.Monad (unless, void)
import Data.Acid (AcidState, EventResult, EventState, UpdateEvent)
import Data.Acid qualified as Acid
import Data.Foldable (toList)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Sequence (Seq, (|>))
import Data.Sequence qualified as Seq
import Data.Time (UTCTime, getCurrentTime)
import Data.Typeable (Typeable, cast)
import GHC.Generics (Generic)
import Text.Show qualified
import Unsafe.Coerce (unsafeCoerce)
-- * Core types
-- | Existential wrapper for any Update event
data SomeUpdate state
= forall event.
( UpdateEvent event
, EventState event ~ state
, Show event
, Typeable event
) =>
SomeUpdate
{ event :: event
, result :: EventResult event
, timestamp :: UTCTime
}
instance Text.Show.Show (SomeUpdate state) where
showsPrec d (SomeUpdate evt _result _time) = Text.Show.showsPrec d evt
-- | Event bus with broadcast channel and circular buffer log of 'SomeUpdate' events
data EventBus someUpdate = EventBus
{ channel :: TChan someUpdate
, eventLog :: TVar (Seq someUpdate)
, maxLogSize :: Int
}
deriving stock (Generic)
-- * Initialization
-- | Create a new 'EventBus' with default settings (1000 event buffer)
newEventBus :: IO (EventBus someUpdate)
newEventBus = newEventBusWithSize 1000
-- | Create a new 'EventBus' with custom buffer size
newEventBusWithSize :: Int -> IO (EventBus someUpdate)
newEventBusWithSize size = do
channel <- newBroadcastTChanIO
eventLog <- newTVarIO Seq.empty
pure
EventBus
{ channel
, eventLog
, maxLogSize = size
}
-- * 'EventBus' operations
-- | Publish an update event to all subscribers (internal)
publishUpdate ::
EventBus someUpdate ->
someUpdate ->
IO ()
publishUpdate bus someUpdate = atomically $ do
-- Publish to subscribers
writeTChan bus.channel someUpdate
-- Append to circular buffer log
modifyTVar' bus.eventLog $ \es ->
let es' = es |> someUpdate
in if Seq.length es' > maxLogSize bus
then Seq.drop 1 es'
else es'
{- | Execute an acid-state update and automatically publish to event bus
This is the main entry point for executing updates - it handles both
the acid-state transaction and event publishing in one operation.
-}
update ::
( UpdateEvent event
, EventState event ~ state
, Show event
, Typeable event
) =>
AcidState state ->
EventBus (SomeUpdate state) ->
event ->
IO (EventResult event)
update acid bus event = do
result <- Acid.update acid event
timestamp <- getCurrentTime
publishUpdate bus (SomeUpdate event result timestamp)
pure result
-- | Subscribe to events (returns duplicate 'TChan' starting from now)
subscribe ::
EventBus someUpdate ->
IO (TChan someUpdate)
subscribe bus = atomically $ do
dup <- dupTChan bus.channel
-- Drain any pending events so subscriber starts fresh
let drainLoop = do
isEmpty <- isEmptyTChan dup
unless isEmpty $ do
void $ readTChan dup
drainLoop
drainLoop
pure dup
{- | Wait for and collect matching events with debouncing
Useful for batching rapid event sequences (e.g., multiple file saves)
into a single processing cycle. Blocks until first matching event,
then collects additional matches during debounce window.
Example: Wait for repo updates, batch them over 2.5s window
@
batch <- awaitBatched chan isRepoUpdate 2_500_000
processBatch batch
@
-}
awaitBatched ::
TChan someUpdate ->
-- | Predicate to filter events
(someUpdate -> Bool) ->
-- | Debounce window in microseconds
Int ->
IO (NonEmpty someUpdate)
awaitBatched chan predicate debounceUs = do
-- Block until first matching event
firstUpdate <- atomically waitForMatch
-- Debounce: wait for more events to batch together
threadDelay debounceUs
-- Collect any additional matching events
moreUpdates <- drainMatching []
pure $ firstUpdate :| moreUpdates
where
-- Wait for an update that matches predicate
waitForMatch = do
evt <- readTChan chan
if predicate evt
then pure evt
else waitForMatch -- Retry in STM - keep reading until match
-- Drain additional matching updates (non-blocking)
drainMatching acc = do
mUpdate <- atomically $ tryReadTChan chan
case mUpdate of
Nothing -> pure $ reverse acc
Just evt ->
if predicate evt
then drainMatching (evt : acc)
else drainMatching acc -- Skip non-matching
-- | Get recent events from the log (for debug UI)
getRecentEvents ::
EventBus someUpdate ->
IO [someUpdate]
getRecentEvents bus = do
toList <$> readTVarIO bus.eventLog
-- * Pattern matching
{- | Pattern match on specific update type
Note: Returns unsafe-coerced result since 'EventResult' is a type family.
This is safe because if the update matches, the result type must match too.
-}
matchUpdate ::
forall event state.
(UpdateEvent event, Typeable event, EventState event ~ state) =>
SomeUpdate state ->
Maybe (event, EventResult event)
matchUpdate (SomeUpdate evt result _timestamp) = do
typedUpdate <- cast evt
-- Safe: if update type matches, result type must match (type family relation)
pure (typedUpdate, unsafeCoerce result)