@@ -1,3 +1,5 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+{-# LANGUAGE StandaloneDeriving #-}
13{-# LANGUAGE GADTs #-}
24{-# LANGUAGE GeneralizedNewtypeDeriving #-}
35{-# LANGUAGE LambdaCase #-}
@@ -24,69 +26,20 @@ import Control.Monad (forever)
2426import Control.Monad.Reader
2527import Control.Monad.Trans (lift)
2628import Data.Aeson (FromJSON, ToJSON)
29+import Data.Align
2730import Data.Attoparsec.ByteString
2831import Data.ByteString (ByteString)
2932import Data.Default
30−import Data.Map.Monoidal (MonoidalMap)
31−import Data.MonoidMap (MonoidMap (..))
33+import Data.Functor.Compose
34+import Data.Map.Monoidal (MonoidalMap (..))
3235import Data.Pool (withResource)
36+import Data.Vessel
3337import qualified Database.PostgreSQL.Simple as PG
3438import qualified Database.PostgreSQL.Simple.Transaction as PG
35−import Reflex (FunctorMaybe (..))
3639import Reflex.Patch (Group)
3740import Reflex.Query.Class (QueryResult, SelectedCount (..))
3841import Snap.Core (Snap)
3942
40−{-
41−-- | A way for a pipeline to retrieve data
42−newtype QueryHandler q m = QueryHandler
43− { runQueryHandler :: q -> m (QueryResult q) }
44−
45−-- | A way to push data into a pipeline
46−newtype Recipient q m = Recipient
47− { tellRecipient :: QueryResult q -> m () }
48−
49−-- | A way of attaching to (and later detaching from) a pipeline
50−newtype Registrar q = Registrar { unRegistrar :: Recipient q IO -> IO (QueryHandler q IO, IO ()) }
51−
52−-- | A way of connecting a source of data to a consumer of data. The consumer
53−-- can pull data from the datasource and the datasource can push data to the
54−-- consumer.
55−--
56−-- q is the consumer side
57−-- q' is the datasource side
58−newtype Pipeline m q q' = Pipeline { unPipeline :: QueryHandler q' m -> Recipient q m -> IO (QueryHandler q m, Recipient q' m) }
59−
60−instance Category (Pipeline m) where
61− id = Pipeline $ \qh r -> return (qh, r)
62− Pipeline f . Pipeline g = Pipeline $ \qh r -> do
63− rec (qh'', r') <- g qh' r
64− (qh', r'') <- f qh r'
65− return (qh'', r'')
66−
67−-- | A key used to track particular consumers
68−newtype ClientKey = ClientKey { unClientKey :: Integer }
69− deriving (Eq, Ord, Show)
70−
71−servePostgres
72− :: ( Monoid q'
73− , Semigroup q'
74− )
75− => Pool Postgresql -- ^ The DB to connect to
76− -> (forall a. request a -> write a)
77− -> (ByteString -> q' -> read (QueryResult q'))
78− -> (q' -> read (QueryResult q))
79− -> Pipeline IO q q'
80− -> (Snap () -> IO a)
81− -> IO a
82−servePostgres db handleApi handleNotify handleQuery pipe = do
83− (getNextNotification, finalizeListener) <- startNotificationListener db
84− rec (qh, finalizeFeed) <- feedPipeline (handleNotify <$> getNextNotification) handleQuery r
85− (qh', r) <- unPipeline pipe qh r'
86− (r', handleListen) <- connectPipelineToWebsockets "" (RequestHandler handleApi) qh'
87− return (handleListen, finalizeFeed >> finalizeListener)
88−-}
89−
9043newtype ReadDb a = ReadDb { unReadDb :: ReaderT PG.Connection IO a }
9144 deriving (Functor, Applicative, Monad)
9245
@@ -114,43 +67,39 @@ unsafeWriteDb :: (PG.Connection -> IO a) -> WriteDb a
11467unsafeWriteDb f = WriteDb $ lift . f =<< ask
11568
11669withApiHandler
117− :: forall query queryResult request a.
118− ( Monoid (query SelectedCount)
119− , QueryResult (query (MonoidMap ClientKey SelectedCount)) ~ queryResult (MonoidMap ClientKey SelectedCount)
120− , QueryResult (query SelectedCount) ~ queryResult SelectedCount
121− , QueryResult (query ()) ~ queryResult ()
122− , Foldable queryResult
123− , Functor query
124− , Functor queryResult
125− , FunctorMaybe queryResult
126− , Monoid (query (MonoidMap ClientKey SelectedCount))
127− , Monoid (queryResult SelectedCount)
128− , Eq (queryResult SelectedCount)
129− , Eq (query (MonoidMap ClientKey SelectedCount))
130− , Eq (query SelectedCount)
131− , Group (query SelectedCount)
132− , ToJSON (queryResult ())
133− , FromJSON (query ())
70+ :: forall v query q0 request a.
71+ ( View v
72+ , Group (v (Const SelectedCount))
73+ , query ~ v (Compose (MonoidalMap ClientKey) (Const SelectedCount))
74+ , q0 ~ v Proxy
75+ , QueryResult query ~ v (Compose (MonoidalMap ClientKey) Identity)
76+ , QueryResult (v (Const SelectedCount)) ~ v Identity
77+ , Monoid (v (Compose (MonoidalMap ClientKey) (Const SelectedCount)))
78+ , Monoid (v Identity)
79+ , ToJSON (v Identity)
80+ , FromJSON (v Proxy)
13481 , Request request
13582 )
13683 => ByteString -- ^ The URI of the DB to connect to
13784 -> (forall x. request x -> WriteDb x) -- ^ How to handle an API request
138− -> (forall x. query x -> Transaction -> ReadDb (QueryResult (query x))) -- ^ How to handle a change
139− -> (forall x. query x -> ReadDb (QueryResult (query x))) -- ^ How to handle a query
85+ -> (v Proxy -> Transaction -> ReadDb (v Identity)) -- ^ How to handle a change
86+ -> (v Proxy -> ReadDb (v Identity)) -- ^ How to handle a query
14087 -> (Snap () -> IO a)
14188 -> IO a
14289withApiHandler dbUri handleApi handleChange handleQuery go = do
14390 withNonEmptyTransactions dbUri $ \transactions -> do
14491 withConnectionPool dbUri $ \db -> do
145− let onChange :: Transaction -> query x -> IO (QueryResult (query x))
146− onChange m q = withResource db $ \conn -> readTransaction conn $ handleChange q m
147− let pipe :: Pipeline IO (MonoidalMap ClientKey (query SelectedCount)) (query (MonoidMap ClientKey SelectedCount))
148− pipe = queryMorphismPipeline $ transposeMonoidMap . monoidMapQueryMorphism
149− rec (qh, finalizeFeed) <- feedPipeline (onChange <$> atomically (readTChan transactions)) (QueryHandler $ \q -> withResource db $ \conn -> readTransaction conn $ handleQuery q) r
92+ let onChange :: Transaction -> v (Compose (MonoidalMap ClientKey) (Const SelectedCount)) -> IO (v (Compose (MonoidalMap ClientKey) Identity))
93+ onChange txn q = withResource db $ \conn -> readTransaction conn $ mapDecomposedV (flip handleChange txn) q
94+ let pipe :: Pipeline IO (MonoidalMap ClientKey (v (Const SelectedCount))) (v (Compose (MonoidalMap ClientKey) (Const SelectedCount)))
95+ pipe = queryMorphismPipeline transposeView
96+ rec (qh, finalizeFeed) <- feedPipeline (onChange <$> atomically (readTChan transactions)) (QueryHandler $ \q -> withResource db $ \conn -> readTransaction conn $ mapDecomposedV handleQuery q) r
15097 (qh', r) <- unPipeline pipe qh r'
15198 (r', handleListen) <- connectPipelineToWebsockets "" (\r'' -> withResource db $ \conn -> writeTransaction conn $ handleApi r'') qh'
15299 go handleListen
153100
101+deriving instance Ord k => Align (MonoidalMap k)
102+
154103withNonEmptyTransactions :: ByteString -> (TChan Transaction -> IO a) -> IO a
155104withNonEmptyTransactions dbUri go = do
156105 let opts = def
@@ -167,21 +116,3 @@ withNonEmptyTransactions dbUri go = do
167116 atomically $ writeTChan transactions transaction
168117 withAsync processLines $ \a -> link a >> do
169118 go transactions
170−{-
171−
172−standardChangeChannel :: PG.Identifier
173−standardChangeChannel = "obelisk_api_updates"
174−
175−standardChangeSubscriberTable :: PG.Identifier
176−standardChangeSubscriberTable = "obelisk_api_updates_subscriber"
177−
178−standardChangeSpillTable :: PG.Identifier
179−standardChangeSpillTable = "obelisk_api_updates_spill"
180−
181−-- | Listen on a Postgres notification channel. Uses a connection from the Pool for as long as it is r-- | Listen for Postgresql changes on a particular channel
182−withChangeListener :: FromJSON change => Pool PG.Connection -> (change -> IO ()) -> IO a -> IO a
183−withChangeListener db onChange = withBigNotificationListener db standardChangeChannel standardChangeSpillTable $ \message ->
184− case Aeson.eitherDecode' $ LBS.fromStrict message of
185− Left e -> fail $ "withChangeListener: " <> e
186− Right change -> onChange change
187−-}