@@ -1,85 +1,41 @@
1−{-# LANGUAGE RankNTypes #-}
2−{-# LANGUAGE ScopedTypeVariables #-}
1+{-# LANGUAGE GADTs #-}
2+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
3+{-# LANGUAGE LambdaCase #-}
34{-# LANGUAGE OverloadedStrings #-}
5+{-# LANGUAGE PatternSynonyms #-}
46{-# LANGUAGE QuasiQuotes #-}
7+{-# LANGUAGE RankNTypes #-}
58{-# LANGUAGE RecursiveDo #-}
6−{-# LANGUAGE GADTs #-}
7−{-# LANGUAGE PatternSynonyms #-}
8−{-# LANGUAGE LambdaCase #-}
9+{-# LANGUAGE ScopedTypeVariables #-}
910module Obelisk.Api where
1011
1112import Prelude hiding ((.))
1213
14+import Obelisk.Api.Pipeline
15+import Obelisk.Db
1316import Obelisk.Postgres.Replication
1417import Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
15−import Obelisk.Db
18+import Obelisk.Request (Request)
1619
1720import Control.Category
18−import Control.Concurrent (forkIO, killThread)
1921import Control.Concurrent.Async
20−import Control.Concurrent.STM (TChan, atomically, dupTChan, newBroadcastTChanIO, readTChan, writeTChan, newTChanIO)
21−import Control.Monad (forever, void)
22+import Control.Concurrent.STM (TChan, atomically, readTChan, writeTChan, newTChanIO)
23+import Control.Monad (forever)
2224import Control.Monad.Reader
23−import Control.Monad.State.Strict
24−import Control.Exception
25−import Control.Exception.Enclosed
26−import Data.Binary (Binary)
27−import qualified Data.Binary as Binary
28−import Data.Binary.Get (ByteOffset)
29−import Data.Aeson (FromJSON, ToJSON, toJSON)
30−import qualified Data.Aeson as Aeson
31−import qualified Data.ByteString as BS
32−import qualified Data.ByteString.Lazy as LBS
33−import Data.List.Split (wordsBy)
34−import Data.Functor.Identity (Identity (..))
35−import Data.Pool (Pool, withResource)
36−import Data.Semigroup ((<>))
37−import Data.String (fromString)
38−import qualified Data.Text as T
39−import Data.Text.Encoding (encodeUtf8, decodeUtf8)
40−import qualified Database.PostgreSQL.Simple as PG
41−import qualified Database.PostgreSQL.Simple.Transaction as PG
42−import Database.PostgreSQL.Simple.SqlQQ
43−import qualified Database.PostgreSQL.Simple.Types as PG
44−import qualified Database.PostgreSQL.Simple.Notification as PG
45−import GHC.Generics (Generic)
46−import Control.Category (Category)
47−import qualified Control.Category as Cat
48−import Control.Concurrent (forkIO, killThread)
49−import Control.Exception (bracket)
50−import Control.Lens (imapM_)
51−import Control.Monad (forever, void, when)
52−import Control.Monad.IO.Class (MonadIO, liftIO)
53−import Control.Monad.State.Strict (evalStateT, get, put)
5425import Control.Monad.Trans (lift)
55−import Data.Aeson (FromJSON, toJSON)
56−import Data.AppendMap (AppendMap)
57−import qualified Data.AppendMap as Map
58−import Data.Map.Monoidal (MonoidalMap)
59−import Data.Foldable (fold)
60−import Data.IORef (atomicModifyIORef', newIORef, readIORef)
61−import Data.MonoidMap (MonoidMap (..), monoidMap)
62−import Data.Pool (Pool)
63−import Data.Semigroup (Semigroup, (<>))
64−import Data.Text (Text)
65−import Data.Typeable (Typeable)
66−import Debug.Trace (trace)
67−import Reflex (FunctorMaybe (..))
68−import Reflex.Patch (Group, negateG, (~~))
69−import Reflex.Query.Base (mapQuery, mapQueryResult)
70−import Reflex.Query.Class (Query, QueryResult, QueryMorphism (..), SelectedCount (..), crop)
71−import Snap.Core (MonadSnap, Snap)
72−import qualified Web.ClientSession as CS
26+import Data.Aeson (FromJSON, ToJSON)
27+import Data.Attoparsec.ByteString
7328import Data.ByteString (ByteString)
74−import Data.Int
75−import Data.Word
76−import Data.Char
7729import Data.Default
78−import Data.Attoparsec
79−
80−import Rhyolite.Backend.DB.PsqlSimple hiding (Binary)
81−import Obelisk.Api.Pipeline
82−--import Rhyolite.Backend.App
30+import Data.Map.Monoidal (MonoidalMap)
31+import Data.MonoidMap (MonoidMap (..))
32+import Data.Pool (withResource)
33+import qualified Database.PostgreSQL.Simple as PG
34+import qualified Database.PostgreSQL.Simple.Transaction as PG
35+import Reflex (FunctorMaybe (..))
36+import Reflex.Patch (Group)
37+import Reflex.Query.Class (QueryResult, SelectedCount (..))
38+import Snap.Core (Snap)
8339
8440{-
8541-- | A way for a pipeline to retrieve data
@@ -132,6 +88,7 @@ servePostgres db handleApi handleNotify handleQuery pipe = do
13288-}
13389
13490newtype ReadDb a = ReadDb { unReadDb :: ReaderT PG.Connection IO a }
91+ deriving (Functor, Applicative, Monad)
13592
13693readTransaction :: PG.Connection -> ReadDb a -> IO a
13794readTransaction conn (ReadDb r) = PG.withTransactionModeRetry m PG.isSerializationError conn $ runReaderT r conn
@@ -141,6 +98,7 @@ readTransaction conn (ReadDb r) = PG.withTransactionModeRetry m PG.isSerializati
14198 }
14299
143100newtype WriteDb a = WriteDb { unWriteDb :: ReaderT PG.Connection IO a }
101+ deriving (Functor, Applicative, Monad)
144102
145103writeTransaction :: PG.Connection -> WriteDb a -> IO a
146104writeTransaction conn (WriteDb r) = PG.withTransactionModeRetry m PG.isSerializationError conn $ runReaderT r conn
@@ -156,7 +114,7 @@ unsafeWriteDb :: (PG.Connection -> IO a) -> WriteDb a
156114unsafeWriteDb f = WriteDb $ lift . f =<< ask
157115
158116withApiHandler
159− :: forall query queryResult request change a.
117+ :: forall query queryResult request a.
160118 ( Monoid (query SelectedCount)
161119 , QueryResult (query (MonoidMap ClientKey SelectedCount)) ~ queryResult (MonoidMap ClientKey SelectedCount)
162120 , QueryResult (query SelectedCount) ~ queryResult SelectedCount
@@ -182,9 +140,6 @@ withApiHandler
182140 -> (Snap () -> IO a)
183141 -> IO a
184142withApiHandler dbUri handleApi handleChange handleQuery go = do
185− let opts = def
186− { _logicalDecodingOptions_pluginOptions = [("skip-empty-xacts", Nothing)]
187− }
188143 withNonEmptyTransactions dbUri $ \transactions -> do
189144 withConnectionPool dbUri $ \db -> do
190145 let onChange :: Transaction -> query x -> IO (QueryResult (query x))
@@ -193,7 +148,7 @@ withApiHandler dbUri handleApi handleChange handleQuery go = do
193148 pipe = queryMorphismPipeline $ transposeMonoidMap . monoidMapQueryMorphism
194149 rec (qh, finalizeFeed) <- feedPipeline (onChange <$> atomically (readTChan transactions)) (QueryHandler $ \q -> withResource db $ \conn -> readTransaction conn $ handleQuery q) r
195150 (qh', r) <- unPipeline pipe qh r'
196− (r', handleListen) <- connectPipelineToWebsockets "" (\r -> withResource db $ \conn -> writeTransaction conn $ handleApi r) qh'
151+ (r', handleListen) <- connectPipelineToWebsockets "" (\r'' -> withResource db $ \conn -> writeTransaction conn $ handleApi r'') qh'
197152 go handleListen
198153
199154withNonEmptyTransactions :: ByteString -> (TChan Transaction -> IO a) -> IO a
@@ -201,11 +156,11 @@ withNonEmptyTransactions dbUri go = do
201156 let opts = def
202157 { _logicalDecodingOptions_pluginOptions = [("skip-empty-xacts", Nothing)]
203158 }
204− withLogicalDecoding dbUri "test_decoding" opts $ \lines -> do
159+ withLogicalDecoding dbUri "test_decoding" opts $ \decodedLines -> do
205160 transactions <- newTChanIO
206161 processLine <- linesToTransactions
207162 let processLines = forever $ do
208− lineRaw <- atomically $ readTChan lines
163+ lineRaw <- atomically $ readTChan decodedLines
209164 let Right l = parseOnly (line <* endOfInput) lineRaw
210165 Right mTransaction <- processLine l
211166 forM_ mTransaction $ \transaction ->