{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE StandaloneKindSignatures #-}
{-# LANGUAGE TypeFamilies #-}

-- | Effectful 'Circuit.Poles.Poles' constructors and queueing strategies.
--
-- This module lives in @circuits-agent@ because it needs @STM@.  Core
-- 'Circuit.Poles' is pure in the base arrow; the queue implementations here
-- are one possible effectful instantiation, placed beside their consumers
-- rather than forcing the core library to depend on @stm@.
module Circuit.Agent.Ends
  ( -- * Queue strategies
    Queue (..),

    -- * Channel policies (mediator-configured buffering)
    ChannelPolicy (..),
    openChannel,
    openChannelSTM,

    -- * Linear default channels
    openLinearChannel,
    openLinearChannelSTM,

    -- * Halt-mark channels (compile-time linearity witness)
    HaltChannel (..),
    IsLinear,
    openHaltChannel,
    writeHaltChannel,
    readHaltChannel,

    -- * STM @Ends@
    openSTM,

    -- * IO @Ends@
    openIO,

    -- * Honest composition
    pipeEnds,
  )
where

import Circuit.Category (K (..))
import Circuit.Poles (HasDual (..), Poles (..), commit, companion, conjoint, emit, open, polesK, splay0)
import Control.Applicative
import Control.Concurrent.Async (async, cancel)
import Control.Concurrent.STM
import Control.Monad (forever, void)
import Data.Kind (Constraint, Type)
import GHC.TypeLits (ErrorMessage (..), TypeError)
import Prelude

-- | How messages are queued between producer and consumer.
data Queue a
  = -- | Unbounded FIFO queue.
    Unbounded
  | -- | Bounded FIFO with backpressure (write blocks when full).
    Bounded Int
  | -- | Single-slot buffer (write blocks when full).
    Single
  | -- | Single-slot buffer, overwrite-on-full.
    -- Write always succeeds; read empties.
    SwapQ
  | -- | Always holds the latest value (overwrites, never blocks).
    Latest a
  | -- | Like @Bounded@ but drops oldest when full.
    Newest Int
  deriving (Int -> Queue a -> ShowS
[Queue a] -> ShowS
Queue a -> String
(Int -> Queue a -> ShowS)
-> (Queue a -> String) -> ([Queue a] -> ShowS) -> Show (Queue a)
forall a. Show a => Int -> Queue a -> ShowS
forall a. Show a => [Queue a] -> ShowS
forall a. Show a => Queue a -> String
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: forall a. Show a => Int -> Queue a -> ShowS
showsPrec :: Int -> Queue a -> ShowS
$cshow :: forall a. Show a => Queue a -> String
show :: Queue a -> String
$cshowList :: forall a. Show a => [Queue a] -> ShowS
showList :: [Queue a] -> ShowS
Show, Queue a -> Queue a -> Bool
(Queue a -> Queue a -> Bool)
-> (Queue a -> Queue a -> Bool) -> Eq (Queue a)
forall a. Eq a => Queue a -> Queue a -> Bool
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: forall a. Eq a => Queue a -> Queue a -> Bool
== :: Queue a -> Queue a -> Bool
$c/= :: forall a. Eq a => Queue a -> Queue a -> Bool
/= :: Queue a -> Queue a -> Bool
Eq)

-- | A channel policy names the residual mediator that governs an effectful
-- channel.  This is the Track-B relocation of the old @Queue@ annotation:
-- the policy is a value passed at allocation time, not a field of the
-- channel type.  The constructors match the ?-modality vocabulary from the
-- B0 spike; 'Linear' is the empty-residual default and the only policy on
-- which halt marks are safe.
data ChannelPolicy a
  = -- | Unbounded FIFO: empty residual, preserves every token in order.
    -- This is the effectful face of a linear process.
    Linear
  | -- | Single-slot buffer with backpressure (write blocks when full).
    SingleSlot
  | -- | Single-slot overwrite: write always succeeds, read empties.
    -- A weakening policy that can drop a halt mark.
    SwapOne
  | -- | Always holds the latest value; requires a seed for the first read.
    -- A weakening policy suitable for diagnostics, not for halt marks.
    LatestValue a
  | -- | Bounded FIFO with backpressure.
    BoundedN Int
  | -- | Bounded FIFO dropping oldest when full.
    -- A weakening policy suitable for bounded diagnostics.
    NewestN Int
  deriving (Int -> ChannelPolicy a -> ShowS
[ChannelPolicy a] -> ShowS
ChannelPolicy a -> String
(Int -> ChannelPolicy a -> ShowS)
-> (ChannelPolicy a -> String)
-> ([ChannelPolicy a] -> ShowS)
-> Show (ChannelPolicy a)
forall a. Show a => Int -> ChannelPolicy a -> ShowS
forall a. Show a => [ChannelPolicy a] -> ShowS
forall a. Show a => ChannelPolicy a -> String
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: forall a. Show a => Int -> ChannelPolicy a -> ShowS
showsPrec :: Int -> ChannelPolicy a -> ShowS
$cshow :: forall a. Show a => ChannelPolicy a -> String
show :: ChannelPolicy a -> String
$cshowList :: forall a. Show a => [ChannelPolicy a] -> ShowS
showList :: [ChannelPolicy a] -> ShowS
Show, ChannelPolicy a -> ChannelPolicy a -> Bool
(ChannelPolicy a -> ChannelPolicy a -> Bool)
-> (ChannelPolicy a -> ChannelPolicy a -> Bool)
-> Eq (ChannelPolicy a)
forall a. Eq a => ChannelPolicy a -> ChannelPolicy a -> Bool
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: forall a. Eq a => ChannelPolicy a -> ChannelPolicy a -> Bool
== :: ChannelPolicy a -> ChannelPolicy a -> Bool
$c/= :: forall a. Eq a => ChannelPolicy a -> ChannelPolicy a -> Bool
/= :: ChannelPolicy a -> ChannelPolicy a -> Bool
Eq)

-- | Convert a channel policy to the concrete queue strategy that implements it.
policyToQueue :: ChannelPolicy a -> Queue a
policyToQueue :: forall a. ChannelPolicy a -> Queue a
policyToQueue = \case
  ChannelPolicy a
Linear -> Queue a
forall a. Queue a
Unbounded
  ChannelPolicy a
SingleSlot -> Queue a
forall a. Queue a
Single
  ChannelPolicy a
SwapOne -> Queue a
forall a. Queue a
SwapQ
  LatestValue a
a -> a -> Queue a
forall a. a -> Queue a
Latest a
a
  BoundedN Int
n -> Int -> Queue a
forall a. Int -> Queue a
Bounded Int
n
  NewestN Int
n -> Int -> Queue a
forall a. Int -> Queue a
Newest Int
n

-- | Open a channel policy as IO @Poles@.
openChannel :: ChannelPolicy a -> IO (Poles (K IO) a a)
openChannel :: forall a. ChannelPolicy a -> IO (Poles (K IO) a a)
openChannel = Queue a -> IO (Poles (K IO) a a)
forall a. Queue a -> IO (Poles (K IO) a a)
openIO (Queue a -> IO (Poles (K IO) a a))
-> (ChannelPolicy a -> Queue a)
-> ChannelPolicy a
-> IO (Poles (K IO) a a)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChannelPolicy a -> Queue a
forall a. ChannelPolicy a -> Queue a
policyToQueue

-- | Open a channel policy as STM @Poles@.
openChannelSTM :: ChannelPolicy a -> STM (Poles (K STM) a a)
openChannelSTM :: forall a. ChannelPolicy a -> STM (Poles (K STM) a a)
openChannelSTM = Queue a -> STM (Poles (K STM) a a)
forall a. Queue a -> STM (Poles (K STM) a a)
openSTM (Queue a -> STM (Poles (K STM) a a))
-> (ChannelPolicy a -> Queue a)
-> ChannelPolicy a
-> STM (Poles (K STM) a a)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ChannelPolicy a -> Queue a
forall a. ChannelPolicy a -> Queue a
policyToQueue

-- | Open a linear channel as IO @Poles@.
--
-- 'Linear' is the default policy: unbounded FIFO, empty residual, preserves
-- every token in order.  This is the effectful face of a linear process.
openLinearChannel :: IO (Poles (K IO) a a)
openLinearChannel :: forall a. IO (Poles (K IO) a a)
openLinearChannel = ChannelPolicy a -> IO (Poles (K IO) a a)
forall a. ChannelPolicy a -> IO (Poles (K IO) a a)
openChannel ChannelPolicy a
forall a. ChannelPolicy a
Linear

-- | Open a linear channel as STM @Poles@.
openLinearChannelSTM :: STM (Poles (K STM) a a)
openLinearChannelSTM :: forall a. STM (Poles (K STM) a a)
openLinearChannelSTM = ChannelPolicy a -> STM (Poles (K STM) a a)
forall a. ChannelPolicy a -> STM (Poles (K STM) a a)
openChannelSTM ChannelPolicy a
forall a. ChannelPolicy a
Linear

-- | Type-level witness that a channel policy is linear.
--
-- Only 'Linear' is allowed to carry halt marks; any other policy produces a
-- compile-time type error.
type family IsLinear (p :: ChannelPolicy a) :: Constraint where
  IsLinear 'Linear = ()
  IsLinear p = TypeError ('Text "only 'Linear' channels can carry halt marks")

-- | A channel statically known to be linear.
--
-- The index @p :: ChannelPolicy a@ is checked by 'IsLinear' at construction
-- time.  Attempting to build a 'HaltChannel' with a non-linear policy fails
-- to typecheck.
type HaltChannel :: ChannelPolicy a -> Type
data HaltChannel p where
  HaltChannel :: (IsLinear p) => Poles (K STM) a a -> HaltChannel (p :: ChannelPolicy a)

-- | Open a halt-mark channel.  This is 'openLinearChannelSTM' with a
-- type-level certificate.
openHaltChannel :: STM (HaltChannel 'Linear)
openHaltChannel :: forall {a}. STM (HaltChannel 'Linear)
openHaltChannel = Poles (K STM) a a -> HaltChannel 'Linear
forall a (p :: ChannelPolicy a).
IsLinear p =>
Poles (K STM) a a -> HaltChannel p
HaltChannel (Poles (K STM) a a -> HaltChannel 'Linear)
-> STM (Poles (K STM) a a) -> STM (HaltChannel 'Linear)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM (Poles (K STM) a a)
forall a. STM (Poles (K STM) a a)
openLinearChannelSTM

-- | Write a token to a halt-mark channel.
writeHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> a -> STM ()
writeHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> a -> STM ()
writeHaltChannel (HaltChannel Poles (K STM) a a
ends) = K STM a () -> a -> STM ()
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (In (K STM) a -> forall x. Out (K STM) x -> K STM a x
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1).
In arr a -> forall (x :: k2). Out arr x -> arr a x
commit (Poles (K STM) a a -> In (K STM) a
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> In arr a
conjoint Poles (K STM) a a
ends) Out (K STM) ()
haltOut)
  where
    haltOut :: Out (K STM) ()
haltOut = Poles (K STM) () () -> Out (K STM) ()
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> Out arr b
companion (Poles (K STM) () ()
unitEndsSTM :: Poles (K STM) () ())

-- | Read a token from a halt-mark channel.
readHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> STM a
readHaltChannel :: forall a (p :: ChannelPolicy a). HaltChannel p -> STM a
readHaltChannel (HaltChannel Poles (K STM) a a
ends) = K STM () a -> () -> STM a
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (Out (K STM) a -> forall x. In (K STM) x -> K STM x a
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k2).
Out arr a -> forall (x :: k1). In arr x -> arr x a
emit (Poles (K STM) a a -> Out (K STM) a
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> Out arr b
companion Poles (K STM) a a
ends) In (K STM) ()
haltIn) ()
  where
    haltIn :: In (K STM) ()
haltIn = Poles (K STM) () () -> In (K STM) ()
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> In arr a
conjoint (Poles (K STM) () ()
unitEndsSTM :: Poles (K STM) () ())

-- | Unit ends specialised to 'K STM'.
unitEndsSTM :: Poles (K STM) () ()
unitEndsSTM :: Poles (K STM) () ()
unitEndsSTM = (() -> STM ()) -> STM () -> Poles (K STM) () ()
forall (m :: * -> *) a b.
Monad m =>
(a -> m ()) -> m b -> Poles (K m) a b
polesK (STM () -> () -> STM ()
forall a b. a -> b -> a
const (() -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())) (() -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())

-- | Internal STM primitive for a queue strategy.
--
-- Returns the raw write/read actions used by 'openSTM'.  Not exported;
-- the canonical API is 'openSTM'.
endsSTM :: Queue a -> STM (a -> STM (), STM a)
endsSTM :: forall a. Queue a -> STM (a -> STM (), STM a)
endsSTM = \case
  Bounded Int
n -> do
    q <- Natural -> STM (TBQueue a)
forall a. Natural -> STM (TBQueue a)
newTBQueue (Int -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
n)
    pure (writeTBQueue q, readTBQueue q)
  Queue a
Unbounded -> do
    q <- STM (TQueue a)
forall a. STM (TQueue a)
newTQueue
    pure (writeTQueue q, readTQueue q)
  Queue a
Single -> do
    m <- STM (TMVar a)
forall a. STM (TMVar a)
newEmptyTMVar
    pure (putTMVar m, takeTMVar m)
  Queue a
SwapQ -> do
    v <- STM (TMVar a)
forall a. STM (TMVar a)
newEmptyTMVar
    let write a
x = TMVar a -> a -> STM Bool
forall a. TMVar a -> a -> STM Bool
tryPutTMVar TMVar a
v a
x STM Bool -> (Bool -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case Bool
True -> () -> STM ()
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (); Bool
False -> STM a -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (TMVar a -> a -> STM a
forall a. TMVar a -> a -> STM a
swapTMVar TMVar a
v a
x)
    pure (write, takeTMVar v)
  Latest a
a -> do
    t <- a -> STM (TVar a)
forall a. a -> STM (TVar a)
newTVar a
a
    pure (writeTVar t, readTVar t)
  Newest Int
n -> do
    q <- Natural -> STM (TBQueue a)
forall a. Natural -> STM (TBQueue a)
newTBQueue (Int -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
n)
    let write a
x = TBQueue a -> a -> STM ()
forall a. TBQueue a -> a -> STM ()
writeTBQueue TBQueue a
q a
x STM () -> STM () -> STM ()
forall a. STM a -> STM a -> STM a
forall (f :: * -> *) a. Alternative f => f a -> f a -> f a
<|> (TBQueue a -> STM (Maybe a)
forall a. TBQueue a -> STM (Maybe a)
tryReadTBQueue TBQueue a
q STM (Maybe a) -> STM () -> STM ()
forall a b. STM a -> STM b -> STM b
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f b
*> a -> STM ()
write a
x)
    pure (write, readTBQueue q)

-- | Open a queue strategy as STM @Poles@.
--
-- Allocates STM primitives and returns a matched pair of ends sharing
-- the same mutable channel.  Both ends live in 'STM', so you can compose
-- operations across channels in a single 'atomically' block.
openSTM :: Queue a -> STM (Poles (K STM) a a)
openSTM :: forall a. Queue a -> STM (Poles (K STM) a a)
openSTM Queue a
q = do
  (write, read') <- Queue a -> STM (a -> STM (), STM a)
forall a. Queue a -> STM (a -> STM (), STM a)
endsSTM Queue a
q
  pure (polesK write read')

-- | Open a queue strategy as IO @Poles@.
--
-- Like 'openSTM', but each primitive operation is wrapped in its own
-- 'atomically'.  You cannot batch multiple writes or a write-plus-read
-- into a single STM transaction; for that use 'openSTM' and wrap in
-- 'atomically' yourself.
openIO :: Queue a -> IO (Poles (K IO) a a)
openIO :: forall a. Queue a -> IO (Poles (K IO) a a)
openIO Queue a
q = do
  e <- STM (Poles (K STM) a a) -> IO (Poles (K STM) a a)
forall a. STM a -> IO a
atomically (Queue a -> STM (Poles (K STM) a a)
forall a. Queue a -> STM (Poles (K STM) a a)
openSTM Queue a
q)
  let (K write, K receive) = splay0 e
  pure (polesK (atomically . write) (atomically (receive ())))

-- | Honest sequential composition of two allocated ends via an intermediate
-- queue and a pump.
--
-- @pipeEnds e1 makeE2@ allocates a queue of @b@ values, builds the right end
-- around that queue with @makeE2@, and starts a pump that moves values from
-- @e1@ into the right end.  The returned 'Poles' uses @e1@ for input and the
-- built right end for output; the close action cancels the pump.
--
-- This is the coend-style composition that 'composePoles' cannot express: the
-- intermediate carrier is a real queue (the residual's home) rather than the
-- unit type, so a multi-read consumer can accumulate inputs before emitting.
pipeEnds ::
  forall a b c.
  Poles (K IO) a b ->
  (TQueue b -> IO (Poles (K IO) b c)) ->
  IO (Poles (K IO) a c, IO ())
pipeEnds :: forall a b c.
Poles (K IO) a b
-> (TQueue b -> IO (Poles (K IO) b c))
-> IO (Poles (K IO) a c, IO ())
pipeEnds Poles (K IO) a b
e1 TQueue b -> IO (Poles (K IO) b c)
makeE2 = do
  q <- IO (TQueue b)
forall a. IO (TQueue a)
newTQueueIO
  e2 <- makeE2 q
  let unitEnds :: Poles (K IO) () ()
      unitEnds = Poles (K IO) () ()
forall {k} (bot :: k) (arr :: k -> k -> *).
HasDual bot arr =>
Poles arr bot bot
open
      readFromE1 :: IO b
      readFromE1 = K IO () b -> () -> IO b
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (Out (K IO) b -> forall x. In (K IO) x -> K IO x b
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k2).
Out arr a -> forall (x :: k1). In arr x -> arr x a
emit (Poles (K IO) a b -> Out (K IO) b
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> Out arr b
companion Poles (K IO) a b
e1) (Poles (K IO) () () -> In (K IO) ()
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> In arr a
conjoint Poles (K IO) () ()
unitEnds)) ()
      writeToE2 :: b -> IO ()
      writeToE2 = K IO b () -> b -> IO ()
forall {k} (m :: k -> *) a (b :: k). K m a b -> a -> m b
runK (In (K IO) b -> forall x. Out (K IO) x -> K IO b x
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1).
In arr a -> forall (x :: k2). Out arr x -> arr a x
commit (Poles (K IO) b c -> In (K IO) b
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> In arr a
conjoint Poles (K IO) b c
e2) (Poles (K IO) () () -> Out (K IO) ()
forall {k1} {k2} (arr :: k1 -> k2 -> *) (a :: k1) (b :: k2).
Poles arr a b -> Out arr b
companion Poles (K IO) () ()
unitEnds))
  pump <- async . forever $ do
    x <- readFromE1
    writeToE2 x
  pure (Poles (conjoint e1) (companion e2), cancel pump)