git.iko.soy/obelisk
on 2cea7df help sign in
7ad33cb
swap
4 ahead, 1276 behind
default.nix+51−1
since splitsnapshots
unifiedsplit
@@ -16,6 +16,20 @@ let
16 openssh
16 openssh
17 ];
17 ];
18
18
19+ beam-src = pkgs.fetchFromGitHub {
20+ owner = "obsidiansystems";
21+ repo = "beam";
22+ rev = "6c9b08d57951083c1f7e4bd49d941619603127f4";
23+ sha256 = "1k4sdz0dxjafwkwlpfcdxkgvvv591nfwc97g1fq23wrs271jr2sm";
24+ };
25+
26+ gargoyle-src = pkgs.fetchFromGitHub {
27+ owner = "obsidiansystems";
28+ repo = "gargoyle";
29+ rev = "2c19c569325ad76694526e9b688ccdbf148df980";
30+ sha256 = "0257p0qd8xx900ngghkjbmjnvn7pjv05g0jm5kkrm4p6alrlhfyl";
31+ };
32+
19 getReflexPlatform = getReflexPlatform' __useLegacyCompilers;
33 getReflexPlatform = getReflexPlatform' __useLegacyCompilers;
20 getReflexPlatform' = __useLegacyCompilers: sys: import ./dep/reflex-platform {
34 getReflexPlatform' = __useLegacyCompilers: sys: import ./dep/reflex-platform {
21 inherit iosSdkVersion __useLegacyCompilers;
35 inherit iosSdkVersion __useLegacyCompilers;
@@ -28,7 +42,7 @@ let
28 })
42 })
29 ];
43 ];
30
44
31− haskellOverlays = [
45+ haskellOverlays = with pkgs.haskell.lib; [
32
46
33 # Fix misc upstream packages
47 # Fix misc upstream packages
34 (self: super: let
48 (self: super: let
@@ -48,6 +62,42 @@ let
48 rev = "6a71119bfa5db2b9990a2491c941469ff8ef5d13";
62 rev = "6a71119bfa5db2b9990a2491c941469ff8ef5d13";
49 sha256 = "0z8smyainnlzcglv3dlx6x1n9j6d2jv48aa8f2421iayfkxg3js5";
63 sha256 = "0z8smyainnlzcglv3dlx6x1n9j6d2jv48aa8f2421iayfkxg3js5";
50 } + /template) {};
64 } + /template) {};
65+
66+ algebraic-graphs = pkgs.haskell.lib.doJailbreak
67+ (self.callCabal2nix "algebraic-graphs" (pkgs.fetchFromGitHub {
68+ owner = "snowleopard";
69+ repo = "alga";
70+ rev = "480a73137e9b38ad3f1bc2c628847953d2fb3e25";
71+ sha256 = "0dpwi5ffs88brl3lz51bwb004c6zm8ds8pkw1vzsg2a6aaiyhlzl";
72+ }) {});
73+
74+ # Need ShowTag, EqTag, and OrdTag instances
75+ dependent-sum-template = self.callCabal2nix "dependent-sum-template" (pkgs.fetchFromGitHub {
76+ owner = "mokus0";
77+ repo = "dependent-sum-template";
78+ rev = "dcb92a98d35d79f712b9a398c8c356acc979ecff";
79+ sha256 = "1v337d810d88jzfriw07pr16d34nibm6q2zkw787lc3sfn07glp5";
80+ }) {};
81+
82+ beam-core = self.callCabal2nix "beam-core" (beam-src + /beam-core) {};
83+ beam-postgres = dontCheck (self.callCabal2nix "beam-postgres" (beam-src + /beam-postgres) {});
84+ beam-migrate = self.callCabal2nix "beam-migrate" (beam-src + /beam-migrate) {};
85+
86+ websockets = self.callCabal2nix "websockets" (pkgs.fetchFromGitHub {
87+ owner = "obsidiansystems";
88+ repo = "websockets";
89+ rev = "b750edf10fc6f532f9fb36588904491af87606c7";
90+ sha256 = "1wd0gkx8a4x6n5cdkra96zvpv8l49i0z90amvrng70ybnzbxpfi5";
91+ }) {};
92+
93+ gargoyle = doJailbreak (self.callCabal2nix "gargoyle" (gargoyle-src + /gargoyle) {});
94+ gargoyle-postgresql-nix = pkgs.haskell.lib.addBuildTool
95+ (self.callCabal2nix "gargoyle-postgresql-nix" (gargoyle-src + /gargoyle-postgresql-nix) {})
96+ pkgs.postgresql; # `staticWhich` requires `psql` on PATH during build time
97+ gargoyle-postgresql = addBuildDepend (doJailbreak (self.callCabal2nix "gargoyle-postgresql" (gargoyle-src + /gargoyle-postgresql) {})) pkgs.postgresql;
98+ postgresql-libpq = enableCabalFlag (overrideCabal super.postgresql-libpq (drv: {
99+ pkgconfigDepends = (drv.pkgconfigDepends or []) ++ [pkgs.postgresql]; #TODO: Obelisk should probably provide this version of postgresql
100+ })) "use-pkg-config";
51 })
101 })
52
102
53 # Add obelisk packages
103 # Add obelisk packages
lib/backend/src/Data/MonoidMap.hsnew file+48
@@ -0,0 +1,48 @@
1+{-# LANGUAGE DeriveFoldable #-}
2+{-# LANGUAGE FlexibleContexts #-}
3+{-# LANGUAGE TypeFamilies #-}
4+module Data.MonoidMap where
5+
6+import Data.Align
7+import Data.Map.Monoidal (MonoidalMap)
8+import Data.Map.Monoidal as Map
9+import Data.Semigroup (Semigroup, (<>))
10+import Reflex (Query, QueryResult, crop, FunctorMaybe (..))
11+
12+newtype MonoidMap k v = MonoidMap { unMonoidMap :: MonoidalMap k v }
13+ deriving (Show, Eq, Ord, Foldable)
14+
15+monoidMap :: (Eq v, Monoid v) => MonoidalMap k v -> MonoidMap k v
16+monoidMap = MonoidMap . Map.filter (/= mempty)
17+
18+instance (Eq (QueryResult q), Ord k, Query q) => Query (MonoidMap k q) where
19+ type QueryResult (MonoidMap k q) = MonoidMap k (QueryResult q)
20+ crop (MonoidMap q) (MonoidMap qr) =
21+ -- This assumes that the query result of a null query should be null
22+ monoidMap $ Map.intersectionWith crop q qr
23+
24+instance (Monoid a, Eq a, Ord k) => Semigroup (MonoidMap k a) where
25+ MonoidMap a <> MonoidMap b =
26+ let combine _ a' b' =
27+ let c = a' `mappend` b'
28+ in if c == mempty
29+ then Nothing
30+ else Just c
31+ in MonoidMap $ Map.mergeWithKey combine id id a b
32+
33+instance (Ord k, Monoid a, Eq a) => Monoid (MonoidMap k a) where
34+ mempty = MonoidMap Map.empty
35+ mappend = (<>)
36+
37+instance (Ord k) => Functor (MonoidMap k) where
38+ fmap f (MonoidMap x) = MonoidMap (fmap f x)
39+
40+instance FunctorMaybe (MonoidMap k) where
41+ fmapMaybe f (MonoidMap m) = MonoidMap $ Map.mapMaybe f m
42+
43+instance (Ord k) => Align (MonoidMap k) where
44+ nil = MonoidMap nil
45+ align (MonoidMap x) (MonoidMap y) = MonoidMap (align x y)
46+ alignWith f (MonoidMap x) (MonoidMap y) = MonoidMap (alignWith f x y)
47+
48+instance (Ord k) => Unalign (MonoidMap k)
lib/backend/src/Obelisk/Api.hsnew file+232
@@ -0,0 +1,232 @@
1+{-# LANGUAGE RankNTypes #-}
2+{-# LANGUAGE ScopedTypeVariables #-}
3+{-# LANGUAGE OverloadedStrings #-}
4+{-# LANGUAGE QuasiQuotes #-}
5+{-# LANGUAGE RecursiveDo #-}
6+{-# LANGUAGE GADTs #-}
7+{-# LANGUAGE PatternSynonyms #-}
8+{-# LANGUAGE LambdaCase #-}
9+module Obelisk.Api where
10+
11+import Prelude hiding ((.))
12+
13+import Obelisk.Postgres.Replication
14+import Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
15+import Obelisk.Db
16+
17+import Control.Category
18+import Control.Concurrent (forkIO, killThread)
19+import Control.Concurrent.Async
20+import Control.Concurrent.STM (TChan, atomically, dupTChan, newBroadcastTChanIO, readTChan, writeTChan, newTChanIO)
21+import Control.Monad (forever, void)
22+import 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)
54+import 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
73+import Data.ByteString (ByteString)
74+import Data.Int
75+import Data.Word
76+import Data.Char
77+import 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
83+
84+{-
85+-- | A way for a pipeline to retrieve data
86+newtype QueryHandler q m = QueryHandler
87+ { runQueryHandler :: q -> m (QueryResult q) }
88+
89+-- | A way to push data into a pipeline
90+newtype Recipient q m = Recipient
91+ { tellRecipient :: QueryResult q -> m () }
92+
93+-- | A way of attaching to (and later detaching from) a pipeline
94+newtype Registrar q = Registrar { unRegistrar :: Recipient q IO -> IO (QueryHandler q IO, IO ()) }
95+
96+-- | A way of connecting a source of data to a consumer of data. The consumer
97+-- can pull data from the datasource and the datasource can push data to the
98+-- consumer.
99+--
100+-- q is the consumer side
101+-- q' is the datasource side
102+newtype Pipeline m q q' = Pipeline { unPipeline :: QueryHandler q' m -> Recipient q m -> IO (QueryHandler q m, Recipient q' m) }
103+
104+instance Category (Pipeline m) where
105+ id = Pipeline $ \qh r -> return (qh, r)
106+ Pipeline f . Pipeline g = Pipeline $ \qh r -> do
107+ rec (qh'', r') <- g qh' r
108+ (qh', r'') <- f qh r'
109+ return (qh'', r'')
110+
111+-- | A key used to track particular consumers
112+newtype ClientKey = ClientKey { unClientKey :: Integer }
113+ deriving (Eq, Ord, Show)
114+
115+servePostgres
116+ :: ( Monoid q'
117+ , Semigroup q'
118+ )
119+ => Pool Postgresql -- ^ The DB to connect to
120+ -> (forall a. request a -> write a)
121+ -> (ByteString -> q' -> read (QueryResult q'))
122+ -> (q' -> read (QueryResult q))
123+ -> Pipeline IO q q'
124+ -> (Snap () -> IO a)
125+ -> IO a
126+servePostgres db handleApi handleNotify handleQuery pipe = do
127+ (getNextNotification, finalizeListener) <- startNotificationListener db
128+ rec (qh, finalizeFeed) <- feedPipeline (handleNotify <$> getNextNotification) handleQuery r
129+ (qh', r) <- unPipeline pipe qh r'
130+ (r', handleListen) <- connectPipelineToWebsockets "" (RequestHandler handleApi) qh'
131+ return (handleListen, finalizeFeed >> finalizeListener)
132+-}
133+
134+newtype ReadDb a = ReadDb { unReadDb :: ReaderT PG.Connection IO a }
135+
136+readTransaction :: PG.Connection -> ReadDb a -> IO a
137+readTransaction conn (ReadDb r) = PG.withTransactionModeRetry m PG.isSerializationError conn $ runReaderT r conn
138+ where m = PG.TransactionMode
139+ { PG.isolationLevel = PG.Serializable
140+ , PG.readWriteMode = PG.ReadOnly
141+ }
142+
143+newtype WriteDb a = WriteDb { unWriteDb :: ReaderT PG.Connection IO a }
144+
145+writeTransaction :: PG.Connection -> WriteDb a -> IO a
146+writeTransaction conn (WriteDb r) = PG.withTransactionModeRetry m PG.isSerializationError conn $ runReaderT r conn
147+ where m = PG.TransactionMode
148+ { PG.isolationLevel = PG.Serializable
149+ , PG.readWriteMode = PG.ReadWrite
150+ }
151+
152+unsafeReadDb :: (PG.Connection -> IO a) -> ReadDb a
153+unsafeReadDb f = ReadDb $ lift . f =<< ask
154+
155+unsafeWriteDb :: (PG.Connection -> IO a) -> WriteDb a
156+unsafeWriteDb f = WriteDb $ lift . f =<< ask
157+
158+withApiHandler
159+ :: forall query queryResult request change a.
160+ ( Monoid (query SelectedCount)
161+ , QueryResult (query (MonoidMap ClientKey SelectedCount)) ~ queryResult (MonoidMap ClientKey SelectedCount)
162+ , QueryResult (query SelectedCount) ~ queryResult SelectedCount
163+ , QueryResult (query ()) ~ queryResult ()
164+ , Foldable queryResult
165+ , Functor query
166+ , Functor queryResult
167+ , FunctorMaybe queryResult
168+ , Monoid (query (MonoidMap ClientKey SelectedCount))
169+ , Monoid (queryResult SelectedCount)
170+ , Eq (queryResult SelectedCount)
171+ , Eq (query (MonoidMap ClientKey SelectedCount))
172+ , Eq (query SelectedCount)
173+ , Group (query SelectedCount)
174+ , ToJSON (queryResult ())
175+ , FromJSON (query ())
176+ , Request request
177+ )
178+ => ByteString -- ^ The URI of the DB to connect to
179+ -> (forall x. request x -> WriteDb x) -- ^ How to handle an API request
180+ -> (forall x. query x -> Transaction -> ReadDb (QueryResult (query x))) -- ^ How to handle a change
181+ -> (forall x. query x -> ReadDb (QueryResult (query x))) -- ^ How to handle a query
182+ -> (Snap () -> IO a)
183+ -> IO a
184+withApiHandler dbUri handleApi handleChange handleQuery go = do
185+ let opts = def
186+ { _logicalDecodingOptions_pluginOptions = [("skip-empty-xacts", Nothing)]
187+ }
188+ withNonEmptyTransactions dbUri $ \transactions -> do
189+ withConnectionPool dbUri $ \db -> do
190+ let onChange :: Transaction -> query x -> IO (QueryResult (query x))
191+ onChange m q = withResource db $ \conn -> readTransaction conn $ handleChange q m
192+ let pipe :: Pipeline IO (MonoidalMap ClientKey (query SelectedCount)) (query (MonoidMap ClientKey SelectedCount))
193+ pipe = queryMorphismPipeline $ transposeMonoidMap . monoidMapQueryMorphism
194+ rec (qh, finalizeFeed) <- feedPipeline (onChange <$> atomically (readTChan transactions)) (QueryHandler $ \q -> withResource db $ \conn -> readTransaction conn $ handleQuery q) r
195+ (qh', r) <- unPipeline pipe qh r'
196+ (r', handleListen) <- connectPipelineToWebsockets "" (\r -> withResource db $ \conn -> writeTransaction conn $ handleApi r) qh'
197+ go handleListen
198+
199+withNonEmptyTransactions :: ByteString -> (TChan Transaction -> IO a) -> IO a
200+withNonEmptyTransactions dbUri go = do
201+ let opts = def
202+ { _logicalDecodingOptions_pluginOptions = [("skip-empty-xacts", Nothing)]
203+ }
204+ withLogicalDecoding dbUri "test_decoding" opts $ \lines -> do
205+ transactions <- newTChanIO
206+ processLine <- linesToTransactions
207+ let processLines = forever $ do
208+ lineRaw <- atomically $ readTChan lines
209+ let Right l = parseOnly (line <* endOfInput) lineRaw
210+ Right mTransaction <- processLine l
211+ forM_ mTransaction $ \transaction ->
212+ atomically $ writeTChan transactions transaction
213+ withAsync processLines $ \a -> link a >> do
214+ go transactions
215+{-
216+
217+standardChangeChannel :: PG.Identifier
218+standardChangeChannel = "obelisk_api_updates"
219+
220+standardChangeSubscriberTable :: PG.Identifier
221+standardChangeSubscriberTable = "obelisk_api_updates_subscriber"
222+
223+standardChangeSpillTable :: PG.Identifier
224+standardChangeSpillTable = "obelisk_api_updates_spill"
225+
226+-- | 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
227+withChangeListener :: FromJSON change => Pool PG.Connection -> (change -> IO ()) -> IO a -> IO a
228+withChangeListener db onChange = withBigNotificationListener db standardChangeChannel standardChangeSpillTable $ \message ->
229+ case Aeson.eitherDecode' $ LBS.fromStrict message of
230+ Left e -> fail $ "withChangeListener: " <> e
231+ Right change -> onChange change
232+-}
lib/backend/src/Obelisk/Api/Pipeline.hsnew file+423
@@ -0,0 +1,423 @@
1+{-# LANGUAGE RecursiveDo #-}
2+{-# LANGUAGE GADTs #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE UndecidableInstances #-}
5+{-# LANGUAGE DeriveGeneric #-}
6+{-# LANGUAGE ScopedTypeVariables #-}
7+{-# LANGUAGE Rank2Types #-}
8+module Obelisk.Api.Pipeline where
9+
10+import Prelude hiding (id, (.))
11+
12+import Control.Concurrent
13+import Control.Monad
14+import Control.Monad.State.Strict
15+import Data.Typeable
16+import GHC.Generics
17+import Control.Lens (imapM_)
18+import Data.Aeson
19+import Data.Aeson.Types (Parser)
20+import Data.Constraint (Dict)
21+import Reflex.Query.Base
22+import Reflex.Query.Class
23+import Reflex.Patch
24+import Reflex.Class
25+import Control.Category
26+import Snap.Core (Snap, MonadSnap)
27+import Data.Text (Text)
28+import Data.Map.Monoidal (MonoidalMap)
29+import qualified Data.AppendMap as Map
30+import Data.MonoidMap (MonoidMap (..), monoidMap)
31+import Data.Foldable
32+import Data.IORef
33+import Debug.Trace
34+import qualified Network.WebSockets as WS
35+import qualified Network.WebSockets.Connection as WS
36+import qualified Network.WebSockets.Stream as WS
37+import Network.WebSockets.Snap
38+import Control.Exception
39+
40+data SomeRequest t where
41+ SomeRequest :: (FromJSON x, ToJSON x) => t x -> SomeRequest t
42+
43+class Request r where
44+ requestToJSON :: r a -> Value
45+ requestParseJSON :: Value -> Parser (SomeRequest r)
46+ requestResponseToJSON :: r a -> Dict (ToJSON a)
47+ requestResponseFromJSON :: r a -> Dict (FromJSON a)
48+
49+instance Request r => FromJSON (SomeRequest r) where
50+ parseJSON = requestParseJSON
51+
52+instance Request r => ToJSON (SomeRequest r) where
53+ toJSON (SomeRequest r) = requestToJSON r
54+
55+-- | A way for a pipeline to retrieve data
56+newtype QueryHandler q m = QueryHandler
57+ { runQueryHandler :: q -> m (QueryResult q) }
58+
59+-- | A way to push data into a pipeline
60+newtype Recipient q m = Recipient
61+ { tellRecipient :: QueryResult q -> m () }
62+
63+-- | A way of attaching to (and later detaching from) a pipeline
64+newtype Registrar q = Registrar { unRegistrar :: Recipient q IO -> IO (QueryHandler q IO, IO ()) }
65+
66+-- | A way of connecting a source of data to a consumer of data. The consumer
67+-- can pull data from the datasource and the datasource can push data to the
68+-- consumer.
69+--
70+-- q is the consumer side
71+-- q' is the datasource side
72+newtype Pipeline m q q' = Pipeline { unPipeline :: QueryHandler q' m -> Recipient q m -> IO (QueryHandler q m, Recipient q' m) }
73+
74+instance Category (Pipeline m) where
75+ id = Pipeline $ \qh r -> return (qh, r)
76+ Pipeline f . Pipeline g = Pipeline $ \qh r -> do
77+ rec (qh'', r') <- g qh' r
78+ (qh', r'') <- f qh r'
79+ return (qh'', r'')
80+
81+-- | A key used to track particular consumers
82+newtype ClientKey = ClientKey { unClientKey :: Integer }
83+ deriving (Eq, Ord, Show)
84+
85+-- | Produce a multi-client 'Recipient' and a single client 'QueryHandler'
86+fanQuery
87+ :: forall k q. (Ord k, Monoid (QueryResult q))
88+ => (k -> IO (Recipient q IO))
89+ -- ^ Look up a recipient
90+ -> QueryHandler (MonoidalMap k q) IO
91+ -- ^ A 'QueryHandler' for multiple clients
92+ -> ( Recipient (MonoidalMap k q) IO
93+ -- Used to notify recipients of new 'QueryResult' data
94+ , k -> QueryHandler q IO
95+ -- Used by 'multiplexQuery' to lookup the 'QueryHandler' for a given client
96+ )
97+fanQuery lookupRecipient qh = (multiRecipient lookupRecipient, fanQueryHandler qh)
98+ where
99+ -- | Given a recipient lookup function, produces a 'Recipient' that can
100+ -- handle distribution of a map of 'QueryResult's to many recipients
101+ multiRecipient lookupR = Recipient $ imapM_ $ \k qr -> do
102+ s <- lookupR k
103+ tellRecipient s qr
104+ -- | Given a multi-client 'QueryHandler', produces a 'QueryHandler' for
105+ -- a single client
106+ fanQueryHandler qh' = \k -> mapQueryHandler (singletonQuery k) qh'
107+
108+-- | Maintains a Map from connected clients to a 'Recipient' that can be used to transmit data to the clients
109+--
110+-- Takes a function that looks up a the 'QueryHandler' for a given client
111+-- Returns: 1. A lookup function into the Map of clients
112+-- 2. A way to register a new client
113+-- a. A 'QueryHandler' for the newly registered client
114+-- b. A removal callback to de-register a particular client
115+multiplexQuery
116+ :: (MonadIO m, Group q)
117+ => (ClientKey -> QueryHandler q m)
118+ -> IO ( ClientKey -> IO (Recipient q m)
119+ , Recipient q m -> IO (QueryHandler q m, m ())
120+ )
121+multiplexQuery lookupQueryHandler = do
122+ clients <- newIORef (ClientKey 0, Map.empty :: MonoidalMap ClientKey (Recipient q m, q))
123+ let
124+ lookupRecipient k = do
125+ (_, cs) <- readIORef clients
126+ case Map.lookup k cs of
127+ Nothing -> do
128+ putStrLn $ mconcat
129+ [ "Rhyolite.Backend.App.multiplexQuery: Failed to find sender for key"
130+ , show k
131+ , " in known keys "
132+ , show $ Map.keys cs
133+ ]
134+ return $ Recipient $ const $ return ()
135+ Just s -> return $ fst s
136+
137+ registerRecipient s = do
138+ cid <- atomicModifyIORef' clients $ \(cid, recipients) ->
139+ ((ClientKey (unClientKey cid + 1), Map.insert cid (s, mempty) recipients), cid)
140+ let
141+ queryHandler = QueryHandler $ \q -> do
142+ liftIO $ atomicModifyIORef' clients $ \(nextCid, recipients) ->
143+ ((nextCid, Map.update (\(r, oldQ) -> Just (r, oldQ <> q)) cid recipients), ())
144+ runQueryHandler (lookupQueryHandler cid) q
145+
146+ unregisterRecipient = do
147+ antiQ <- liftIO $ atomicModifyIORef' clients $ \(nextCid, recipients) ->
148+ case Map.updateLookupWithKey (\_ _ -> Nothing) cid recipients of
149+ (Nothing, _) -> trace
150+ ("Rhyolite.Backend.App.multiplexQuery: Tried to unregister a client key that is not registered " <> show cid)
151+ ((nextCid, recipients), mempty)
152+ (Just (_, removedQuery), newRecipients) -> ((nextCid, newRecipients), negateG removedQuery)
153+
154+ -- TODO: Should we have a way of ensuring that this doesn't actually cause a query to be run?
155+ -- It shouldn't cause the query to be run again but it depends on if the callee will notice
156+ -- that the new query is strictly smaller than the old one.
157+ runQueryHandler (lookupQueryHandler cid) antiQ
158+ return ()
159+
160+ return (queryHandler, unregisterRecipient)
161+
162+ return (lookupRecipient, registerRecipient)
163+
164+-- | Represents a WebSocket message from one of two channels: ViewSelector declarations or API requests
165+data WebSocketRequest q r
166+ = WebSocketRequest_ViewSelector q
167+ | WebSocketRequest_Api (TaggedRequest r)
168+ deriving (Typeable, Generic)
169+
170+instance (Request r, FromJSON q) => FromJSON (WebSocketRequest q r)
171+instance (Request r, ToJSON q) => ToJSON (WebSocketRequest q r)
172+
173+-- | Represents a WebSocket response from one of three channels: incoming 'View's, API responses, or version info
174+data WebSocketResponse q
175+ = WebSocketResponse_View (QueryResult q)
176+ | WebSocketResponse_Api TaggedResponse
177+ | WebSocketResponse_Version Text
178+ deriving (Typeable, Generic)
179+
180+instance FromJSON (QueryResult q) => FromJSON (WebSocketResponse q)
181+instance ToJSON (QueryResult q) => ToJSON (WebSocketResponse q)
182+
183+-- | A request tagged with an identifier
184+data TaggedRequest r = TaggedRequest Value (SomeRequest r)
185+ deriving (Typeable, Generic)
186+
187+instance Request r => FromJSON (TaggedRequest r)
188+instance Request r => ToJSON (TaggedRequest r)
189+
190+-- | A response tagged with an identifier matching the one in the 'TaggedRequest'. The identifier is the first argument.
191+data TaggedResponse = TaggedResponse Value Value
192+ deriving (Typeable, Generic)
193+
194+instance FromJSON TaggedResponse
195+instance ToJSON TaggedResponse
196+
197+-- | Handles a websocket connection
198+handleWebsocket
199+ :: forall q qr r.
200+ ( Eq (q SelectedCount)
201+ , Eq (qr SelectedCount)
202+ , QueryResult (q SelectedCount) ~ qr SelectedCount
203+ , QueryResult (q ()) ~ qr ()
204+ , ToJSON (qr ())
205+ , Functor qr
206+ , Group (q SelectedCount)
207+ , Monoid (qr SelectedCount)
208+ , FromJSON (q ())
209+ , Request r
210+ , Functor q
211+ )
212+ => Text -- ^ Version
213+ -> (forall a. r a -> IO a)
214+ -> Registrar (q SelectedCount)
215+ -> Snap ()
216+handleWebsocket v rh register = withWebsocketsConnection $ \conn -> do
217+ let sender = Recipient $ sendEncodedDataMessage conn . (\a -> WebSocketResponse_View (void a) :: WebSocketResponse (q ()))
218+ sendEncodedDataMessage conn (WebSocketResponse_Version v :: WebSocketResponse (q ()))
219+ bracket (unRegistrar register sender) snd $ \(vsHandler, _) -> flip evalStateT mempty $ forever $ do
220+ (wsr :: WebSocketRequest (q ()) r) <- liftIO $ getDataMessage conn
221+ case wsr of
222+ WebSocketRequest_Api (TaggedRequest reqId (SomeRequest req)) -> lift $ do
223+ a <- rh req
224+ sendEncodedDataMessage conn
225+ (WebSocketResponse_Api $ TaggedResponse reqId (toJSON a) :: WebSocketResponse (q ()))
226+ WebSocketRequest_ViewSelector new -> do
227+ old <- get
228+ let new' = SelectedCount 1 <$ new
229+ vsDiff = new' ~~ old
230+ when (vsDiff /= mempty) $ do
231+ qr <- lift $ runQueryHandler vsHandler vsDiff
232+ put new'
233+ when (qr /= mempty) $ lift $
234+ sendEncodedDataMessage conn (WebSocketResponse_View (void qr) :: WebSocketResponse (q ()))
235+
236+-------------------------------------------------------------------------------
237+
238+-- | Connect a datasource (e.g., a database) to a pipeline
239+--
240+-- Data taken from 'getNextNotification' is pushed into the pipeline and
241+-- when the pipeline pulls data, it is retrieved using 'qh'
242+feedPipeline
243+ :: Monoid q
244+ => IO (q -> IO (QueryResult q))
245+ -- ^ Get the next notification to be sent to the pipeline. If no notification
246+ -- is available, this should block until one is available
247+ -> QueryHandler q IO
248+ -- ^ Retrieve data when requested by pipeline
249+ -> Recipient q IO
250+ -- ^ A way to push data into the pipeline
251+ -> IO (QueryHandler q IO, IO ())
252+ -- ^ A way for the pipeline to request data
253+feedPipeline getNextNotification qh r = do
254+ currentQuery <- newIORef mempty
255+ let qhSaveQuery = QueryHandler $ \q -> do
256+ atomicModifyIORef' currentQuery $ \old -> (q <> old, ())
257+ runQueryHandler qh q
258+ tid <- forkIO . forever $ do
259+ nm <- getNextNotification
260+ q <- readIORef currentQuery
261+ qr <- nm q
262+ tellRecipient r qr
263+ return (qhSaveQuery, killThread tid)
264+
265+-- | Connects the pipeline to websockets consumers
266+connectPipelineToWebsockets
267+ :: ( Eq (q SelectedCount)
268+ , QueryResult (q SelectedCount) ~ qr SelectedCount
269+ , QueryResult (q ()) ~ qr ()
270+ , Group (q SelectedCount)
271+ , Monoid (qr SelectedCount)
272+ , Eq (qr SelectedCount)
273+ , ToJSON (qr ())
274+ , FromJSON (q ())
275+ , Request r
276+ , Functor qr
277+ , Functor q
278+ )
279+ => Text
280+ -> (forall a. r a -> IO a)
281+ -- ^ API handler
282+ -> QueryHandler (MonoidalMap ClientKey (q SelectedCount)) IO
283+ -- ^ A way to retrieve more data for each consumer
284+ -> IO (Recipient (MonoidalMap ClientKey (q SelectedCount)) IO, Snap ())
285+ -- ^ A way to send data to many consumers and a handler for websockets connections
286+connectPipelineToWebsockets ver rh qh = do
287+ (allRecipients, registerRecipient) <- connectPipelineToWebsockets' qh
288+ return (allRecipients, handleWebsocket ver rh registerRecipient)
289+
290+-- | Like 'connectPipelineToWebsockets' but returns a Registrar that can
291+-- be used to construct a handler for a particular client
292+connectPipelineToWebsockets'
293+ :: (Monoid (QueryResult q), Group q)
294+ => QueryHandler (MonoidalMap ClientKey q) IO
295+ -> IO (Recipient (MonoidalMap ClientKey q) IO, Registrar q)
296+ -- ^ A way to send data to many consumers, and a way to register new consumers
297+connectPipelineToWebsockets' qh = do
298+ rec (lookupRecipient, registerRecipient) <- multiplexQuery clientQueryHandler
299+ let (allRecipients, clientQueryHandler) = fanQuery lookupRecipient qh
300+ return (allRecipients, Registrar registerRecipient)
301+
302+-- | Extends a 'Registrar' with a 'Pipeline'
303+extendRegistrar :: Pipeline IO q q' -> Registrar q' -> Registrar q
304+extendRegistrar (Pipeline p) (Registrar r) = Registrar $ \recipient -> do
305+ rec (qh', close) <- r recipient'
306+ (qh, recipient') <- p qh' recipient
307+ return (qh, close)
308+
309+-------------------------------------------------------------------------------
310+
311+singletonQuery
312+ :: ( Monoid (QueryResult q)
313+ , Ord k
314+ )
315+ => k
316+ -> QueryMorphism q (MonoidalMap k q)
317+singletonQuery k = QueryMorphism { _queryMorphism_mapQuery = Map.singleton k
318+ , _queryMorphism_mapQueryResult = Map.findWithDefault mempty k
319+ }
320+
321+monoidMapQueryMorphism :: (Eq q, Monoid q) => QueryMorphism (MonoidalMap k q) (MonoidMap k q)
322+monoidMapQueryMorphism = QueryMorphism
323+ { _queryMorphism_mapQuery = monoidMap
324+ , _queryMorphism_mapQueryResult = unMonoidMap
325+ }
326+
327+transposeMonoidMap
328+ :: forall a a' k q qr.
329+ ( Ord k
330+ , Eq a
331+ , Monoid a
332+ , Monoid a'
333+ , Eq (q (MonoidMap k a))
334+ , Eq (QueryResult (q a))
335+ , Foldable qr
336+ , Functor q
337+ , Functor qr
338+ , FunctorMaybe qr
339+ , Monoid (q (MonoidMap k a))
340+ , Monoid (QueryResult (q a))
341+ , QueryResult (q (MonoidMap k a)) ~ qr (MonoidMap k a')
342+ , QueryResult (q a) ~ qr a'
343+ )
344+ => QueryMorphism (MonoidMap k (q a)) (q (MonoidMap k a))
345+transposeMonoidMap = QueryMorphism
346+ { _queryMorphism_mapQuery = aggregateQueries
347+ , _queryMorphism_mapQueryResult = distributeResults
348+ }
349+ where
350+ aggregateQueries :: MonoidMap k (q a) -> q (MonoidMap k a)
351+ aggregateQueries = fold . monoidMap . Map.mapWithKey (\k q -> fmap (monoidMap . Map.singleton k) q) . unMonoidMap
352+ distributeResults :: qr (MonoidMap k a') -> MonoidMap k (qr a')
353+ distributeResults v = monoidMap $ Map.mapWithKey (\k _ -> fmapMaybe (Map.lookup k . unMonoidMap) v) $ fold $ fmap unMonoidMap v
354+
355+mapQueryHandlerAndRecipient
356+ :: Functor f
357+ => QueryMorphism q q'
358+ -> QueryHandler q' f
359+ -> Recipient q f
360+ -> (QueryHandler q f, Recipient q' f)
361+mapQueryHandlerAndRecipient qm qh s = (mapQueryHandler qm qh, mapRecipient qm s)
362+
363+mapQueryHandler :: Functor f => QueryMorphism q q' -> QueryHandler q' f -> QueryHandler q f
364+mapQueryHandler qm qh = QueryHandler $ \qs -> mapQueryResult qm <$> runQueryHandler qh (mapQuery qm qs)
365+
366+mapRecipient :: QueryMorphism q q' -> Recipient q m -> Recipient q' m
367+mapRecipient qm s = Recipient $ \qr -> tellRecipient s $ mapQueryResult qm qr
368+
369+fmapQueryMorphism
370+ :: ( Functor f, QueryResult (f q) ~ f (QueryResult q)
371+ , QueryResult (f q') ~ f (QueryResult q') )
372+ => QueryMorphism q q'
373+ -> QueryMorphism (f q) (f q')
374+fmapQueryMorphism qm = QueryMorphism
375+ { _queryMorphism_mapQuery = fmap $ _queryMorphism_mapQuery qm
376+ , _queryMorphism_mapQueryResult = fmap $ _queryMorphism_mapQueryResult qm
377+ }
378+
379+queryMorphismPipeline :: Functor m => QueryMorphism q q' -> Pipeline m q q'
380+queryMorphismPipeline qm = Pipeline $ \qh r -> pure $ mapQueryHandlerAndRecipient qm qh r
381+
382+-- | Accepts a websockets connection and runs the supplied action with it
383+withWebsocketsConnection :: MonadSnap m => (WS.Connection -> IO ()) -> m ()
384+withWebsocketsConnection f = runWebSocketsSnap $ \pc -> do
385+ conn <- WS.acceptRequest pc
386+ handleSomeException $ handleConnectionException pc $ f conn
387+ where
388+ handleSomeException = handle $ \(SomeException e) -> putStrLn $ "withWebsocketsConnection: " <> displayException e
389+ handleConnectionException pc = handle $ \e -> case e of
390+ WS.ConnectionClosed -> return ()
391+ WS.CloseRequest _ _ -> print e >> WS.close (WS.pendingStream pc) >> throwIO e
392+ _ -> do putStr $ "withWebsocketsConnection: Exception: " <> displayException e
393+ throwIO e
394+
395+-- | Attempts to json decode a websockets data message
396+decodeWebsocketsDataMessage :: FromJSON a => WS.DataMessage -> Either String a
397+decodeWebsocketsDataMessage dm = eitherDecode' $ case dm of
398+ WS.Text r' _ -> r'
399+ WS.Binary r' -> r'
400+
401+-- | Parse and process a single websocket data message
402+getDataMessage
403+ :: FromJSON a
404+ => WS.Connection
405+ -> IO a
406+getDataMessage conn = do
407+ dm <- WS.receiveDataMessage conn
408+ case decodeWebsocketsDataMessage dm of
409+ Left err -> liftIO $ throwIO $ AssertionFailed $ mconcat
410+ [ "getDataMessage: error: "
411+ , err
412+ , "; received: "
413+ , show dm
414+ ]
415+ Right a -> return a
416+
417+-- | Send a json encoded data message over the websocket connection
418+sendEncodedDataMessage
419+ :: ToJSON a
420+ => WS.Connection
421+ -> a
422+ -> IO ()
423+sendEncodedDataMessage conn = WS.sendDataMessage conn . (\x -> WS.Text x Nothing) . encode
lib/backend/src/Obelisk/Db.hsnew file+50
@@ -0,0 +1,50 @@
1+module Obelisk.Db (withDb, withDbUri, withConnectionPool) where
2+
3+import Control.Exception (bracket)
4+import Data.ByteString (ByteString)
5+import qualified Data.ByteString.Char8 as BS
6+import Data.Pool (Pool, createPool, destroyAllResources)
7+import Database.PostgreSQL.Simple as PG
8+import Gargoyle
9+import Gargoyle.PostgreSQL.Nix (postgresNix)
10+import System.Directory (doesFileExist)
11+import System.FilePath
12+
13+-- | Connects to the database using information at the given location
14+withDb
15+ :: FilePath
16+ -- ^ The path of either a 'Gargoyle'-wrapped Postgres database or a file
17+ -- containing the URL of a Postgres database. If the path does not exist, it
18+ -- will be created with 'Gargoyle'.
19+ -> (Pool PG.Connection -> IO a)
20+ -- ^ An IO action to run while the database is open; the database will remain
21+ -- open until this action exits.
22+ -> IO a
23+withDb dbPath a = withDbUri dbPath $ \dbUri -> withConnectionPool dbUri a
24+
25+withDbUri :: FilePath -> (ByteString -> IO a) -> IO a
26+withDbUri dbPath a = do
27+ dbExists <- doesFileExist dbPath
28+ if dbExists
29+ -- use the file contents as the uri for an existing server
30+ then do
31+ dbUri <- BS.readFile dbPath
32+ a dbUri
33+ -- otherwise assume its a folder for a local database
34+ else do
35+ defaultConfig <- postgresNix
36+ let myConfig = defaultConfig
37+ { _gargoyle_init = \workDir -> do
38+ _gargoyle_init defaultConfig workDir
39+ appendFile (workDir </> "postgresql.conf") "\n# Added by Obelisk.Db\nwal_level = logical\n"
40+ }
41+ withGargoyle myConfig dbPath a
42+
43+openDbUri :: ByteString -> IO (Pool PG.Connection)
44+openDbUri dbUri = do
45+ let openPostgresql = connectPostgreSQL dbUri
46+ closePostgresql p = close p
47+ createPool openPostgresql closePostgresql 1 5 20
48+
49+withConnectionPool :: ByteString -> (Pool PG.Connection -> IO a) -> IO a
50+withConnectionPool dbUri = bracket (openDbUri dbUri) destroyAllResources
lib/backend/src/Obelisk/Postgres/LogicalDecoding/Plugins/TestDecoding.hsnew file+218
@@ -0,0 +1,218 @@
1+{-# LANGUAGE OverloadedStrings #-}
2+{-# LANGUAGE LambdaCase #-}
3+module Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
4+ ( line
5+ , Transaction
6+ , LineError (..)
7+ , linesToTransactions
8+ ) where
9+
10+import Prelude hiding (takeWhile)
11+
12+import Control.Applicative
13+import Control.Monad
14+import Data.Attoparsec.ByteString (word8, notWord8)
15+import Data.Attoparsec.ByteString.Char8
16+import Data.ByteString (ByteString)
17+import qualified Data.ByteString as BS
18+import Data.Char (ord)
19+import qualified Data.Char as Char
20+import Data.Word
21+import Data.HashMap.Strict (HashMap)
22+import qualified Data.HashMap.Strict as HashMap
23+import Data.Sequence (Seq, (|>))
24+import Data.Text (Text)
25+import Data.Text.Encoding
26+import Database.PostgreSQL.Simple.Types
27+import qualified Data.Map as Map
28+import Data.Map (Map)
29+import Data.IORef
30+import Data.Maybe
31+
32+--------------------------------------------------------------------------------
33+-- Potentially upstreamable
34+--------------------------------------------------------------------------------
35+
36+stringWithSingleQuote :: Parser ByteString
37+stringWithSingleQuote = stringWithQuote $ fromIntegral $ ord '\''
38+
39+stringWithDoubleQuote :: Parser ByteString
40+stringWithDoubleQuote = stringWithQuote $ fromIntegral $ ord '"'
41+
42+stringWithQuote :: Word8 -> Parser ByteString
43+stringWithQuote q = do
44+ _ <- word8 q
45+ s <- many $ notWord8 q <|> doubledWord8 q
46+ _ <- word8 q
47+ pure $ BS.pack s
48+
49+doubledWord8 :: Word8 -> Parser Word8
50+doubledWord8 c = do
51+ _ <- word8 c
52+ _ <- word8 c
53+ pure c
54+
55+identifierLike :: (Char -> Bool) -> (Char -> Bool) -> Parser Text
56+identifierLike validFirstChar validOtherChar = fmap decodeUtf8 stringWithDoubleQuote <|> do
57+ -- Avoid plucking the first character off, so that we don't have to cons it back on (which would reallocate the ByteString)
58+ guard . validFirstChar =<< peekChar'
59+ decodeUtf8 <$> takeWhile1 validOtherChar
60+
61+identifier :: Parser Identifier
62+identifier = Identifier <$> identifierLike identifierFirstChar identifierOtherChar
63+
64+identifierFirstChar :: Char -> Bool
65+identifierFirstChar c = Char.isLetter c || c == '_'
66+
67+identifierOtherChar :: Char -> Bool
68+identifierOtherChar c = identifierFirstChar c || isDigit c || c == '$'
69+
70+qualifiedIdentifier :: Parser QualifiedIdentifier
71+qualifiedIdentifier = do
72+ namespace <- identifier
73+ _ <- char '.'
74+ name <- identifier
75+ pure $ QualifiedIdentifier (Just $ fromIdentifier namespace) $ fromIdentifier name
76+
77+newtype TypeName = TypeName { unTypeName :: Text } deriving (Show, Read, Eq, Ord)
78+
79+typeName :: Parser TypeName
80+typeName = do
81+ let otherChar c = identifierOtherChar c || c `elem` (" ()," :: String)
82+ TypeName <$> identifierLike identifierFirstChar otherChar
83+
84+newtype Xid = Xid { unXid :: Word64 } deriving (Show, Read, Eq, Ord)
85+
86+xid :: Parser Xid
87+xid = Xid <$> decimal
88+
89+--------------------------------------------------------------------------------
90+
91+rowEntry :: Parser a -> Parser (Identifier, (TypeName, a))
92+rowEntry item = do
93+ colName <- identifier
94+ _ <- char '['
95+ colType <- typeName
96+ _ <- string "]:"
97+ v <- item
98+ pure (colName, (colType, v))
99+
100+rowLike :: Parser a -> Parser (HashMap Identifier (TypeName, a))
101+rowLike item = {- (mempty <$ string "(no-tuple-data)") <|> -} do
102+ fmap HashMap.fromList $ flip sepBy1 (char ' ') $ rowEntry item
103+
104+type Row = HashMap Identifier (TypeName, Maybe Literal)
105+
106+data Literal
107+ = Literal_UnchangedToastDatum
108+ | Literal_Present ByteString
109+ deriving (Show, Read, Eq, Ord)
110+
111+nullableLiteral :: Parser (Maybe Literal)
112+nullableLiteral = Nothing <$ string "null" <|> Just <$> literal
113+
114+literal :: Parser Literal
115+literal = unchangedToastDatum <|> presentDatum
116+ where
117+ unchangedToastDatum = Literal_UnchangedToastDatum <$ string "unchanged-toast-datum"
118+ -- | We convert all present datums to their normal Postgres ByteString
119+ -- encoding, so that we don't have to preserve special cases throughout our
120+ -- code
121+ presentDatum :: Parser Literal
122+ presentDatum = Literal_Present <$> foldr1 (<|>)
123+ [ stringWithSingleQuote
124+ , char 'B' *> stringWithSingleQuote
125+ , "t" <$ string "true"
126+ , "f" <$ string "false"
127+ -- We assume anything else must be numeric
128+ , takeWhile1 (/= ' ')
129+ ]
130+
131+row :: Parser Row
132+row = rowLike nullableLiteral
133+
134+type Key = HashMap Identifier (TypeName, Literal) -- Keys can't contain nulls
135+
136+key :: Parser Key
137+key = rowLike literal
138+
139+data Change
140+ = Change_Insert Row
141+ | Change_Update Key Row
142+ | Change_Delete Key
143+ deriving (Show, Read, Eq, Ord)
144+
145+change :: Parser Change
146+change = do
147+ t <- takeWhile (/= ':')
148+ _ <- string ": "
149+ case t of
150+ "INSERT" -> Change_Insert <$> row
151+ "UPDATE" -> foldr1 (<|>)
152+ [ do _ <- string "old-key: "
153+ old <- key
154+ _ <- string " new-tuple: "
155+ new <- row
156+ pure $ Change_Update old new
157+ , Change_Update mempty <$> row
158+ ]
159+ "DELETE" -> Change_Delete <$> key
160+ _ -> fail $ "Unrecognized change type: " <> show t
161+
162+data Line
163+ = Line_Begin Xid
164+ | Line_Change QualifiedIdentifier Change
165+ | Line_Commit Xid
166+ deriving (Show, Read, Eq, Ord)
167+
168+line :: Parser Line
169+line = do
170+ leader <- takeWhile (/= ' ')
171+ _ <- char ' '
172+ case leader of
173+ "BEGIN" -> Line_Begin <$> xid
174+ "table" -> do
175+ table <- qualifiedIdentifier
176+ _ <- string ": "
177+ Line_Change table <$> change
178+ "COMMIT" -> Line_Commit <$> xid
179+ _ -> fail $ "Unrecognized line leader: " <> show leader
180+
181+type Transaction = (Xid, Map QualifiedIdentifier (Seq Change))
182+
183+data LineError
184+ = LineError_OtherTransactionAlreadyInProgress
185+ Xid -- ^ Existing transaction ID
186+ Xid -- ^ New transaction ID that tried to start
187+ | LineError_CommittingWithoutTransaction
188+ Xid -- ^ Transaction ID that tried to commit
189+ | LineError_CommittingWrongTransaction
190+ Xid -- ^ Existing transaction ID
191+ Xid -- ^ New transaction ID that tried to start
192+ | LineError_ChangeOutsideTransaction
193+
194+-- | Create a function that will statefully accumulate lines until a Transaction
195+-- is produced. If the given line is not valid in the current state, an error
196+-- will be produced and the line will be ignored. This should never happen for
197+-- a valid LogicalDecoding session, so all errors should be treated as serious.
198+linesToTransactions :: IO (Line -> IO (Either LineError (Maybe Transaction)))
199+linesToTransactions = do
200+ pending <- newIORef Nothing
201+ return $ \l -> do
202+ let -- Update the pending transaction and possibly emit a message
203+ update :: Maybe Transaction -> Either LineError (Maybe Transaction, Maybe Transaction)
204+ update = case l of
205+ Line_Begin newXid -> \case
206+ Nothing -> Right (Just (newXid, mempty), Nothing)
207+ Just (existingXid, _) -> Left $ LineError_OtherTransactionAlreadyInProgress existingXid newXid
208+ Line_Commit newXid -> \case
209+ Just txn@(existingXid, _)
210+ | existingXid == newXid -> Right (Nothing, Just txn)
211+ | otherwise -> Left $ LineError_CommittingWrongTransaction existingXid newXid
212+ Nothing -> Left $ LineError_CommittingWithoutTransaction newXid
213+ Line_Change qid c -> \old -> case old of
214+ Just (existingXid, changes) -> Right (Just (existingXid, Map.alter (Just . (|> c) . fromMaybe mempty) qid changes), Nothing)
215+ Nothing -> Left LineError_ChangeOutsideTransaction
216+ atomicModifyIORef pending $ \old -> case update old of
217+ Left e -> (old, Left e)
218+ Right (new, a) -> (new, Right a)
lib/backend/src/Obelisk/Postgres/Replication.hsnew file+178
@@ -0,0 +1,178 @@
1+{-# LANGUAGE FlexibleContexts #-}
2+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
3+{-# LANGUAGE LambdaCase #-}
4+{-# LANGUAGE OverloadedStrings #-}
5+{-# LANGUAGE QuasiQuotes #-}
6+{-# LANGUAGE ScopedTypeVariables #-}
7+{-# LANGUAGE TypeApplications #-}
8+
9+-- | Listen to database changes as a replication client
10+--
11+-- In order for this to work you need the following settings in postgresql.conf:
12+--
13+-- @
14+-- wal_level = logical
15+-- max_replication_slots = n (where n > 0)
16+-- @
17+module Obelisk.Postgres.Replication
18+ ( withLogicalDecoding
19+ , LogicalDecodingOptions (..)
20+ ) where
21+
22+import Data.Time
23+import Data.Default
24+import Control.Concurrent.Async
25+import Control.Monad.Except
26+import Data.Word
27+import Control.Exception
28+import Control.Concurrent
29+import Control.Concurrent.STM
30+import Data.ByteString (ByteString)
31+import qualified Data.ByteString as BS
32+import qualified Data.ByteString.Lazy as LBS
33+import Data.IORef
34+import Data.Text (Text)
35+import qualified Data.Text as T
36+import Data.Text.Encoding
37+import qualified Database.PostgreSQL.Simple as PG
38+import qualified Database.PostgreSQL.Simple.Copy as PG
39+import qualified Database.PostgreSQL.LibPQ as PQ
40+import qualified Database.PostgreSQL.Simple.Types as PG
41+import qualified Database.PostgreSQL.Simple.Internal as PG
42+import Data.Int
43+import Data.Binary (Binary)
44+import qualified Data.Binary as Binary
45+import qualified Data.Binary.Get as Binary
46+import qualified Data.Binary.Put as Binary
47+import Data.Char
48+
49+import Rhyolite.Backend.DB.PsqlSimple hiding (Binary)
50+
51+-- | Start a replication session using logical decoding
52+withLogicalDecoding
53+ :: ByteString -- ^ Database URI
54+ -> PG.Identifier -- ^ Logical decoding plugin
55+ -> LogicalDecodingOptions -- ^ Options; use `def` for default
56+ -> (TChan ByteString -> IO a) -- ^ Action to run while replication session is active
57+ -> IO a
58+withLogicalDecoding dbUri pluginName opts go = bracket (PQ.connectdb $ dbUri <> "?connect_timeout=0&replication=database") PQ.finish $ \unwrappedConn -> do
59+ wrappedConn <- newMVar unwrappedConn
60+ let withConn :: forall a. (PQ.Connection -> IO a) -> IO a
61+ withConn = withMVar wrappedConn
62+ pgConn <- PG.Connection
63+ <$> pure wrappedConn
64+ <*> newMVar mempty
65+ <*> newIORef 0
66+ backendPid <- PQ.backendPID unwrappedConn
67+ let slotName = PG.Identifier $ slotNamePrefix <> T.pack (show backendPid)
68+ [(_, _, _, _) :: (Text, Text, Maybe Text, Text)] <- uncurry (PG.query pgConn) [sqlQ| CREATE_REPLICATION_SLOT ?slotName TEMPORARY LOGICAL ?pluginName NOEXPORT_SNAPSHOT |]
69+ withConn $ \conn -> do
70+ Just escapedSlotName <- PQ.escapeIdentifier conn $ encodeUtf8 $ PG.fromIdentifier slotName
71+ optionsString <- case _logicalDecodingOptions_pluginOptions opts of
72+ [] -> pure ""
73+ pluginOptions -> fmap ((" (" <>) . (<> ")") . BS.intercalate ", ") $ forM pluginOptions $ \(name, mValue) -> do
74+ Just nameString <- PQ.escapeIdentifier conn $ encodeUtf8 $ PG.fromIdentifier name
75+ valueString <- case mValue of
76+ Just value -> do
77+ Just encodedValue <- PQ.escapeStringConn conn value
78+ pure $ " " <> encodedValue
79+ Nothing -> pure ""
80+ pure $ nameString <> valueString
81+ Just _ <- PQ.exec conn $ "START_REPLICATION SLOT " <> escapedSlotName <> " LOGICAL 0/0" <> optionsString
82+ pure ()
83+ writtenLsn <- newIORef 0
84+ chan <- newTChanIO
85+ let get = PG.getCopyData pgConn >>= \case
86+ PG.CopyOutDone _ -> pure () --TODO: Can this ever actually happen? Maybe on database shutdown
87+ PG.CopyOutRow bs -> case decodeReplicationDownstreamMessage bs of
88+ Left e -> print e
89+ Right msg -> case msg of
90+ ReplicationDownstreamMessage_Data lsn _ _ payload -> do
91+ atomicModifyIORef writtenLsn $ \old -> (max old lsn, ())
92+ atomically $ writeTChan chan payload
93+ ReplicationDownstreamMessage_Keepalive lsn _ shouldReply -> do
94+ atomicModifyIORef writtenLsn $ \old -> (max old lsn, ())
95+ when shouldReply sendStatus
96+ sendStatus = do
97+ lsn <- readIORef writtenLsn
98+ now <- toPostgresTime <$> getCurrentTime
99+ let msg = ReplicationUpstreamMessage_StandbyStatus lsn lsn lsn now True
100+ let encodedMessage = encodeReplicationUpstreamMessage msg
101+ PQ.CopyInOk <- PQ.putCopyData unwrappedConn encodedMessage
102+ PQ.FlushOk <- PQ.flush unwrappedConn
103+ pure ()
104+ withAsync (forever $ sendStatus >> threadDelay (10*1000*1000)) $ \_ ->
105+ withAsync (forever get) $ \_ -> do
106+ go chan
107+
108+data LogicalDecodingOptions = LogicalDecodingOptions
109+ { _logicalDecodingOptions_pluginOptions :: [(PG.Identifier, Maybe ByteString)]
110+ -- ^ Options to pass to the logical decoding plugin
111+ }
112+
113+instance Default LogicalDecodingOptions where
114+ def = LogicalDecodingOptions
115+ { _logicalDecodingOptions_pluginOptions = []
116+ }
117+
118+slotNamePrefix :: Text
119+slotNamePrefix = "obelisk_"
120+
121+-- | See https://www.postgresql.org/docs/current/static/protocol-replication.html
122+data ReplicationDownstreamMessage
123+ = ReplicationDownstreamMessage_Data Int64 Int64 PostgresTime ByteString
124+ | ReplicationDownstreamMessage_Keepalive Int64 PostgresTime Bool
125+ deriving (Show, Read, Eq, Ord)
126+
127+-- | See https://www.postgresql.org/docs/current/static/protocol-replication.html
128+data ReplicationUpstreamMessage
129+ = ReplicationUpstreamMessage_StandbyStatus Int64 Int64 Int64 PostgresTime Bool
130+ | ReplicationUpstreamMessage_HotStandbyStatus PostgresTime Int32 Int32 Int32 Int32
131+ deriving (Show, Read, Eq, Ord)
132+
133+--NOTE: we assume we are talking to a little-endian Postgres instance
134+decodeReplicationDownstreamMessage :: MonadError Text m => ByteString -> m ReplicationDownstreamMessage
135+decodeReplicationDownstreamMessage bs = do
136+ let decodeHeader = do
137+ Binary.get @Word8 >>= \case
138+ -- 'w'
139+ 119 -> do
140+ header :: ByteString -> ReplicationDownstreamMessage
141+ <- ReplicationDownstreamMessage_Data <$> Binary.get <*> Binary.get <*> Binary.get
142+ pure $ pure . header
143+ -- 'k'
144+ 107 -> do
145+ result <- ReplicationDownstreamMessage_Keepalive <$> Binary.get <*> Binary.get <*> Binary.get
146+ pure $ \remaining -> if BS.null remaining
147+ then pure result
148+ else throwError $ "decodeReplicationDownstreamMessage: keepalive message had extra data"
149+ leadByte -> fail $ "decodeReplicationDownstreamMessage: unexpected leading byte: " <> show (chr $ fromIntegral leadByte)
150+ case Binary.runGetOrFail decodeHeader $ LBS.fromStrict bs of
151+ Left (_, offset, e) -> throwError $ T.pack $ "decodeReplicationDownstreamMessage: at " <> show offset <> ": " <> e
152+ Right (remaining, _, finish) -> finish $ LBS.toStrict remaining
153+
154+encodeReplicationUpstreamMessage :: ReplicationUpstreamMessage -> ByteString
155+encodeReplicationUpstreamMessage = LBS.toStrict . Binary.runPut . \case
156+ ReplicationUpstreamMessage_StandbyStatus written flushed applied time pleaseReply -> do
157+ Binary.put @Word8 114 -- 'r'
158+ Binary.put written
159+ Binary.put flushed
160+ Binary.put applied
161+ Binary.put time
162+ Binary.put pleaseReply
163+ ReplicationUpstreamMessage_HotStandbyStatus time xmin xminEpoch lowestCatalogXmin lowestCatalogXminEopch -> do
164+ Binary.put @Word8 104 -- 'h'
165+ Binary.put time
166+ Binary.put xmin
167+ Binary.put xminEpoch
168+ Binary.put lowestCatalogXmin
169+ Binary.put lowestCatalogXminEopch
170+
171+newtype PostgresTime = PostgresTime { unPostgresTime :: Int64 } deriving (Show, Read, Eq, Ord, Binary)
172+
173+toPostgresTime :: UTCTime -> PostgresTime
174+toPostgresTime t = PostgresTime $ fromIntegral $
175+ (toModifiedJulianDay (utctDay t) - toModifiedJulianDay postgresEpoch) * (24*60*60*1000*1000)
176+ + round (utctDayTime t * (1000*1000))
177+ where
178+ postgresEpoch = fromGregorian 2000 1 1
skeleton/backend/src/Backend.hs+2
@@ -7,6 +7,8 @@ module Backend where
7
7
8import Common.Route
8import Common.Route
9import Obelisk.Backend
9import Obelisk.Backend
10+import Obelisk.Route
11+import Obelisk.Api.Pipeline
10
12
11backend :: Backend BackendRoute FrontendRoute
13backend :: Backend BackendRoute FrontendRoute
12backend = Backend
14backend = Backend