acid-state-events (empty) → 0.1.0.0

raw patch · 2 files changed

+274/−0 lines, 2 filesdep +acid-statedep +basedep +containers

Dependencies added: acid-state, base, containers, stm, time

Files

• acid-state-events.cabal +60/−0
• src/Data/Acid/Events.hs +214/−0

@@ -0,0 +1,60 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.38.2.+--+-- see: https://github.com/sol/hpack++name:           acid-state-events+version:        0.1.0.0+synopsis:       Generic event bus for acid-state applications+description:    A generic event bus implementation for acid-state applications.+                Provides publish/subscribe with event logging and replay capabilities.+category:       Database+homepage:       https://github.com/juspay/vira/tree/main/packages/acid-state-events+author:         Sridhar Ratnakumar+maintainer:     srid@srid.ca+copyright:      2025 Sridhar Ratnakumar+license:        MIT+build-type:     Simple++library+  exposed-modules:+      Data.Acid.Events+  other-modules:+      Paths_acid_state_events+  hs-source-dirs:+      src+  default-extensions:+      DataKinds+      DeriveDataTypeable+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DerivingVia+      ExplicitForAll+      FlexibleContexts+      FlexibleInstances+      GeneralizedNewtypeDeriving+      ImportQualifiedPost+      LambdaCase+      MultiParamTypeClasses+      MultiWayIf+      NamedFieldPuns+      NoStarIsType+      NumericUnderscores+      OverloadedStrings+      ScopedTypeVariables+      StrictData+      TypeApplications+      TypeFamilies+      TypeOperators+      TypeSynonymInstances+      ViewPatterns+  ghc-options: -Wall -optP-Wno-nonportable-include-path -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-deriving-strategies -Wunused-foralls -fprint-explicit-foralls -fprint-explicit-kinds+  build-depends:+      acid-state+    , base ==4.*+    , containers+    , stm+    , time+  default-language: GHC2021

@@ -0,0 +1,214 @@+{-# 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)