{-# LANGUAGE GADTs #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TypeFamilies #-}

-- | Inspectable shard pipelines over addressed posts.
--
-- A 'Pipeline' is a reified sequence of pure list-transformer stages.  Each
-- stage consumes a stream of @a@ and produces a stream of @b@; the whole
-- pipeline folds into a single pure function and is then wrapped as a stateful
-- 'Circuit.Agent.Shard'.
module Free.Agent.Pipeline
  ( Pipeline (..),
    runPipeline,
    pipelineShard,
    filterP,
    mapP,
    routeP,
    routeTo,
    routeBy,
    broadcast,
    forName,
    fromName,
  )
where

import Circuit (Body (..))
import Circuit.Agent (Name, Post (..), deliversTo)
import Circuit.Category (Category (..), K (..))
import Circuit.Poles (Poles (..), poles0)
import Data.Text (Text)
import Prelude hiding (id, (.))

-- $setup
-- >>> :set -XOverloadedStrings
-- >>> import Free.Agent.Pipeline
-- >>> import Circuit.Agent

-- | A pipeline stage or composition of stages.
data Pipeline a b where
  -- | Keep only inputs that satisfy the predicate.
  Filter :: (a -> Bool) -> Pipeline a a
  -- | Transform each input.
  Map :: (a -> b) -> Pipeline a b
  -- | Expand each input into zero or more outputs.
  Route :: (a -> [b]) -> Pipeline a b
  -- | Sequence two pipelines.
  Compose :: Pipeline b c -> Pipeline a b -> Pipeline a c

-- | Pipelines form a category: 'id' is @'Map' id@; composition is 'Compose'.
instance Category Pipeline where
  id :: forall a. Pipeline a a
id = (a -> a) -> Pipeline a a
forall a b. (a -> b) -> Pipeline a b
Map a -> a
forall a. a -> a
forall k (arr :: k -> k -> *) (a :: k). Category arr => arr a a
id
  . :: forall b c a. Pipeline b c -> Pipeline a b -> Pipeline a c
(.) = Pipeline b c -> Pipeline a b -> Pipeline a c
forall b c a. Pipeline b c -> Pipeline a b -> Pipeline a c
Compose

-- | Keep only inputs that satisfy the predicate.
filterP :: (a -> Bool) -> Pipeline a a
filterP :: forall a. (a -> Bool) -> Pipeline a a
filterP = (a -> Bool) -> Pipeline a a
forall a. (a -> Bool) -> Pipeline a a
Filter

-- | Transform each input.
mapP :: (a -> b) -> Pipeline a b
mapP :: forall a b. (a -> b) -> Pipeline a b
mapP = (a -> b) -> Pipeline a b
forall a b. (a -> b) -> Pipeline a b
Map

-- | Expand each input into zero or more outputs.
routeP :: (a -> [b]) -> Pipeline a b
routeP :: forall a b. (a -> [b]) -> Pipeline a b
routeP = (a -> [b]) -> Pipeline a b
forall a b. (a -> [b]) -> Pipeline a b
Route

-- | Route every post to a single recipient.
routeTo :: Name -> Pipeline (Post Text) (Post Text)
routeTo :: Name -> Pipeline (Post Name) (Post Name)
routeTo Name
name = (Post Name -> Post Name) -> Pipeline (Post Name) (Post Name)
forall a b. (a -> b) -> Pipeline a b
Map (\Post Name
p -> Post Name
p {to = [name]})

-- | Route posts using a function from the post to a recipient list.
routeBy :: (Post Text -> [Name]) -> Pipeline (Post Text) (Post Text)
routeBy :: (Post Name -> [Name]) -> Pipeline (Post Name) (Post Name)
routeBy Post Name -> [Name]
f = (Post Name -> Post Name) -> Pipeline (Post Name) (Post Name)
forall a b. (a -> b) -> Pipeline a b
Map (\Post Name
p -> Post Name
p {to = f p})

-- | Broadcast every post to a list of recipients.
broadcast :: [Name] -> Pipeline (Post Text) (Post Text)
broadcast :: [Name] -> Pipeline (Post Name) (Post Name)
broadcast [Name]
names = (Post Name -> Post Name) -> Pipeline (Post Name) (Post Name)
forall a b. (a -> b) -> Pipeline a b
Map (\Post Name
p -> Post Name
p {to = names})

-- | Keep posts addressed to @name@ (via 'deliversTo').
forName :: Name -> Pipeline (Post Text) (Post Text)
forName :: Name -> Pipeline (Post Name) (Post Name)
forName Name
name = (Post Name -> Bool) -> Pipeline (Post Name) (Post Name)
forall a. (a -> Bool) -> Pipeline a a
Filter (Post Name -> [Name] -> Bool
forall a. Post a -> [Name] -> Bool
`deliversTo` [Name
name])

-- | Keep posts whose sender is @name@.
fromName :: Name -> Pipeline (Post Text) (Post Text)
fromName :: Name -> Pipeline (Post Name) (Post Name)
fromName Name
name = (Post Name -> Bool) -> Pipeline (Post Name) (Post Name)
forall a. (a -> Bool) -> Pipeline a a
Filter (\Post Name
p -> Post Name -> Name
forall a. Post a -> Name
from Post Name
p Name -> Name -> Bool
forall a. Eq a => a -> a -> Bool
== Name
name)

-- | Fold a pipeline into a pure list function.
runPipeline :: Pipeline a b -> ([a] -> [b])
runPipeline :: forall a b. Pipeline a b -> [a] -> [b]
runPipeline (Filter a -> Bool
p) = (a -> Bool) -> [a] -> [a]
forall a. (a -> Bool) -> [a] -> [a]
filter a -> Bool
p
runPipeline (Map a -> b
f) = (a -> b) -> [a] -> [b]
forall a b. (a -> b) -> [a] -> [b]
map a -> b
f
runPipeline (Route a -> [b]
f) = (a -> [b]) -> [a] -> [b]
forall (t :: * -> *) a b. Foldable t => (a -> [b]) -> t a -> [b]
concatMap a -> [b]
f
runPipeline (Compose Pipeline b b
g Pipeline a b
f) = Pipeline b b -> [b] -> [b]
forall a b. Pipeline a b -> [a] -> [b]
runPipeline Pipeline b b
g ([b] -> [b]) -> ([a] -> [b]) -> [a] -> [b]
forall b c a. (b -> c) -> (a -> b) -> a -> c
forall k (arr :: k -> k -> *) (b :: k) (c :: k) (a :: k).
Category arr =>
arr b c -> arr a b -> arr a c
. Pipeline a b -> [a] -> [b]
forall a b. Pipeline a b -> [a] -> [b]
runPipeline Pipeline a b
f

-- | Run a pipeline as a closed stateful shard.
--
-- The state holds the pending input batch.  Commit replaces it; emit applies
-- the pipeline and clears the buffer.
pipelineShard :: Pipeline a b -> Poles (Body (,) [a] (K IO)) [a] [b]
pipelineShard :: forall a b. Pipeline a b -> Poles (Body (,) [a] (K IO)) [a] [b]
pipelineShard Pipeline a b
p = Body (,) [a] (K IO) [a] ()
-> Body (,) [a] (K IO) () [b]
-> Poles (Body (,) [a] (K IO)) [a] [b]
forall (arr :: * -> * -> *) a b.
HasDual () arr =>
arr a () -> arr () b -> Poles arr a b
poles0 Body (,) [a] (K IO) [a] ()
forall {ch}. Body (,) ch (K IO) ch ()
writeBatch Body (,) [a] (K IO) () [b]
readBatch
  where
    writeBatch :: Body (,) ch (K IO) ch ()
writeBatch = K IO (ch, ch) (ch, ()) -> Body (,) ch (K IO) ch ()
forall {k} {k1} {k2} (t :: k -> k1 -> k2) (ch :: k)
       (arr :: k2 -> k2 -> *) (a :: k1) (b :: k1).
arr (t ch a) (t ch b) -> Body t ch arr a b
Body (K IO (ch, ch) (ch, ()) -> Body (,) ch (K IO) ch ())
-> K IO (ch, ch) (ch, ()) -> Body (,) ch (K IO) ch ()
forall a b. (a -> b) -> a -> b
$ ((ch, ch) -> IO (ch, ())) -> K IO (ch, ch) (ch, ())
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K (((ch, ch) -> IO (ch, ())) -> K IO (ch, ch) (ch, ()))
-> ((ch, ch) -> IO (ch, ())) -> K IO (ch, ch) (ch, ())
forall a b. (a -> b) -> a -> b
$ \(ch
_, ch
xs) -> (ch, ()) -> IO (ch, ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (ch
xs, ())
    readBatch :: Body (,) [a] (K IO) () [b]
readBatch = K IO ([a], ()) ([a], [b]) -> Body (,) [a] (K IO) () [b]
forall {k} {k1} {k2} (t :: k -> k1 -> k2) (ch :: k)
       (arr :: k2 -> k2 -> *) (a :: k1) (b :: k1).
arr (t ch a) (t ch b) -> Body t ch arr a b
Body (K IO ([a], ()) ([a], [b]) -> Body (,) [a] (K IO) () [b])
-> K IO ([a], ()) ([a], [b]) -> Body (,) [a] (K IO) () [b]
forall a b. (a -> b) -> a -> b
$ (([a], ()) -> IO ([a], [b])) -> K IO ([a], ()) ([a], [b])
forall {k} (m :: k -> *) a (b :: k). (a -> m b) -> K m a b
K ((([a], ()) -> IO ([a], [b])) -> K IO ([a], ()) ([a], [b]))
-> (([a], ()) -> IO ([a], [b])) -> K IO ([a], ()) ([a], [b])
forall a b. (a -> b) -> a -> b
$ \([a]
s, ()) -> ([a], [b]) -> IO ([a], [b])
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([], Pipeline a b -> [a] -> [b]
forall a b. Pipeline a b -> [a] -> [b]
runPipeline Pipeline a b
p [a]
s)