Trigger.hs

{-# LANGUAGE DeriveFunctor #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeOperators #-}
{-# LANGUAGE UndecidableInstances #-}

-- |
-- Module: Reflex.Vty.Host.Trigger
-- Description: A bounded 'TriggerEvent' layer that backpressures hot producers
--
-- This is an internal module used by "Reflex.Vty.Host". It provides a
-- direct replacement for reflex's
-- @'Reflex.TriggerEvent.Base.TriggerEventT'@ whose underlying queue is
-- bounded and closeable. External triggers
-- ('Reflex.performEventAsync', 'Reflex.newTriggerEvent', etc.) write
-- into a fixed-capacity queue. A producer that fires faster than the
-- host can process is throttled: its @fire@ callback blocks when the
-- queue is full. Memory use stays bounded. Because the producer's own
-- write blocks, no event occurrences are dropped.
--
-- The queue is also closeable. On shutdown the host closes it. This
-- releases any producer currently blocked on a full queue.
module Reflex.Vty.Host.Trigger
  ( -- * Bounded event queue
    BoundedEventQueue
  , Batch
  , newBoundedEventQueue
  , closeBoundedEventQueue
  , writeBoundedEventQueue
  , drainBoundedEventQueue

    -- * TriggerEvent transformer
  , BoundedTriggerT
  , runBoundedTriggerT
  ) where

import Control.Concurrent.STM
  ( STM
  , TBQueue
  , TMVar
  , atomically
  , flushTBQueue
  , newEmptyTMVarIO
  , newTBQueueIO
  , orElse
  , readTBQueue
  , readTMVar
  , tryPutTMVar
  , writeTBQueue
  )
import Control.Monad (void)
import Control.Monad.Catch (MonadCatch, MonadMask, MonadThrow)
import Control.Monad.Fix (MonadFix)
import Control.Monad.IO.Class (MonadIO)
import Control.Monad.Morph (MFunctor (hoist))
import Control.Monad.Primitive (PrimMonad (..))
import Control.Monad.Reader (ReaderT, ask, runReaderT)
import Control.Monad.Ref (MonadAtomicRef (..), MonadRef (..), Ref)
import Control.Monad.Trans (MonadTrans, lift)
import Data.Coerce (coerce)
import Data.Dependent.Sum (DSum ((:=>)))
import Numeric.Natural (Natural)
import Reflex
import Reflex.Host.Class (MonadReflexCreateTrigger (..), newEventWithTriggerRef)

-- | A pending batch of trigger invocations. Each external @fire@ enqueues one
-- such batch (almost always a singleton).
type Batch t = [DSum (EventTriggerRef t) TriggerInvocation]

-- | A bounded, closeable queue of pending trigger invocations.
--
-- * Writes ('writeBoundedEventQueue') block when the queue is full.
--   This applies backpressure to the producer.
-- * Once closed, writes return immediately without enqueueing.
--   Producers blocked at shutdown are released.
-- * Reads ('drainBoundedEventQueue') block until at least one batch is
--   available (or the queue is closed), then drain everything
--   currently queued.
data BoundedEventQueue t = BoundedEventQueue
  { _beqQueue :: !(TBQueue (Batch t))
  , _beqClosed :: !(TMVar ())
  -- ^ Full when the queue has been closed.
  }

-- | Create a bounded event queue with the given capacity. Capacity
-- must be at least 1. Smaller values are clamped to 1.
newBoundedEventQueue :: Natural -> IO (BoundedEventQueue t)
newBoundedEventQueue cap =
  BoundedEventQueue
    <$> newTBQueueIO (max 1 cap)
    <*> newEmptyTMVarIO

-- | Close a queue. Idempotent. After this call, blocked and future writers
-- return immediately without enqueueing.
closeBoundedEventQueue :: BoundedEventQueue t -> IO ()
closeBoundedEventQueue q = atomically $ void $ tryPutTMVar (_beqClosed q) ()

-- | Enqueue a batch, blocking if the queue is full. This becomes a
-- no-op once the queue is closed. A producer blocked here then
-- unblocks on shutdown. Closure is checked first, in the same
-- transaction, so a closed queue refuses writes even when it has
-- spare capacity.
writeBoundedEventQueue :: BoundedEventQueue t -> Batch t -> STM ()
writeBoundedEventQueue q batch =
  void (readTMVar (_beqClosed q)) `orElse` writeTBQueue (_beqQueue q) batch

-- | Block until at least one batch is available. Then return it
-- together with every other batch currently queued, as a list of
-- distinct batches (not concatenated). The host fires each batch in
-- its own frame, so multiple occurrences of the same trigger that
-- landed in the same drain are preserved. Reflex collapses
-- simultaneous same-trigger firings; firing them as one batch would
-- drop occurrences. Returns 'Nothing' if the queue is closed and
-- empty.
drainBoundedEventQueue :: BoundedEventQueue t -> IO (Maybe [Batch t])
drainBoundedEventQueue q =
  atomically $
    let readFirst = Just <$> readTBQueue (_beqQueue q)
        waitClosed = Nothing <$ readTMVar (_beqClosed q)
    in do
         mFirst <- readFirst `orElse` waitClosed
         case mFirst of
           Nothing -> return Nothing
           Just first -> do
             rest <- flushTBQueue (_beqQueue q)
             return $ Just (first : rest)

----------------------------------------------------------------------------
-- TriggerEvent transformer
----------------------------------------------------------------------------

-- | A 'TriggerEvent' implementation that writes external triggers into a
-- 'BoundedEventQueue', applying backpressure when the queue is full. It is a
-- direct replacement for @TriggerEventT@ with the same instance set. Only its
-- 'TriggerEvent' instance differs. It routes writes through
-- 'writeBoundedEventQueue'.
newtype BoundedTriggerT t m a = BoundedTriggerT
  { unBoundedTriggerT :: ReaderT (BoundedEventQueue t) m a
  }
  deriving
    ( Applicative
    , Functor
    , Monad
    , MonadCatch
    , MonadFix
    , MonadIO
    , MonadMask
    , MonadThrow
    )

-- | Run a 'BoundedTriggerT' action against a particular 'BoundedEventQueue'.
runBoundedTriggerT :: BoundedTriggerT t m a -> BoundedEventQueue t -> m a
runBoundedTriggerT = runReaderT . unBoundedTriggerT

instance MonadTrans (BoundedTriggerT t) where
  lift = BoundedTriggerT . lift

instance MFunctor (BoundedTriggerT t) where
  hoist f = BoundedTriggerT . hoist f . unBoundedTriggerT

instance PrimMonad m => PrimMonad (BoundedTriggerT t m) where
  type PrimState (BoundedTriggerT t m) = PrimState m
  primitive = lift . primitive

instance PerformEvent t m => PerformEvent t (BoundedTriggerT t m) where
  type Performable (BoundedTriggerT t m) = Performable m
  performEvent_ = lift . performEvent_
  performEvent = lift . performEvent

instance PostBuild t m => PostBuild t (BoundedTriggerT t m) where
  getPostBuild = lift getPostBuild

instance MonadReflexCreateTrigger t m => MonadReflexCreateTrigger t (BoundedTriggerT t m) where
  newEventWithTrigger = lift . newEventWithTrigger
  newFanEventWithTrigger f = lift $ newFanEventWithTrigger f

instance NotReady t m => NotReady t (BoundedTriggerT t m) where
  notReadyUntil = lift . notReadyUntil
  notReady = lift notReady

instance MonadRef m => MonadRef (BoundedTriggerT t m) where
  type Ref (BoundedTriggerT t m) = Ref m
  newRef = lift . newRef
  readRef = lift . readRef
  writeRef r = lift . writeRef r

instance MonadAtomicRef m => MonadAtomicRef (BoundedTriggerT t m) where
  atomicModifyRef r = lift . atomicModifyRef r

instance MonadSample t m => MonadSample t (BoundedTriggerT t m) where
  sample = lift . sample

instance MonadHold t m => MonadHold t (BoundedTriggerT t m) where
  hold v0 v' = lift $ hold v0 v'
  holdDyn v0 v' = lift $ holdDyn v0 v'
  holdIncremental v0 v' = lift $ holdIncremental v0 v'
  buildDynamic a0 = lift . buildDynamic a0
  headE = lift . headE
  now = lift now

instance Adjustable t m => Adjustable t (BoundedTriggerT t m) where
  runWithReplace (BoundedTriggerT a0) a' = BoundedTriggerT $ runWithReplace a0 (coerceEvent a')
  traverseIntMapWithKeyWithAdjust f dm0 dm' =
    BoundedTriggerT $ traverseIntMapWithKeyWithAdjust (coerce . f) dm0 dm'
  traverseDMapWithKeyWithAdjust f dm0 dm' =
    BoundedTriggerT $ traverseDMapWithKeyWithAdjust (coerce . f) dm0 dm'
  traverseDMapWithKeyWithAdjustWithMove f dm0 dm' =
    BoundedTriggerT $ traverseDMapWithKeyWithAdjustWithMove (coerce . f) dm0 dm'

instance
  ( Monad m
  , MonadRef m
  , Ref m ~ Ref IO
  , MonadReflexCreateTrigger t m
  )
  => TriggerEvent t (BoundedTriggerT t m)
  where
  newTriggerEvent = do
    (e, t) <- newTriggerEventWithOnComplete
    return (e, \a -> t a $ return ())
  newTriggerEventWithOnComplete = do
    q <- BoundedTriggerT ask
    (eResult, reResultTrigger) <- lift newEventWithTriggerRef
    return . (,) eResult $ \a cb ->
      atomically $ writeBoundedEventQueue q [EventTriggerRef reResultTrigger :=> TriggerInvocation a cb]
  newEventWithLazyTriggerWithOnComplete f = do
    q <- BoundedTriggerT ask
    lift . newEventWithTrigger $ \t ->
      f $ \a cb -> do
        reResultTrigger <- newRef $ Just t
        atomically $ writeBoundedEventQueue q [EventTriggerRef reResultTrigger :=> TriggerInvocation a cb]