git.iko.soy/obelisk
on 31a32dc help sign in
7ad33cb
swap
6 ahead, 1276 behind
default.nix+42−1
since splitsnapshots
unifiedsplit
@@ -16,6 +16,10 @@ let
16 openssh
16 openssh
17 ];
17 ];
18
18
19+ beam-src = hackGet ./dep/beam;
20+
21+ gargoyle-src = hackGet ./dep/gargoyle;
22+
19 getReflexPlatform = getReflexPlatform' __useLegacyCompilers;
23 getReflexPlatform = getReflexPlatform' __useLegacyCompilers;
20 getReflexPlatform' = __useLegacyCompilers: sys: import ./dep/reflex-platform {
24 getReflexPlatform' = __useLegacyCompilers: sys: import ./dep/reflex-platform {
21 inherit iosSdkVersion __useLegacyCompilers;
25 inherit iosSdkVersion __useLegacyCompilers;
@@ -28,7 +32,7 @@ let
28 })
32 })
29 ];
33 ];
30
34
31− haskellOverlays = [
35+ haskellOverlays = with pkgs.haskell.lib; [
32
36
33 # Fix misc upstream packages
37 # Fix misc upstream packages
34 (self: super: let
38 (self: super: let
@@ -48,6 +52,42 @@ let
48 rev = "6a71119bfa5db2b9990a2491c941469ff8ef5d13";
52 rev = "6a71119bfa5db2b9990a2491c941469ff8ef5d13";
49 sha256 = "0z8smyainnlzcglv3dlx6x1n9j6d2jv48aa8f2421iayfkxg3js5";
53 sha256 = "0z8smyainnlzcglv3dlx6x1n9j6d2jv48aa8f2421iayfkxg3js5";
50 } + /template) {};
54 } + /template) {};
55+
56+ algebraic-graphs = pkgs.haskell.lib.doJailbreak
57+ (self.callCabal2nix "algebraic-graphs" (pkgs.fetchFromGitHub {
58+ owner = "snowleopard";
59+ repo = "alga";
60+ rev = "480a73137e9b38ad3f1bc2c628847953d2fb3e25";
61+ sha256 = "0dpwi5ffs88brl3lz51bwb004c6zm8ds8pkw1vzsg2a6aaiyhlzl";
62+ }) {});
63+
64+ # Need ShowTag, EqTag, and OrdTag instances
65+ dependent-sum-template = self.callCabal2nix "dependent-sum-template" (pkgs.fetchFromGitHub {
66+ owner = "mokus0";
67+ repo = "dependent-sum-template";
68+ rev = "dcb92a98d35d79f712b9a398c8c356acc979ecff";
69+ sha256 = "1v337d810d88jzfriw07pr16d34nibm6q2zkw787lc3sfn07glp5";
70+ }) {};
71+
72+ beam-core = dontCheck (self.callCabal2nix "beam-core" (beam-src + /beam-core) {});
73+ beam-postgres = dontCheck (self.callCabal2nix "beam-postgres" (beam-src + /beam-postgres) {});
74+ beam-migrate = self.callCabal2nix "beam-migrate" (beam-src + /beam-migrate) {};
75+
76+ websockets = self.callCabal2nix "websockets" (pkgs.fetchFromGitHub {
77+ owner = "obsidiansystems";
78+ repo = "websockets";
79+ rev = "b750edf10fc6f532f9fb36588904491af87606c7";
80+ sha256 = "1wd0gkx8a4x6n5cdkra96zvpv8l49i0z90amvrng70ybnzbxpfi5";
81+ }) {};
82+
83+ gargoyle = doJailbreak (self.callCabal2nix "gargoyle" (gargoyle-src + /gargoyle) {});
84+ gargoyle-postgresql-nix = pkgs.haskell.lib.addBuildTool
85+ (self.callCabal2nix "gargoyle-postgresql-nix" (gargoyle-src + /gargoyle-postgresql-nix) {})
86+ pkgs.postgresql100; # `staticWhich` requires `psql` on PATH during build time
87+ gargoyle-postgresql = addBuildDepend (doJailbreak (self.callCabal2nix "gargoyle-postgresql" (gargoyle-src + /gargoyle-postgresql) {})) pkgs.postgresql100;
88+ postgresql-libpq = enableCabalFlag (overrideCabal super.postgresql-libpq (drv: {
89+ pkgconfigDepends = (drv.pkgconfigDepends or []) ++ [pkgs.postgresql100]; #TODO: Obelisk should probably provide this version of postgresql
90+ })) "use-pkg-config";
51 })
91 })
52
92
53 # Add obelisk packages
93 # Add obelisk packages
@@ -63,6 +103,7 @@ let
63 obelisk-cliapp = self.callCabal2nix "obelisk-cliapp" (cleanSource ./lib/cliapp) {};
103 obelisk-cliapp = self.callCabal2nix "obelisk-cliapp" (cleanSource ./lib/cliapp) {};
64 obelisk-command = self.callCabal2nix "obelisk-command" (cleanSource ./lib/command) {};
104 obelisk-command = self.callCabal2nix "obelisk-command" (cleanSource ./lib/command) {};
65 obelisk-frontend = self.callCabal2nix "obelisk-frontend" (cleanSource ./lib/frontend) {};
105 obelisk-frontend = self.callCabal2nix "obelisk-frontend" (cleanSource ./lib/frontend) {};
106+ obelisk-request = self.callCabal2nix "obelisk-request" (cleanSource ./lib/request) {};
66 obelisk-run = self.callCabal2nix "obelisk-run" (cleanSource ./lib/run) {};
107 obelisk-run = self.callCabal2nix "obelisk-run" (cleanSource ./lib/run) {};
67 obelisk-route = self.callCabal2nix "obelisk-route" (cleanSource ./lib/route) {};
108 obelisk-route = self.callCabal2nix "obelisk-route" (cleanSource ./lib/route) {};
68 obelisk-selftest = self.callCabal2nix "obelisk-selftest" (cleanSource ./lib/selftest) {};
109 obelisk-selftest = self.callCabal2nix "obelisk-selftest" (cleanSource ./lib/selftest) {};
lib/backend/obelisk-backend.cabal+37−2
@@ -6,22 +6,57 @@ build-type: Simple
6library
6library
7 hs-source-dirs: src
7 hs-source-dirs: src
8 build-depends: base,
8 build-depends: base,
9+ aeson,
10+ async,
11+ attoparsec,
12+ binary,
9 bytestring,
13 bytestring,
10 categories,
14 categories,
15+ clientsession,
16+ constraints,
17+ containers,
11 data-default,
18 data-default,
12 dependent-sum,
19 dependent-sum,
13 dependent-sum-template,
20 dependent-sum-template,
21+ directory,
22+ --TODO: Do we really want enclosed-exceptions?
23+ enclosed-exceptions,
24+ filepath,
25+ gargoyle,
26+ gargoyle-postgresql-nix,
14 lens,
27 lens,
28+ monoidal-containers,
15 mtl,
29 mtl,
16 obelisk-asset-serve-snap,
30 obelisk-asset-serve-snap,
17 obelisk-executable-config-inject,
31 obelisk-executable-config-inject,
18 obelisk-frontend,
32 obelisk-frontend,
33+ obelisk-request,
19 obelisk-route,
34 obelisk-route,
20 obelisk-snap-extras,
35 obelisk-snap-extras,
36+ postgresql-libpq,
37+ postgresql-simple,
38+ reflex,
21 reflex-dom,
39 reflex-dom,
40+ resource-pool,
22 snap,
41 snap,
42+ snap-core,
23 snap-server,
43 snap-server,
44+ split,
45+ stm,
46+ template-haskell,
24 text,
47 text,
48+ these,
49+ time,
25− universe
50+ universe,
51+ unordered-containers,
52+ websockets,
53+ websockets-snap
26− exposed-modules: Obelisk.Backend
54+ exposed-modules:
55+ Data.MonoidMap
56+ Obelisk.Api
57+ Obelisk.Api.Pipeline
58+ Obelisk.Backend
59+ Obelisk.Db
60+ Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
61+ Obelisk.Postgres.Replication
27 ghc-options: -Wall
62 ghc-options: -Wall
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 $ MonoidalMap nil
45+ align (MonoidMap (MonoidalMap x)) (MonoidMap (MonoidalMap y)) = MonoidMap $ MonoidalMap $ align x y
46+ alignWith f (MonoidMap (MonoidalMap x)) (MonoidMap (MonoidalMap y)) = MonoidMap $ MonoidalMap $ alignWith f x y
47+
48+instance Ord k => Unalign (MonoidMap k)
lib/backend/src/Obelisk/Api.hsnew file+187
@@ -0,0 +1,187 @@
1+{-# LANGUAGE GADTs #-}
2+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
3+{-# LANGUAGE LambdaCase #-}
4+{-# LANGUAGE OverloadedStrings #-}
5+{-# LANGUAGE PatternSynonyms #-}
6+{-# LANGUAGE QuasiQuotes #-}
7+{-# LANGUAGE RankNTypes #-}
8+{-# LANGUAGE RecursiveDo #-}
9+{-# LANGUAGE ScopedTypeVariables #-}
10+module Obelisk.Api where
11+
12+import Prelude hiding ((.))
13+
14+import Obelisk.Api.Pipeline
15+import Obelisk.Db
16+import Obelisk.Postgres.Replication
17+import Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
18+import Obelisk.Request (Request)
19+
20+import Control.Category
21+import Control.Concurrent.Async
22+import Control.Concurrent.STM (TChan, atomically, readTChan, writeTChan, newTChanIO)
23+import Control.Monad (forever)
24+import Control.Monad.Reader
25+import Control.Monad.Trans (lift)
26+import Data.Aeson (FromJSON, ToJSON)
27+import Data.Attoparsec.ByteString
28+import Data.ByteString (ByteString)
29+import Data.Default
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)
39+
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+
90+newtype ReadDb a = ReadDb { unReadDb :: ReaderT PG.Connection IO a }
91+ deriving (Functor, Applicative, Monad)
92+
93+readTransaction :: PG.Connection -> ReadDb a -> IO a
94+readTransaction conn (ReadDb r) = PG.withTransactionModeRetry m PG.isSerializationError conn $ runReaderT r conn
95+ where m = PG.TransactionMode
96+ { PG.isolationLevel = PG.Serializable
97+ , PG.readWriteMode = PG.ReadOnly
98+ }
99+
100+newtype WriteDb a = WriteDb { unWriteDb :: ReaderT PG.Connection IO a }
101+ deriving (Functor, Applicative, Monad)
102+
103+writeTransaction :: PG.Connection -> WriteDb a -> IO a
104+writeTransaction conn (WriteDb r) = PG.withTransactionModeRetry m PG.isSerializationError conn $ runReaderT r conn
105+ where m = PG.TransactionMode
106+ { PG.isolationLevel = PG.Serializable
107+ , PG.readWriteMode = PG.ReadWrite
108+ }
109+
110+unsafeReadDb :: (PG.Connection -> IO a) -> ReadDb a
111+unsafeReadDb f = ReadDb $ lift . f =<< ask
112+
113+unsafeWriteDb :: (PG.Connection -> IO a) -> WriteDb a
114+unsafeWriteDb f = WriteDb $ lift . f =<< ask
115+
116+withApiHandler
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 ())
134+ , Request request
135+ )
136+ => ByteString -- ^ The URI of the DB to connect to
137+ -> (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
140+ -> (Snap () -> IO a)
141+ -> IO a
142+withApiHandler dbUri handleApi handleChange handleQuery go = do
143+ withNonEmptyTransactions dbUri $ \transactions -> do
144+ 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
150+ (qh', r) <- unPipeline pipe qh r'
151+ (r', handleListen) <- connectPipelineToWebsockets "" (\r'' -> withResource db $ \conn -> writeTransaction conn $ handleApi r'') qh'
152+ go handleListen
153+
154+withNonEmptyTransactions :: ByteString -> (TChan Transaction -> IO a) -> IO a
155+withNonEmptyTransactions dbUri go = do
156+ let opts = def
157+ { _logicalDecodingOptions_pluginOptions = [("skip-empty-xacts", Nothing)]
158+ }
159+ withLogicalDecoding dbUri "test_decoding" opts $ \decodedLines -> do
160+ transactions <- newTChanIO
161+ processLine <- linesToTransactions
162+ let processLines = forever $ do
163+ lineRaw <- atomically $ readTChan decodedLines
164+ let Right l = parseOnly (line <* endOfInput) lineRaw
165+ Right mTransaction <- processLine l
166+ forM_ mTransaction $ \transaction ->
167+ atomically $ writeTChan transactions transaction
168+ withAsync processLines $ \a -> link a >> do
169+ 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+-}
lib/backend/src/Obelisk/Api/Pipeline.hsnew file+408
@@ -0,0 +1,408 @@
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 Obelisk.Request (Request, SomeRequest (..))
13+
14+import Control.Concurrent
15+import Control.Monad
16+import Control.Monad.State.Strict
17+import Data.Typeable
18+import GHC.Generics
19+import Control.Lens (imapM_)
20+import Data.Aeson
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+-- | A way for a pipeline to retrieve data
41+newtype QueryHandler q m = QueryHandler
42+ { runQueryHandler :: q -> m (QueryResult q) }
43+
44+-- | A way to push data into a pipeline
45+newtype Recipient q m = Recipient
46+ { tellRecipient :: QueryResult q -> m () }
47+
48+-- | A way of attaching to (and later detaching from) a pipeline
49+newtype Registrar q = Registrar { unRegistrar :: Recipient q IO -> IO (QueryHandler q IO, IO ()) }
50+
51+-- | A way of connecting a source of data to a consumer of data. The consumer
52+-- can pull data from the datasource and the datasource can push data to the
53+-- consumer.
54+--
55+-- q is the consumer side
56+-- q' is the datasource side
57+newtype Pipeline m q q' = Pipeline { unPipeline :: QueryHandler q' m -> Recipient q m -> IO (QueryHandler q m, Recipient q' m) }
58+
59+instance Category (Pipeline m) where
60+ id = Pipeline $ \qh r -> return (qh, r)
61+ Pipeline f . Pipeline g = Pipeline $ \qh r -> do
62+ rec (qh'', r') <- g qh' r
63+ (qh', r'') <- f qh r'
64+ return (qh'', r'')
65+
66+-- | A key used to track particular consumers
67+newtype ClientKey = ClientKey { unClientKey :: Integer }
68+ deriving (Eq, Ord, Show)
69+
70+-- | Produce a multi-client 'Recipient' and a single client 'QueryHandler'
71+fanQuery
72+ :: forall k q. (Ord k, Monoid (QueryResult q))
73+ => (k -> IO (Recipient q IO))
74+ -- ^ Look up a recipient
75+ -> QueryHandler (MonoidalMap k q) IO
76+ -- ^ A 'QueryHandler' for multiple clients
77+ -> ( Recipient (MonoidalMap k q) IO
78+ -- Used to notify recipients of new 'QueryResult' data
79+ , k -> QueryHandler q IO
80+ -- Used by 'multiplexQuery' to lookup the 'QueryHandler' for a given client
81+ )
82+fanQuery lookupRecipient qh = (multiRecipient lookupRecipient, fanQueryHandler qh)
83+ where
84+ -- | Given a recipient lookup function, produces a 'Recipient' that can
85+ -- handle distribution of a map of 'QueryResult's to many recipients
86+ multiRecipient lookupR = Recipient $ imapM_ $ \k qr -> do
87+ s <- lookupR k
88+ tellRecipient s qr
89+ -- | Given a multi-client 'QueryHandler', produces a 'QueryHandler' for
90+ -- a single client
91+ fanQueryHandler qh' = \k -> mapQueryHandler (singletonQuery k) qh'
92+
93+-- | Maintains a Map from connected clients to a 'Recipient' that can be used to transmit data to the clients
94+--
95+-- Takes a function that looks up a the 'QueryHandler' for a given client
96+-- Returns: 1. A lookup function into the Map of clients
97+-- 2. A way to register a new client
98+-- a. A 'QueryHandler' for the newly registered client
99+-- b. A removal callback to de-register a particular client
100+multiplexQuery
101+ :: (MonadIO m, Group q)
102+ => (ClientKey -> QueryHandler q m)
103+ -> IO ( ClientKey -> IO (Recipient q m)
104+ , Recipient q m -> IO (QueryHandler q m, m ())
105+ )
106+multiplexQuery lookupQueryHandler = do
107+ clients <- newIORef (ClientKey 0, Map.empty :: MonoidalMap ClientKey (Recipient q m, q))
108+ let
109+ lookupRecipient k = do
110+ (_, cs) <- readIORef clients
111+ case Map.lookup k cs of
112+ Nothing -> do
113+ putStrLn $ mconcat
114+ [ "Rhyolite.Backend.App.multiplexQuery: Failed to find sender for key"
115+ , show k
116+ , " in known keys "
117+ , show $ Map.keys cs
118+ ]
119+ return $ Recipient $ const $ return ()
120+ Just s -> return $ fst s
121+
122+ registerRecipient s = do
123+ cid <- atomicModifyIORef' clients $ \(cid, recipients) ->
124+ ((ClientKey (unClientKey cid + 1), Map.insert cid (s, mempty) recipients), cid)
125+ let
126+ queryHandler = QueryHandler $ \q -> do
127+ liftIO $ atomicModifyIORef' clients $ \(nextCid, recipients) ->
128+ ((nextCid, Map.update (\(r, oldQ) -> Just (r, oldQ <> q)) cid recipients), ())
129+ runQueryHandler (lookupQueryHandler cid) q
130+
131+ unregisterRecipient = do
132+ antiQ <- liftIO $ atomicModifyIORef' clients $ \(nextCid, recipients) ->
133+ case Map.updateLookupWithKey (\_ _ -> Nothing) cid recipients of
134+ (Nothing, _) -> trace
135+ ("Rhyolite.Backend.App.multiplexQuery: Tried to unregister a client key that is not registered " <> show cid)
136+ ((nextCid, recipients), mempty)
137+ (Just (_, removedQuery), newRecipients) -> ((nextCid, newRecipients), negateG removedQuery)
138+
139+ -- TODO: Should we have a way of ensuring that this doesn't actually cause a query to be run?
140+ -- It shouldn't cause the query to be run again but it depends on if the callee will notice
141+ -- that the new query is strictly smaller than the old one.
142+ _ <- runQueryHandler (lookupQueryHandler cid) antiQ
143+ return ()
144+
145+ return (queryHandler, unregisterRecipient)
146+
147+ return (lookupRecipient, registerRecipient)
148+
149+-- | Represents a WebSocket message from one of two channels: ViewSelector declarations or API requests
150+data WebSocketRequest q r
151+ = WebSocketRequest_ViewSelector q
152+ | WebSocketRequest_Api (TaggedRequest r)
153+ deriving (Typeable, Generic)
154+
155+instance (Request r, FromJSON q) => FromJSON (WebSocketRequest q r)
156+instance (Request r, ToJSON q) => ToJSON (WebSocketRequest q r)
157+
158+-- | Represents a WebSocket response from one of three channels: incoming 'View's, API responses, or version info
159+data WebSocketResponse q
160+ = WebSocketResponse_View (QueryResult q)
161+ | WebSocketResponse_Api TaggedResponse
162+ | WebSocketResponse_Version Text
163+ deriving (Typeable, Generic)
164+
165+instance FromJSON (QueryResult q) => FromJSON (WebSocketResponse q)
166+instance ToJSON (QueryResult q) => ToJSON (WebSocketResponse q)
167+
168+-- | A request tagged with an identifier
169+data TaggedRequest r = TaggedRequest Value (SomeRequest r)
170+ deriving (Typeable, Generic)
171+
172+instance Request r => FromJSON (TaggedRequest r)
173+instance Request r => ToJSON (TaggedRequest r)
174+
175+-- | A response tagged with an identifier matching the one in the 'TaggedRequest'. The identifier is the first argument.
176+data TaggedResponse = TaggedResponse Value Value
177+ deriving (Typeable, Generic)
178+
179+instance FromJSON TaggedResponse
180+instance ToJSON TaggedResponse
181+
182+-- | Handles a websocket connection
183+handleWebsocket
184+ :: forall q qr r.
185+ ( Eq (q SelectedCount)
186+ , Eq (qr SelectedCount)
187+ , QueryResult (q SelectedCount) ~ qr SelectedCount
188+ , QueryResult (q ()) ~ qr ()
189+ , ToJSON (qr ())
190+ , Functor qr
191+ , Group (q SelectedCount)
192+ , Monoid (qr SelectedCount)
193+ , FromJSON (q ())
194+ , Request r
195+ , Functor q
196+ )
197+ => Text -- ^ Version
198+ -> (forall a. r a -> IO a)
199+ -> Registrar (q SelectedCount)
200+ -> Snap ()
201+handleWebsocket v rh register = withWebsocketsConnection $ \conn -> do
202+ let sender = Recipient $ sendEncodedDataMessage conn . (\a -> WebSocketResponse_View (void a) :: WebSocketResponse (q ()))
203+ sendEncodedDataMessage conn (WebSocketResponse_Version v :: WebSocketResponse (q ()))
204+ bracket (unRegistrar register sender) snd $ \(vsHandler, _) -> flip evalStateT mempty $ forever $ do
205+ (wsr :: WebSocketRequest (q ()) r) <- liftIO $ getDataMessage conn
206+ case wsr of
207+ WebSocketRequest_Api (TaggedRequest reqId (SomeRequest req)) -> lift $ do
208+ a <- rh req
209+ sendEncodedDataMessage conn
210+ (WebSocketResponse_Api $ TaggedResponse reqId (toJSON a) :: WebSocketResponse (q ()))
211+ WebSocketRequest_ViewSelector new -> do
212+ old <- get
213+ let new' = SelectedCount 1 <$ new
214+ vsDiff = new' ~~ old
215+ when (vsDiff /= mempty) $ do
216+ qr <- lift $ runQueryHandler vsHandler vsDiff
217+ put new'
218+ when (qr /= mempty) $ lift $
219+ sendEncodedDataMessage conn (WebSocketResponse_View (void qr) :: WebSocketResponse (q ()))
220+
221+-------------------------------------------------------------------------------
222+
223+-- | Connect a datasource (e.g., a database) to a pipeline
224+--
225+-- Data taken from 'getNextNotification' is pushed into the pipeline and
226+-- when the pipeline pulls data, it is retrieved using 'qh'
227+feedPipeline
228+ :: Monoid q
229+ => IO (q -> IO (QueryResult q))
230+ -- ^ Get the next notification to be sent to the pipeline. If no notification
231+ -- is available, this should block until one is available
232+ -> QueryHandler q IO
233+ -- ^ Retrieve data when requested by pipeline
234+ -> Recipient q IO
235+ -- ^ A way to push data into the pipeline
236+ -> IO (QueryHandler q IO, IO ())
237+ -- ^ A way for the pipeline to request data
238+feedPipeline getNextNotification qh r = do
239+ currentQuery <- newIORef mempty
240+ let qhSaveQuery = QueryHandler $ \q -> do
241+ atomicModifyIORef' currentQuery $ \old -> (q <> old, ())
242+ runQueryHandler qh q
243+ tid <- forkIO . forever $ do
244+ nm <- getNextNotification
245+ q <- readIORef currentQuery
246+ qr <- nm q
247+ tellRecipient r qr
248+ return (qhSaveQuery, killThread tid)
249+
250+-- | Connects the pipeline to websockets consumers
251+connectPipelineToWebsockets
252+ :: ( Eq (q SelectedCount)
253+ , QueryResult (q SelectedCount) ~ qr SelectedCount
254+ , QueryResult (q ()) ~ qr ()
255+ , Group (q SelectedCount)
256+ , Monoid (qr SelectedCount)
257+ , Eq (qr SelectedCount)
258+ , ToJSON (qr ())
259+ , FromJSON (q ())
260+ , Request r
261+ , Functor qr
262+ , Functor q
263+ )
264+ => Text
265+ -> (forall a. r a -> IO a)
266+ -- ^ API handler
267+ -> QueryHandler (MonoidalMap ClientKey (q SelectedCount)) IO
268+ -- ^ A way to retrieve more data for each consumer
269+ -> IO (Recipient (MonoidalMap ClientKey (q SelectedCount)) IO, Snap ())
270+ -- ^ A way to send data to many consumers and a handler for websockets connections
271+connectPipelineToWebsockets ver rh qh = do
272+ (allRecipients, registerRecipient) <- connectPipelineToWebsockets' qh
273+ return (allRecipients, handleWebsocket ver rh registerRecipient)
274+
275+-- | Like 'connectPipelineToWebsockets' but returns a Registrar that can
276+-- be used to construct a handler for a particular client
277+connectPipelineToWebsockets'
278+ :: (Monoid (QueryResult q), Group q)
279+ => QueryHandler (MonoidalMap ClientKey q) IO
280+ -> IO (Recipient (MonoidalMap ClientKey q) IO, Registrar q)
281+ -- ^ A way to send data to many consumers, and a way to register new consumers
282+connectPipelineToWebsockets' qh = do
283+ rec (lookupRecipient, registerRecipient) <- multiplexQuery clientQueryHandler
284+ let (allRecipients, clientQueryHandler) = fanQuery lookupRecipient qh
285+ return (allRecipients, Registrar registerRecipient)
286+
287+-- | Extends a 'Registrar' with a 'Pipeline'
288+extendRegistrar :: Pipeline IO q q' -> Registrar q' -> Registrar q
289+extendRegistrar (Pipeline p) (Registrar r) = Registrar $ \recipient -> do
290+ rec (qh', close) <- r recipient'
291+ (qh, recipient') <- p qh' recipient
292+ return (qh, close)
293+
294+-------------------------------------------------------------------------------
295+
296+singletonQuery
297+ :: ( Monoid (QueryResult q)
298+ , Ord k
299+ )
300+ => k
301+ -> QueryMorphism q (MonoidalMap k q)
302+singletonQuery k = QueryMorphism { _queryMorphism_mapQuery = Map.singleton k
303+ , _queryMorphism_mapQueryResult = Map.findWithDefault mempty k
304+ }
305+
306+monoidMapQueryMorphism :: (Eq q, Monoid q) => QueryMorphism (MonoidalMap k q) (MonoidMap k q)
307+monoidMapQueryMorphism = QueryMorphism
308+ { _queryMorphism_mapQuery = monoidMap
309+ , _queryMorphism_mapQueryResult = unMonoidMap
310+ }
311+
312+transposeMonoidMap
313+ :: forall a a' k q qr.
314+ ( Ord k
315+ , Eq a
316+ , Monoid a
317+ , Monoid a'
318+ , Eq (q (MonoidMap k a))
319+ , Eq (QueryResult (q a))
320+ , Foldable qr
321+ , Functor q
322+ , Functor qr
323+ , FunctorMaybe qr
324+ , Monoid (q (MonoidMap k a))
325+ , Monoid (QueryResult (q a))
326+ , QueryResult (q (MonoidMap k a)) ~ qr (MonoidMap k a')
327+ , QueryResult (q a) ~ qr a'
328+ )
329+ => QueryMorphism (MonoidMap k (q a)) (q (MonoidMap k a))
330+transposeMonoidMap = QueryMorphism
331+ { _queryMorphism_mapQuery = aggregateQueries
332+ , _queryMorphism_mapQueryResult = distributeResults
333+ }
334+ where
335+ aggregateQueries :: MonoidMap k (q a) -> q (MonoidMap k a)
336+ aggregateQueries = fold . monoidMap . Map.mapWithKey (\k q -> fmap (monoidMap . Map.singleton k) q) . unMonoidMap
337+ distributeResults :: qr (MonoidMap k a') -> MonoidMap k (qr a')
338+ distributeResults v = monoidMap $ Map.mapWithKey (\k _ -> fmapMaybe (Map.lookup k . unMonoidMap) v) $ fold $ fmap unMonoidMap v
339+
340+mapQueryHandlerAndRecipient
341+ :: Functor f
342+ => QueryMorphism q q'
343+ -> QueryHandler q' f
344+ -> Recipient q f
345+ -> (QueryHandler q f, Recipient q' f)
346+mapQueryHandlerAndRecipient qm qh s = (mapQueryHandler qm qh, mapRecipient qm s)
347+
348+mapQueryHandler :: Functor f => QueryMorphism q q' -> QueryHandler q' f -> QueryHandler q f
349+mapQueryHandler qm qh = QueryHandler $ \qs -> mapQueryResult qm <$> runQueryHandler qh (mapQuery qm qs)
350+
351+mapRecipient :: QueryMorphism q q' -> Recipient q m -> Recipient q' m
352+mapRecipient qm s = Recipient $ \qr -> tellRecipient s $ mapQueryResult qm qr
353+
354+fmapQueryMorphism
355+ :: ( Functor f, QueryResult (f q) ~ f (QueryResult q)
356+ , QueryResult (f q') ~ f (QueryResult q') )
357+ => QueryMorphism q q'
358+ -> QueryMorphism (f q) (f q')
359+fmapQueryMorphism qm = QueryMorphism
360+ { _queryMorphism_mapQuery = fmap $ _queryMorphism_mapQuery qm
361+ , _queryMorphism_mapQueryResult = fmap $ _queryMorphism_mapQueryResult qm
362+ }
363+
364+queryMorphismPipeline :: Functor m => QueryMorphism q q' -> Pipeline m q q'
365+queryMorphismPipeline qm = Pipeline $ \qh r -> pure $ mapQueryHandlerAndRecipient qm qh r
366+
367+-- | Accepts a websockets connection and runs the supplied action with it
368+withWebsocketsConnection :: MonadSnap m => (WS.Connection -> IO ()) -> m ()
369+withWebsocketsConnection f = runWebSocketsSnap $ \pc -> do
370+ conn <- WS.acceptRequest pc
371+ handleSomeException $ handleConnectionException pc $ f conn
372+ where
373+ handleSomeException = handle $ \(SomeException e) -> putStrLn $ "withWebsocketsConnection: " <> displayException e
374+ handleConnectionException pc = handle $ \e -> case e of
375+ WS.ConnectionClosed -> return ()
376+ WS.CloseRequest _ _ -> print e >> WS.close (WS.pendingStream pc) >> throwIO e
377+ _ -> do putStr $ "withWebsocketsConnection: Exception: " <> displayException e
378+ throwIO e
379+
380+-- | Attempts to json decode a websockets data message
381+decodeWebsocketsDataMessage :: FromJSON a => WS.DataMessage -> Either String a
382+decodeWebsocketsDataMessage dm = eitherDecode' $ case dm of
383+ WS.Text r' _ -> r'
384+ WS.Binary r' -> r'
385+
386+-- | Parse and process a single websocket data message
387+getDataMessage
388+ :: FromJSON a
389+ => WS.Connection
390+ -> IO a
391+getDataMessage conn = do
392+ dm <- WS.receiveDataMessage conn
393+ case decodeWebsocketsDataMessage dm of
394+ Left err -> liftIO $ throwIO $ AssertionFailed $ mconcat
395+ [ "getDataMessage: error: "
396+ , err
397+ , "; received: "
398+ , show dm
399+ ]
400+ Right a -> return a
401+
402+-- | Send a json encoded data message over the websocket connection
403+sendEncodedDataMessage
404+ :: ToJSON a
405+ => WS.Connection
406+ -> a
407+ -> IO ()
408+sendEncodedDataMessage conn = WS.sendDataMessage conn . (\x -> WS.Text x Nothing) . encode
lib/backend/src/Obelisk/Db.hsnew file+51
@@ -0,0 +1,51 @@
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\nmax_wal_senders = 10\n"
40+ appendFile (workDir </> "pg_hba.conf") "\n# Added by Obelisk.Db\nlocal replication postgres trust\n"
41+ }
42+ withGargoyle myConfig dbPath a
43+
44+openDbUri :: ByteString -> IO (Pool PG.Connection)
45+openDbUri dbUri = do
46+ let openPostgresql = connectPostgreSQL dbUri
47+ closePostgresql p = close p
48+ createPool openPostgresql closePostgresql 1 5 20
49+
50+withConnectionPool :: ByteString -> (Pool PG.Connection -> IO a) -> IO a
51+withConnectionPool dbUri = bracket (openDbUri dbUri) destroyAllResources
lib/backend/src/Obelisk/Postgres/LogicalDecoding/Plugins/TestDecoding.hsnew file+219
@@ -0,0 +1,219 @@
1+{-# LANGUAGE OverloadedStrings #-}
2+{-# LANGUAGE LambdaCase #-}
3+module Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
4+ ( line
5+ , Transaction
6+ , Change (..)
7+ , LineError (..)
8+ , linesToTransactions
9+ ) where
10+
11+import Prelude hiding (takeWhile)
12+
13+import Control.Applicative
14+import Control.Monad
15+import Data.Attoparsec.ByteString (word8, notWord8)
16+import Data.Attoparsec.ByteString.Char8
17+import Data.ByteString (ByteString)
18+import qualified Data.ByteString as BS
19+import Data.Char (ord)
20+import qualified Data.Char as Char
21+import Data.Word
22+import Data.HashMap.Strict (HashMap)
23+import qualified Data.HashMap.Strict as HashMap
24+import Data.Sequence (Seq, (|>))
25+import Data.Text (Text)
26+import Data.Text.Encoding
27+import Database.PostgreSQL.Simple.Types
28+import qualified Data.Map as Map
29+import Data.Map (Map)
30+import Data.IORef
31+import Data.Maybe
32+
33+--------------------------------------------------------------------------------
34+-- Potentially upstreamable
35+--------------------------------------------------------------------------------
36+
37+stringWithSingleQuote :: Parser ByteString
38+stringWithSingleQuote = stringWithQuote $ fromIntegral $ ord '\''
39+
40+stringWithDoubleQuote :: Parser ByteString
41+stringWithDoubleQuote = stringWithQuote $ fromIntegral $ ord '"'
42+
43+stringWithQuote :: Word8 -> Parser ByteString
44+stringWithQuote q = do
45+ _ <- word8 q
46+ s <- many $ notWord8 q <|> doubledWord8 q
47+ _ <- word8 q
48+ pure $ BS.pack s
49+
50+doubledWord8 :: Word8 -> Parser Word8
51+doubledWord8 c = do
52+ _ <- word8 c
53+ _ <- word8 c
54+ pure c
55+
56+identifierLike :: (Char -> Bool) -> (Char -> Bool) -> Parser Text
57+identifierLike validFirstChar validOtherChar = fmap decodeUtf8 stringWithDoubleQuote <|> do
58+ -- Avoid plucking the first character off, so that we don't have to cons it back on (which would reallocate the ByteString)
59+ guard . validFirstChar =<< peekChar'
60+ decodeUtf8 <$> takeWhile1 validOtherChar
61+
62+identifier :: Parser Identifier
63+identifier = Identifier <$> identifierLike identifierFirstChar identifierOtherChar
64+
65+identifierFirstChar :: Char -> Bool
66+identifierFirstChar c = Char.isLetter c || c == '_'
67+
68+identifierOtherChar :: Char -> Bool
69+identifierOtherChar c = identifierFirstChar c || isDigit c || c == '$'
70+
71+qualifiedIdentifier :: Parser QualifiedIdentifier
72+qualifiedIdentifier = do
73+ namespace <- identifier
74+ _ <- char '.'
75+ name <- identifier
76+ pure $ QualifiedIdentifier (Just $ fromIdentifier namespace) $ fromIdentifier name
77+
78+newtype TypeName = TypeName { unTypeName :: Text } deriving (Show, Read, Eq, Ord)
79+
80+typeName :: Parser TypeName
81+typeName = do
82+ let otherChar c = identifierOtherChar c || c `elem` (" ()," :: String)
83+ TypeName <$> identifierLike identifierFirstChar otherChar
84+
85+newtype Xid = Xid { unXid :: Word64 } deriving (Show, Read, Eq, Ord)
86+
87+xid :: Parser Xid
88+xid = Xid <$> decimal
89+
90+--------------------------------------------------------------------------------
91+
92+rowEntry :: Parser a -> Parser (Identifier, (TypeName, a))
93+rowEntry item = do
94+ colName <- identifier
95+ _ <- char '['
96+ colType <- typeName
97+ _ <- string "]:"
98+ v <- item
99+ pure (colName, (colType, v))
100+
101+rowLike :: Parser a -> Parser (HashMap Identifier (TypeName, a))
102+rowLike item = {- (mempty <$ string "(no-tuple-data)") <|> -} do
103+ fmap HashMap.fromList $ flip sepBy1 (char ' ') $ rowEntry item
104+
105+type Row = HashMap Identifier (TypeName, Maybe Literal)
106+
107+data Literal
108+ = Literal_UnchangedToastDatum
109+ | Literal_Present ByteString
110+ deriving (Show, Read, Eq, Ord)
111+
112+nullableLiteral :: Parser (Maybe Literal)
113+nullableLiteral = Nothing <$ string "null" <|> Just <$> literal
114+
115+literal :: Parser Literal
116+literal = unchangedToastDatum <|> presentDatum
117+ where
118+ unchangedToastDatum = Literal_UnchangedToastDatum <$ string "unchanged-toast-datum"
119+ -- | We convert all present datums to their normal Postgres ByteString
120+ -- encoding, so that we don't have to preserve special cases throughout our
121+ -- code
122+ presentDatum :: Parser Literal
123+ presentDatum = Literal_Present <$> foldr1 (<|>)
124+ [ stringWithSingleQuote
125+ , char 'B' *> stringWithSingleQuote
126+ , "t" <$ string "true"
127+ , "f" <$ string "false"
128+ -- We assume anything else must be numeric
129+ , takeWhile1 (/= ' ')
130+ ]
131+
132+row :: Parser Row
133+row = rowLike nullableLiteral
134+
135+type Key = HashMap Identifier (TypeName, Literal) -- Keys can't contain nulls
136+
137+key :: Parser Key
138+key = rowLike literal
139+
140+data Change
141+ = Change_Insert Row
142+ | Change_Update Key Row
143+ | Change_Delete Key
144+ deriving (Show, Read, Eq, Ord)
145+
146+change :: Parser Change
147+change = do
148+ t <- takeWhile (/= ':')
149+ _ <- string ": "
150+ case t of
151+ "INSERT" -> Change_Insert <$> row
152+ "UPDATE" -> foldr1 (<|>)
153+ [ do _ <- string "old-key: "
154+ old <- key
155+ _ <- string " new-tuple: "
156+ new <- row
157+ pure $ Change_Update old new
158+ , Change_Update mempty <$> row
159+ ]
160+ "DELETE" -> Change_Delete <$> key
161+ _ -> fail $ "Unrecognized change type: " <> show t
162+
163+data Line
164+ = Line_Begin Xid
165+ | Line_Change QualifiedIdentifier Change
166+ | Line_Commit Xid
167+ deriving (Show, Read, Eq, Ord)
168+
169+line :: Parser Line
170+line = do
171+ leader <- takeWhile (/= ' ')
172+ _ <- char ' '
173+ case leader of
174+ "BEGIN" -> Line_Begin <$> xid
175+ "table" -> do
176+ table <- qualifiedIdentifier
177+ _ <- string ": "
178+ Line_Change table <$> change
179+ "COMMIT" -> Line_Commit <$> xid
180+ _ -> fail $ "Unrecognized line leader: " <> show leader
181+
182+type Transaction = (Xid, Map QualifiedIdentifier (Seq Change))
183+
184+data LineError
185+ = LineError_OtherTransactionAlreadyInProgress
186+ Xid -- Existing transaction ID
187+ Xid -- New transaction ID that tried to start
188+ | LineError_CommittingWithoutTransaction
189+ Xid -- Transaction ID that tried to commit
190+ | LineError_CommittingWrongTransaction
191+ Xid -- Existing transaction ID
192+ Xid -- New transaction ID that tried to start
193+ | LineError_ChangeOutsideTransaction
194+
195+-- | Create a function that will statefully accumulate lines until a Transaction
196+-- is produced. If the given line is not valid in the current state, an error
197+-- will be produced and the line will be ignored. This should never happen for
198+-- a valid LogicalDecoding session, so all errors should be treated as serious.
199+linesToTransactions :: IO (Line -> IO (Either LineError (Maybe Transaction)))
200+linesToTransactions = do
201+ pending <- newIORef Nothing
202+ return $ \l -> do
203+ let -- Update the pending transaction and possibly emit a message
204+ update :: Maybe Transaction -> Either LineError (Maybe Transaction, Maybe Transaction)
205+ update = case l of
206+ Line_Begin newXid -> \case
207+ Nothing -> Right (Just (newXid, mempty), Nothing)
208+ Just (existingXid, _) -> Left $ LineError_OtherTransactionAlreadyInProgress existingXid newXid
209+ Line_Commit newXid -> \case
210+ Just txn@(existingXid, _)
211+ | existingXid == newXid -> Right (Nothing, Just txn)
212+ | otherwise -> Left $ LineError_CommittingWrongTransaction existingXid newXid
213+ Nothing -> Left $ LineError_CommittingWithoutTransaction newXid
214+ Line_Change qid c -> \old -> case old of
215+ Just (existingXid, changes) -> Right (Just (existingXid, Map.alter (Just . (|> c) . fromMaybe mempty) qid changes), Nothing)
216+ Nothing -> Left LineError_ChangeOutsideTransaction
217+ atomicModifyIORef pending $ \old -> case update old of
218+ Left e -> (old, Left e)
219+ Right (new, a) -> (new, Right a)
lib/backend/src/Obelisk/Postgres/QQ.hsnew file+69
@@ -0,0 +1,69 @@
1+{-# LANGUAGE TemplateHaskellQuotes #-}
2+module Obelisk.Postgres.QQ
3+ ( queryQ
4+ , executeQ
5+ , sqlQ
6+ , traceQueryQ
7+ ) where
8+
9+import Database.PostgreSQL.Simple.SqlQQ (sql)
10+import Database.PostgreSQL.Simple.ToField (toField, Action)
11+import Language.Haskell.TH (Exp, Name, Q, appE, mkName, tupE, varE, listE, sigE)
12+import Language.Haskell.TH.Quote (QuasiQuoter (..))
13+
14+-- | This quasiquoter is the obvious combination of 'sqlQ' and 'query'.
15+queryQ :: QuasiQuoter
16+queryQ = QuasiQuoter
17+ { quotePat = error "queryQ: quasiquoter used in pattern context"
18+ , quoteType = error "queryQ: quasiquoter used in type context"
19+ , quoteExp = \s -> appE [| uncurry query |] (sqlQExp s)
20+ , quoteDec = error "queryQ: quasiquoter used in declaration context"
21+ }
22+
23+-- | This quasiquoter is the obvious combination of 'sqlQ' and 'execute'.
24+executeQ :: QuasiQuoter
25+executeQ = QuasiQuoter
26+ { quotePat = error "executeQ: quasiquoter used in pattern context"
27+ , quoteType = error "executeQ: quasiquoter used in type context"
28+ , quoteExp = \s -> appE [| uncurry execute |] (sqlQExp s)
29+ , quoteDec = error "executeQ: quasiquoter used in declaration context"
30+ }
31+
32+traceQueryQ :: QuasiQuoter
33+traceQueryQ = QuasiQuoter
34+ { quotePat = error "traceQueryQ: quasiquoter used in pattern context"
35+ , quoteType = error "traceQueryQ: quasiquoter used in type context"
36+ , quoteExp = \s -> appE [| uncurry traceQuery |] (sqlQExp s)
37+ , quoteDec = error "traceQueryQ: quasiquoter used in declaration context"
38+ }
39+-- | This quasiquoter takes a SQL query with named arguments in the form "?var" and generates a pair
40+-- consisting of the Query string itself and a tuple of variables in corresponding order.
41+--
42+-- For example: uncurry query [sqlQ| SELECT * FROM 'Book' b WHERE b.title = ?title AND b.author = ?author |]
43+--
44+-- will be equivalent to query [sql| SELECT * FROM 'Book' b WHERE b.title = ? AND b.author = ? |] [toField title, toField author]
45+sqlQ :: QuasiQuoter
46+sqlQ = QuasiQuoter
47+ { quotePat = error "sqlQ: quasiquoter used in pattern context"
48+ , quoteType = error "sqlQ: quasiquoter used in type context"
49+ , quoteExp = sqlQExp
50+ , quoteDec = error "sqlQ: quasiquoter used in declaration context"
51+ }
52+
53+sqlQExp :: String -> Q Exp
54+sqlQExp s =
55+ let (s',vs) = extractVars s
56+ in tupE [quoteExp sql s', sigE (listE $ map (appE (varE 'toField) . varE) vs) [t| [Action] |]]
57+
58+extractVars :: String -> (String, [Name])
59+extractVars = extractVars'
60+ where
61+ extractVars' [] = ([],[])
62+ extractVars' ('?':s') =
63+ let [(var,rest)] = lex s'
64+ (s'',vars) = extractVars' rest
65+ in ('?':s'', mkName var : vars)
66+ extractVars' s' =
67+ let (pre,post) = break (=='?') s'
68+ (s'',vars) = extractVars' post
69+ in (pre ++ s'', vars)
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 Obelisk.Postgres.QQ
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
lib/request/obelisk-request.cabalnew file+18
@@ -0,0 +1,18 @@
1+name: obelisk-request
2+version: 0.1
3+cabal-version: >= 1.2
4+build-type: Simple
5+
6+library
7+ hs-source-dirs: src
8+ build-depends: base,
9+ aeson,
10+ constraints,
11+ template-haskell,
12+ vector
13+ exposed-modules:
14+ Obelisk.Request
15+ Obelisk.Request.TH
16+ other-modules:
17+ Obelisk.Request.HList
18+ ghc-options: -Wall
lib/request/src/Obelisk/Request.hsnew file+25
@@ -0,0 +1,25 @@
1+{-# LANGUAGE GADTs #-}
2+
3+module Obelisk.Request
4+ ( Request (..)
5+ , SomeRequest (..)
6+ ) where
7+
8+import Data.Aeson (FromJSON, ToJSON, Value, parseJSON, toJSON)
9+import Data.Aeson.Types (Parser)
10+import Data.Constraint (Dict)
11+
12+class Request r where
13+ requestToJSON :: r a -> Value
14+ requestParseJSON :: Value -> Parser (SomeRequest r)
15+ requestResponseToJSON :: r a -> Dict (ToJSON a)
16+ requestResponseFromJSON :: r a -> Dict (FromJSON a)
17+
18+data SomeRequest t where
19+ SomeRequest :: (FromJSON x, ToJSON x) => t x -> SomeRequest t
20+
21+instance Request r => FromJSON (SomeRequest r) where
22+ parseJSON = requestParseJSON
23+
24+instance Request r => ToJSON (SomeRequest r) where
25+ toJSON (SomeRequest r) = requestToJSON r
lib/request/src/Obelisk/Request/HList.hsnew file+59
@@ -0,0 +1,59 @@
1+{-# LANGUAGE DataKinds #-}
2+{-# LANGUAGE FlexibleContexts #-}
3+{-# LANGUAGE FlexibleInstances #-}
4+{-# LANGUAGE KindSignatures #-}
5+{-# LANGUAGE StandaloneDeriving #-}
6+{-# LANGUAGE TypeFamilies #-}
7+{-# LANGUAGE TypeOperators #-}
8+
9+module Obelisk.Request.HList (HList (..)) where
10+
11+import Data.Aeson (FromJSON, ToJSON, Value (Array), parseJSON, toJSON, withArray)
12+import Data.Vector (Vector)
13+import qualified Data.Vector as V
14+
15+data family HList (l::[*])
16+
17+data instance HList '[] = HNil
18+data instance HList (x ': xs) = x `HCons` HList xs
19+
20+infixr 2 `HCons`
21+
22+deriving instance Eq (HList '[])
23+deriving instance (Eq x, Eq (HList xs)) => Eq (HList (x ': xs))
24+
25+deriving instance Ord (HList '[])
26+deriving instance (Ord x, Ord (HList xs)) => Ord (HList (x ': xs))
27+
28+deriving instance Bounded (HList '[])
29+deriving instance (Bounded x, Bounded (HList xs)) => Bounded (HList (x ': xs))
30+
31+instance (FromJSON h, FromJSON (HList t)) => FromJSON (HList (h ': t)) where
32+ parseJSON = withArray "HList (a ': t)" $ \v -> do
33+ Just (aVal, v') <- return $ vectorView v
34+ a <- parseJSON aVal
35+ b <- parseJSON $ Array v'
36+ return $ HCons a b
37+
38+instance FromJSON (HList '[]) where
39+ parseJSON = withArray "HList '[]" $ \v -> do
40+ Nothing <- return $ vectorView v
41+ return HNil
42+
43+class HListToJSON l where
44+ hListToJSON :: HList l -> [Value]
45+
46+instance (ToJSON h, HListToJSON t) => HListToJSON (h ': t) where
47+ hListToJSON (HCons h t) = toJSON h : hListToJSON t
48+
49+instance HListToJSON '[] where
50+ hListToJSON HNil = []
51+
52+instance HListToJSON l => ToJSON (HList l) where
53+ toJSON = Array . V.fromList . hListToJSON
54+
55+vectorView :: Vector a -> Maybe (a, Vector a)
56+vectorView v =
57+ if V.length v > 0
58+ then Just (V.head v, V.tail v)
59+ else Nothing
lib/request/src/Obelisk/Request/TH.hsnew file+142
@@ -0,0 +1,142 @@
1+{-# LANGUAGE TemplateHaskell #-}
2+
3+module Obelisk.Request.TH
4+ ( makeJson
5+ , makeRequestForDataInstance
6+ , makeRequestForData
7+ ) where
8+
9+import Control.Monad (guard, replicateM)
10+import Data.Aeson.Types
11+import Data.Constraint (Dict (..))
12+import Data.List (isPrefixOf)
13+import Data.Semigroup ((<>))
14+import Language.Haskell.TH
15+
16+import Obelisk.Request
17+import Obelisk.Request.HList (HList (..))
18+
19+makeJson :: Name -> DecsQ
20+makeJson n = do
21+ x <- reify n
22+ let base = nameBase n
23+ toBeStripped = base <> "_"
24+ modifyConName cn = if length cons == 1 then cn else if toBeStripped `isPrefixOf` cn then drop (length toBeStripped) cn else error $ "makeRequest: expecting name beginning with " <> show toBeStripped <> ", got " <> show cn
25+ cons = case x of
26+ TyConI d -> decCons d
27+ _ -> error $ "cant do constructor:" ++ show x
28+ typeNames = map tvbName $ case x of
29+ TyConI d -> decTvbs d
30+ _ -> error $ "cant do typename:" ++ show x
31+ let wild = match wildP (normalB [|fail "invalid message"|]) []
32+ [d|
33+ instance ToJSON $(foldl appT (conT n) $ map varT typeNames) where
34+ toJSON r = $(caseE [|r|] $ map (conToJson modifyConName) cons)
35+ instance FromJSON $(foldl appT (conT n) $ map varT typeNames) where
36+ parseJSON v = do
37+ (tag', v') <- parseJSON v
38+ $(caseE [|tag' :: String|] $ map (conParseJson modifyConName id [|v'|]) cons ++ [wild])
39+ |]
40+
41+makeRequestForDataInstance :: Name -> Name -> DecsQ
42+makeRequestForDataInstance n n' = do
43+ x <- reify n
44+ let base = nameBase n
45+ toBeStripped = base <> "_"
46+ modifyConName cn = if length cons == 1 then cn else if toBeStripped `isPrefixOf` cn then drop (length toBeStripped) cn else error $ "makeRequest: expecting name beginning with " <> show toBeStripped <> ", got " <> show cn
47+ cons = case x of
48+ (FamilyI _ dataInstances) -> do
49+ (DataInstD _ _ (ConT m:_) _ xs _) <- dataInstances
50+ guard $ m == n'
51+ xs
52+ _ -> error $ "cant do family:" ++ show x
53+ let wild = match wildP (normalB [|fail "invalid message"|]) []
54+ [d|
55+ instance Request $(appT (conT n) (conT n')) where
56+ requestToJSON r = $(caseE [|r|] $ map (conToJson modifyConName) cons)
57+ requestParseJSON v = do
58+ (tag', v') <- parseJSON v
59+ $(caseE [|tag' :: String|] $ map (conParseJson modifyConName (\body -> [|SomeRequest <$> $body|]) [|v'|]) cons ++ [wild])
60+ requestResponseToJSON r = $(caseE [|r|] $ map (\c -> match (conP (conName c) $ replicate (conArity c) wildP) (normalB [|Dict|]) []) cons)
61+ requestResponseFromJSON r = $(caseE [|r|] $ map (\c -> match (conP (conName c) $ replicate (conArity c) wildP) (normalB [|Dict|]) []) cons)
62+ |]
63+
64+makeRequestForData :: Name -> DecsQ
65+makeRequestForData n = do
66+ x <- reify n
67+ let base = nameBase n
68+ toBeStripped = base <> "_"
69+ modifyConName cn = if length cons == 1 then cn else if toBeStripped `isPrefixOf` cn then drop (length toBeStripped) cn else error $ "makeRequest: expecting name beginning with " <> show toBeStripped <> ", got " <> show cn
70+ cons ::[Con]
71+ cons = case x of
72+ TyConI (DataD _ _ _ _ xs _) -> xs
73+ _ -> error $ "cant do data:" ++ show x
74+ let wild = match wildP (normalB [|fail "invalid message"|]) []
75+ [d|
76+ instance Request $(conT n) where
77+ requestToJSON r = $(caseE [|r|] $ map (conToJson modifyConName) cons)
78+ requestParseJSON v = do
79+ (tag', v') <- parseJSON v
80+ $(caseE [|tag' :: String|] $ map (conParseJson modifyConName (\body -> [|SomeRequest <$> $body|]) [|v'|]) cons ++ [wild])
81+ requestResponseToJSON r = $(caseE [|r|] $ map (\c -> match (conP (conName c) $ replicate (conArity c) wildP) (normalB [|Dict|]) []) cons)
82+ requestResponseFromJSON r = $(caseE [|r|] $ map (\c -> match (conP (conName c) $ replicate (conArity c) wildP) (normalB [|Dict|]) []) cons)
83+ |]
84+
85+conParseJson :: (String -> String) -> (ExpQ -> ExpQ) -> ExpQ -> Con -> MatchQ
86+conParseJson modifyName wrapBody e c = do
87+ let name = conName c
88+ varNames <- replicateM (conArity c) $ newName "f"
89+ let fields = map varE varNames
90+ tuple = foldr (\a b -> conP 'HCons [varP a, b]) (conP 'HNil []) varNames
91+ body = doE [ bindS tuple [|parseJSON $e|]
92+ , noBindS [|return $(appsE (conE name : fields))|]
93+ ]
94+ match (litP (StringL (modifyName $ nameBase name))) (normalB (wrapBody body)) []
95+
96+conToJson :: (String -> String) -> Con -> MatchQ
97+conToJson modifyName c = do
98+ let name = conName c
99+ base = nameBase name
100+ tag' = modifyName base
101+ varNames <- replicateM (conArity c) $ newName "f"
102+ let tuple = foldr (\a b -> appsE [conE 'HCons, varE a, b]) (conE 'HNil) varNames
103+ body = [|toJSON (tag' :: String, toJSON $tuple)|]
104+ match (conP name $ map varP varNames) (normalB body) []
105+
106+conArity :: Con -> Int
107+conArity c = case c of
108+ NormalC _ ts -> length ts
109+ RecC _ ts -> length ts
110+ InfixC _ _ _ -> 2
111+ ForallC _ _ c' -> conArity c'
112+ GadtC _ ts _ -> length ts
113+ RecGadtC _ ts _ -> length ts
114+
115+
116+decCons :: Dec -> [Con]
117+decCons d = case d of
118+ DataD _ _ _ _ cs _ -> cs
119+ NewtypeD _ _ _ _ c _ -> [c]
120+ _ -> error $ "not a data/newtype:" ++ show d
121+
122+-- | Extracts the name from a type variable binder.
123+tvbName :: TyVarBndr -> Name
124+tvbName (PlainTV name ) = name
125+tvbName (KindedTV name _) = name
126+
127+
128+decTvbs :: Dec -> [TyVarBndr]
129+decTvbs d = case d of
130+ DataD _ _ tvbs _ _ _ -> tvbs
131+ NewtypeD _ _ tvbs _ _ _ -> tvbs
132+ _ -> error $ "not a data/newtype:" ++ show d
133+
134+conName :: Con -> Name
135+conName c = case c of
136+ NormalC n _ -> n
137+ RecC n _ -> n
138+ InfixC _ n _ -> n
139+ ForallC _ _ c' -> conName c'
140+ GadtC [n] _ _ -> n
141+ RecGadtC [n] _ _ -> n
142+ _ -> error "conName: GADT constructors with multiple names not yet supported"
skeleton/.gitignore+1
@@ -5,3 +5,4 @@ result-ios
5result-exe
5result-exe
6.attr-cache
6.attr-cache
7ghcid-output.txt
7ghcid-output.txt
8+db
skeleton/backend/backend.cabal+4
@@ -11,7 +11,11 @@ library
11 , common
11 , common
12 , frontend
12 , frontend
13 , obelisk-backend
13 , obelisk-backend
14+ , obelisk-request
14 , obelisk-route
15 , obelisk-route
16+ , beam-core
17+ , beam-migrate
18+ , beam-postgres
15 exposed-modules:
19 exposed-modules:
16 Backend
20 Backend
17 ghc-options: -Wall
21 ghc-options: -Wall
skeleton/backend/src/Backend.hs+111−1
@@ -1,15 +1,125 @@
1{-# LANGUAGE EmptyCase #-}
1{-# LANGUAGE EmptyCase #-}
2+{-# LANGUAGE MultiParamTypeClasses #-}
3+{-# LANGUAGE AllowAmbiguousTypes #-}
4+{-# LANGUAGE TemplateHaskell #-}
5+{-# LANGUAGE GADTs #-}
6+{-# LANGUAGE TupleSections #-}
7+{-# LANGUAGE DeriveFunctor #-}
8+{-# LANGUAGE DeriveFoldable #-}
9+{-# LANGUAGE DeriveTraversable #-}
10+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
11+{-# LANGUAGE TypeApplications #-}
12+{-# LANGUAGE ScopedTypeVariables #-}
2{-# LANGUAGE FlexibleContexts #-}
13{-# LANGUAGE FlexibleContexts #-}
14+{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE LambdaCase #-}
15{-# LANGUAGE LambdaCase #-}
16+{-# LANGUAGE OverloadedStrings #-}
4{-# LANGUAGE RankNTypes #-}
17{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE TypeFamilies #-}
18{-# LANGUAGE TypeFamilies #-}
6module Backend where
19module Backend where
7
20
21+import Common.Api
22+import Backend.Schema
8import Common.Route
23import Common.Route
24+import Data.Text (Text)
25+import Data.Map.Monoidal (MonoidalMap (..))
26+import Data.Pool
9import Obelisk.Backend
27import Obelisk.Backend
28+import Obelisk.Route
29+import Obelisk.Api
30+import Obelisk.Db
31+import Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
32+import qualified Data.Map as Map
33+import Data.Sequence (Seq)
34+
35+import Database.Beam
36+import Database.Beam.Postgres
37+import Database.Beam.Postgres.Migrate
38+import Database.Beam.Migrate.Simple
39+import Database.Beam.Schema.Tables
40+import Database.PostgreSQL.Simple.Types
41+import Control.Monad.Reader
42+
43+import Data.Proxy
44+
45+import Data.Functor.Identity
46+import Data.Functor.Const
47+import Control.Lens
48+
49+import Debug.Trace
10
50
11backend :: Backend BackendRoute FrontendRoute
51backend :: Backend BackendRoute FrontendRoute
12backend = Backend
52backend = Backend
13− { _backend_run = \serve -> serve $ const $ return ()
53+ { _backend_run = \serve -> do
54+ withDbUri "db" $ \dbUri -> do
55+ withConnectionPool dbUri $ \pool -> do
56+ withResource pool $ \conn -> do
57+ runBeamPostgres conn $ autoMigrate migrationBackend checkedDb
58+ let onNotify :: forall a. VS Int Text a -> Transaction -> ReadDb (V Int Text a)
59+ onNotify (VS _) txn = do
60+ traceM $ show txn
61+ traceM $ show $ decodeTransaction txn
62+ tasks <- runSelectReturningList' $ select $ all_ $ _db_tasks db
63+ traceM $ show tasks
64+ pure $ V $ MonoidalMap mempty
65+ onSubscribe :: forall a. VS Int Text a -> ReadDb (V Int Text a)
66+ onSubscribe (VS _) = do
67+ tasks <- runSelectReturningList' $ select $ all_ $ _db_tasks db
68+ traceM $ show tasks
69+ pure $ V $ MonoidalMap mempty
70+ onRequest :: forall a. MyRequest a -> WriteDb a
71+ onRequest = \case
72+ MyRequest_Echo a -> do
73+ runInsert' $ insert (_db_tasks db) $ insertExpressions $ (:[]) $ Task
74+ { _task_id = default_
75+ , _task_description = val_ a
76+ , _task_completed = val_ False
77+ }
78+ pure a
79+ withApiHandler dbUri onRequest onNotify onSubscribe $ \handler -> do
80+ serve $ \case
81+ BackendRoute_Api :/ () -> handler
82+ _ -> return ()
14 , _backend_routeEncoder = backendRouteEncoder
83 , _backend_routeEncoder = backendRouteEncoder
15 }
84 }
85+
86+class MonadBeamRead m where
87+ runSelectReturningList' :: FromBackendRow Postgres a => SqlSelect Postgres a -> m [a]
88+ runSelectReturningOne' :: FromBackendRow Postgres a => SqlSelect Postgres a -> m (Maybe a)
89+
90+unsafeRunPgReadDb :: Pg a -> ReadDb a
91+unsafeRunPgReadDb a = ReadDb $ do
92+ conn <- ask
93+ lift $ runBeamPostgres conn a
94+
95+unsafeRunPgWriteDb :: Pg a -> WriteDb a
96+unsafeRunPgWriteDb a = WriteDb $ do
97+ conn <- ask
98+ lift $ runBeamPostgres conn a
99+
100+instance MonadBeamRead WriteDb where
101+ runSelectReturningList' = unsafeRunPgWriteDb . runSelectReturningList
102+ runSelectReturningOne' = unsafeRunPgWriteDb . runSelectReturningOne
103+
104+instance MonadBeamRead ReadDb where
105+ runSelectReturningList' = unsafeRunPgReadDb . runSelectReturningList
106+ runSelectReturningOne' = unsafeRunPgReadDb . runSelectReturningOne
107+
108+class MonadBeamWrite m where
109+ runInsert' :: SqlInsert Postgres table -> m ()
110+ runUpdate' :: SqlUpdate Postgres table -> m ()
111+ runDelete' :: SqlDelete Postgres table -> m ()
112+
113+instance MonadBeamWrite WriteDb where
114+ runInsert' = unsafeRunPgWriteDb . runInsert
115+ runUpdate' = unsafeRunPgWriteDb . runUpdate
116+ runDelete' = unsafeRunPgWriteDb . runDelete
117+
118+decodeTransaction :: Transaction -> Db (Const (Seq Change))
119+decodeTransaction (_, tableChanges) = runIdentity $ zipTables (Proxy :: Proxy Postgres) f db db
120+ where
121+ f :: IsDatabaseEntity Postgres entityType => DatabaseEntity Postgres Db entityType -> DatabaseEntity Postgres Db entityType -> Identity (Const (Seq Change) entityType)
122+ f (DatabaseEntity descriptor) _ = pure $ Const $
123+ let n = view dbEntityName descriptor
124+ s = Just "public" -- view dbEntitySchema descriptor --TODO: Beam has 'Nothing' while the replication connection reports 'Just "public"'
125+ in trace (show (s, n)) $ Map.findWithDefault mempty (QualifiedIdentifier s n) tableChanges
skeleton/backend/src/Backend/Schema.hsnew file+50
@@ -0,0 +1,50 @@
1+{-# LANGUAGE StandaloneDeriving #-}
2+{-# LANGUAGE UndecidableInstances #-}
3+{-# LANGUAGE ScopedTypeVariables #-}
4+{-# LANGUAGE MultiParamTypeClasses #-}
5+{-# LANGUAGE TypeFamilies #-}
6+{-# LANGUAGE FlexibleInstances #-}
7+{-# LANGUAGE DeriveGeneric #-}
8+{-# LANGUAGE TypeSynonymInstances #-}
9+module Backend.Schema where
10+
11+import Data.Int
12+import Data.Text (Text)
13+import Database.Beam
14+import Database.Beam.Migrate.Simple
15+import Database.Beam.Migrate
16+import Database.Beam.Postgres
17+
18+data TaskT f = Task
19+ { _task_id :: Columnar f Int64
20+ , _task_description :: Columnar f Text
21+ , _task_completed :: Columnar f Bool
22+ }
23+ deriving (Generic)
24+type Task = TaskT Identity
25+
26+deriving instance Show Task
27+deriving instance Eq Task
28+
29+instance Beamable TaskT
30+
31+instance Table TaskT where
32+ data PrimaryKey TaskT f = TaskId (Columnar f Int64) deriving (Generic)
33+ primaryKey = TaskId . _task_id
34+
35+instance Beamable (PrimaryKey TaskT)
36+
37+data Db f = Db
38+ { _db_tasks :: f (TableEntity TaskT)
39+ }
40+ deriving (Generic)
41+
42+deriving instance Show (f (TableEntity TaskT)) => Show (Db f)
43+
44+instance Database Postgres Db
45+
46+checkedDb :: CheckedDatabaseSettings Postgres Db
47+checkedDb = defaultMigratableDbSettings
48+
49+db :: DatabaseSettings Postgres Db
50+db = unCheckDatabase checkedDb
skeleton/common/src/Common/Api.hs+43−2
@@ -1,4 +1,45 @@
1+{-# LANGUAGE EmptyCase #-}
2+{-# LANGUAGE TemplateHaskell #-}
3+{-# LANGUAGE GADTs #-}
4+{-# LANGUAGE TupleSections #-}
5+{-# LANGUAGE DeriveFunctor #-}
6+{-# LANGUAGE DeriveFoldable #-}
7+{-# LANGUAGE DeriveTraversable #-}
8+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
9+{-# LANGUAGE TypeApplications #-}
10+{-# LANGUAGE ScopedTypeVariables #-}
11+{-# LANGUAGE FlexibleContexts #-}
12+{-# LANGUAGE FlexibleInstances #-}
13+{-# LANGUAGE LambdaCase #-}
14+{-# LANGUAGE OverloadedStrings #-}
15+{-# LANGUAGE RankNTypes #-}
16+{-# LANGUAGE TypeFamilies #-}
1module Common.Api where
17module Common.Api where
2
18
3−commonStuff :: String
19+import Data.Aeson
4−commonStuff = "Here is a string defined in code common to the frontend and backend."
20+import Data.AppendMap ()
21+import Data.Map.Monoidal (MonoidalMap (..))
22+import qualified Data.Map.Monoidal as Map
23+import Data.Semigroup
24+import Data.Text (Text)
25+import Obelisk.Request.TH
26+import Reflex.Class
27+import Reflex.Query.Class
28+
29+newtype VS k v a = VS (MonoidalMap k a)
30+ deriving (Semigroup, Monoid, Functor, Foldable, Traversable, Group, ToJSON, FromJSON, Eq)
31+
32+newtype V k v a = V (MonoidalMap k (a, First (Maybe v)))
33+ deriving (Semigroup, Monoid, Functor, Foldable, Traversable, ToJSON, FromJSON, Eq)
34+
35+instance FunctorMaybe (V k v) where
36+ fmapMaybe f (V m) = V $ Map.mapMaybe (\(a, v) -> (,v) <$> f a) m
37+
38+instance (Ord k, Semigroup a) => Query (VS k v a) where
39+ type QueryResult (VS k v a) = V k v a
40+ crop _ = id
41+
42+data MyRequest a where
43+ MyRequest_Echo :: Text -> MyRequest Text
44+
45+makeRequestForData ''MyRequest
skeleton/common/src/Common/Route.hs+2
@@ -29,6 +29,7 @@ data BackendRoute :: * -> * where
29 BackendRoute_Missing :: BackendRoute ()
29 BackendRoute_Missing :: BackendRoute ()
30 -- You can define any routes that will be handled specially by the backend here.
30 -- You can define any routes that will be handled specially by the backend here.
31 -- i.e. These do not serve the frontend, but do something different, such as serving static files.
31 -- i.e. These do not serve the frontend, but do something different, such as serving static files.
32+ BackendRoute_Api :: BackendRoute ()
32
33
33data FrontendRoute :: * -> * where
34data FrontendRoute :: * -> * where
34 FrontendRoute_Main :: FrontendRoute ()
35 FrontendRoute_Main :: FrontendRoute ()
@@ -40,6 +41,7 @@ backendRouteEncoder = handleEncoder (const (InL BackendRoute_Missing :/ ())) $
40 pathComponentEncoder $ \case
41 pathComponentEncoder $ \case
41 InL backendRoute -> case backendRoute of
42 InL backendRoute -> case backendRoute of
42 BackendRoute_Missing -> PathSegment "missing" $ unitEncoder mempty
43 BackendRoute_Missing -> PathSegment "missing" $ unitEncoder mempty
44+ BackendRoute_Api -> PathSegment "api" $ unitEncoder mempty
43 InR obeliskRoute -> obeliskRouteSegment obeliskRoute $ \case
45 InR obeliskRoute -> obeliskRouteSegment obeliskRoute $ \case
44 -- The encoder given to PathEnd determines how to parse query parameters,
46 -- The encoder given to PathEnd determines how to parse query parameters,
45 -- in this example, we have none, so we insist on it.
47 -- in this example, we have none, so we insist on it.
skeleton/frontend/src/Frontend.hs+8−7
@@ -1,23 +1,24 @@
1{-# LANGUAGE DataKinds #-}
1{-# LANGUAGE DataKinds #-}
2+{-# LANGUAGE LambdaCase #-}
3+{-# LANGUAGE TypeFamilies #-}
4+{-# LANGUAGE FlexibleContexts #-}
5+{-# LANGUAGE ScopedTypeVariables #-}
6+{-# LANGUAGE RankNTypes #-}
7+{-# LANGUAGE RecursiveDo #-}
2{-# LANGUAGE OverloadedStrings #-}
8{-# LANGUAGE OverloadedStrings #-}
3{-# LANGUAGE TypeApplications #-}
9{-# LANGUAGE TypeApplications #-}
4module Frontend where
10module Frontend where
5
11
6−import qualified Data.Text as T
12+import Common.Route
7import Obelisk.Frontend
13import Obelisk.Frontend
14+import Obelisk.Generated.Static
8import Obelisk.Route
15import Obelisk.Route
9import Reflex.Dom.Core
16import Reflex.Dom.Core
10
17
11−import Common.Api
12−import Common.Route
13−import Obelisk.Generated.Static
14−
15−
16frontend :: Frontend (R FrontendRoute)
18frontend :: Frontend (R FrontendRoute)
17frontend = Frontend
19frontend = Frontend
18 { _frontend_head = el "title" $ text "Obelisk Minimal Example"
20 { _frontend_head = el "title" $ text "Obelisk Minimal Example"
19 , _frontend_body = do
21 , _frontend_body = do
20 text "Welcome to Obelisk!"
22 text "Welcome to Obelisk!"
21− el "p" $ text $ T.pack commonStuff
22 elAttr "img" ("src" =: static @"obelisk.jpg") blank
23 elAttr "img" ("src" =: static @"obelisk.jpg") blank
23 }
24 }