git.iko.soy/obelisk
on master help sign in
Decode transactions with Beam
cab55e2Ryan Trinkle
default.nix+4
compare with master
unifiedsplit
@@ -20,6 +20,8 @@ let
20
20
21 gargoyle-src = hackGet ./dep/gargoyle;
21 gargoyle-src = hackGet ./dep/gargoyle;
22
22
23+ postgresql-simple-src = hackGet ./dep/postgresql-simple;
24+
23 getReflexPlatform = getReflexPlatform' __useLegacyCompilers;
25 getReflexPlatform = getReflexPlatform' __useLegacyCompilers;
24 getReflexPlatform' = __useLegacyCompilers: sys: import ./dep/reflex-platform {
26 getReflexPlatform' = __useLegacyCompilers: sys: import ./dep/reflex-platform {
25 inherit iosSdkVersion __useLegacyCompilers;
27 inherit iosSdkVersion __useLegacyCompilers;
@@ -73,6 +75,8 @@ let
73 beam-postgres = dontCheck (self.callCabal2nix "beam-postgres" (beam-src + /beam-postgres) {});
75 beam-postgres = dontCheck (self.callCabal2nix "beam-postgres" (beam-src + /beam-postgres) {});
74 beam-migrate = self.callCabal2nix "beam-migrate" (beam-src + /beam-migrate) {};
76 beam-migrate = self.callCabal2nix "beam-migrate" (beam-src + /beam-migrate) {};
75
77
78+ postgresql-simple = dontCheck (self.callCabal2nix "postgresql-simple" postgresql-simple-src {});
79+
76 websockets = self.callCabal2nix "websockets" (pkgs.fetchFromGitHub {
80 websockets = self.callCabal2nix "websockets" (pkgs.fetchFromGitHub {
77 owner = "obsidiansystems";
81 owner = "obsidiansystems";
78 repo = "websockets";
82 repo = "websockets";
dep/beam/github.json+3−3
@@ -1,7 +1,7 @@
1{
1{
2 "owner": "obsidiansystems",
2 "owner": "obsidiansystems",
3 "repo": "beam",
3 "repo": "beam",
4− "branch": "int-and-serial-size",
4+ "branch": "obelisk",
5− "rev": "87caf239ea198bad03acb73e65f9a43566ee4373",
5+ "rev": "d060ed41016aa62f4a571e69430f64ef660db894",
6− "sha256": "1lnzwds131cxix0dwkql980w9s2jdxfp2b9q7x8d4wh5i5jfbh6g"
6+ "sha256": "0ag19lc2izay5qg3j9mv44izsccglh89191gmh9vfx2kqv20h945"
7}
7}
dep/postgresql-simple/default.nixnew file+7
@@ -0,0 +1,7 @@
1+# DO NOT HAND-EDIT THIS FILE
2+import ((import <nixpkgs> {}).fetchFromGitHub (
3+ let json = builtins.fromJSON (builtins.readFile ./github.json);
4+ in { inherit (json) owner repo rev sha256;
5+ private = json.private or false;
6+ }
7+))
dep/postgresql-simple/github.jsonnew file+7
@@ -0,0 +1,7 @@
1+{
2+ "owner": "obsidiansystems",
3+ "repo": "postgresql-simple",
4+ "branch": "synthesizable-fields",
5+ "rev": "dc162dc21f586ba4c6e5c44f92b76aac653a85db",
6+ "sha256": "1mpx6w94z9b3vv1981v7csj2y4p97irv3p7cvv1lyz9c595kkisp"
7+}
lib/backend/src/Obelisk/Postgres/LogicalDecoding/Plugins/TestDecoding.hs+2
@@ -4,7 +4,9 @@ module Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
4 ( line
4 ( line
5 , Transaction
5 , Transaction
6 , Change (..)
6 , Change (..)
7+ , Literal (..)
7 , LineError (..)
8 , LineError (..)
9+ , TypeName (..)
8 , linesToTransactions
10 , linesToTransactions
9 ) where
11 ) where
10
12
skeleton/backend/src/Backend.hs+161−9
@@ -1,4 +1,7 @@
1{-# LANGUAGE EmptyCase #-}
1{-# LANGUAGE EmptyCase #-}
2+{-# LANGUAGE StandaloneDeriving #-}
3+{-# LANGUAGE TypeOperators #-}
4+{-# LANGUAGE UndecidableInstances #-}
2{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE MultiParamTypeClasses #-}
3{-# LANGUAGE AllowAmbiguousTypes #-}
6{-# LANGUAGE AllowAmbiguousTypes #-}
4{-# LANGUAGE TemplateHaskell #-}
7{-# LANGUAGE TemplateHaskell #-}
@@ -16,13 +19,16 @@
16{-# LANGUAGE OverloadedStrings #-}
19{-# LANGUAGE OverloadedStrings #-}
17{-# LANGUAGE RankNTypes #-}
20{-# LANGUAGE RankNTypes #-}
18{-# LANGUAGE TypeFamilies #-}
21{-# LANGUAGE TypeFamilies #-}
22+
23+{-# OPTIONS_GHC -fno-warn-orphans #-} --TODO: Eliminate this
19module Backend where
24module Backend where
20
25
26+import Control.Exception
27+import Control.Concurrent
21import Common.Api
28import Common.Api
22import Common.Schema
29import Common.Schema
23import Backend.Schema
30import Backend.Schema
24import Common.Route
31import Common.Route
25−import Data.Map.Monoidal (MonoidalMap (..))
26import qualified Data.Map.Monoidal as Map
32import qualified Data.Map.Monoidal as Map
27import Data.Pool
33import Data.Pool
28import Data.Semigroup
34import Data.Semigroup
@@ -33,8 +39,19 @@ import Obelisk.Db
33import Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
39import Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
34import qualified Data.Map as OldMap
40import qualified Data.Map as OldMap
35import Data.Sequence (Seq)
41import Data.Sequence (Seq)
42+import Data.These
43+import Data.Text.Encoding
44+import Data.HashMap.Strict (HashMap)
45+import qualified Data.HashMap.Strict as HashMap
36
46
47+import Database.PostgreSQL.Simple.FromField (FromField (..), runConversion)
48+import Database.PostgreSQL.Simple.Ok
49+import Database.PostgreSQL.Simple.Internal
50+import qualified Database.PostgreSQL.Simple.TypeInfo.Static as TI
51+import qualified Database.PostgreSQL.LibPQ as PQ
37import Database.Beam
52import Database.Beam
53+import Database.Beam.Backend.Types
54+import Database.Beam.Backend.SQL.Types
38import Database.Beam.Postgres
55import Database.Beam.Postgres
39import Database.Beam.Postgres.Full (PgInsertReturning, returning, runPgInsertReturningList)
56import Database.Beam.Postgres.Full (PgInsertReturning, returning, runPgInsertReturningList)
40import Database.Beam.Postgres.Migrate
57import Database.Beam.Postgres.Migrate
@@ -45,9 +62,10 @@ import Control.Monad.Reader
45
62
46import Data.Proxy
63import Data.Proxy
47
64
48−import Data.Functor.Identity
49−import Data.Functor.Const
50import Control.Lens
65import Control.Lens
66+import qualified GHC.Generics as Generic
67+import GHC.Generics hiding (R, C)
68+import GHC.Types
51
69
52import Debug.Trace
70import Debug.Trace
53
71
@@ -57,9 +75,18 @@ backend = Backend
57 withDbUri "db" $ \dbUri -> do
75 withDbUri "db" $ \dbUri -> do
58 withConnectionPool dbUri $ \pool -> do
76 withConnectionPool dbUri $ \pool -> do
59 withResource pool $ \conn -> do
77 withResource pool $ \conn -> do
60− runBeamPostgres conn $ autoMigrate migrationBackend checkedDb
78+ try (runBeamPostgres conn $ autoMigrate migrationBackend checkedDb) >>= \case
79+ Left (_ :: SomeException) -> do
80+ putStrLn "Required migration is unsafe; refusing to start"
81+ forever $ threadDelay maxBound
82+ Right _ -> pure ()
61 let onNotify :: forall a. VS (Id Task) Task a -> Transaction -> ReadDb (V (Id Task) Task a)
83 let onNotify :: forall a. VS (Id Task) Task a -> Transaction -> ReadDb (V (Id Task) Task a)
62 onNotify (VS requested) txn = do
84 onNotify (VS requested) txn = do
85+ traceM "Hello"
86+ decoded <- ReadDb $ do
87+ conn <- ask
88+ lift $ decodeTransaction conn txn
89+ traceM $ show $ _db_tasks decoded
63 tasks <- runSelectReturningList' $ select $ all_ $ _db_tasks db
90 tasks <- runSelectReturningList' $ select $ all_ $ _db_tasks db
64 pure $ V $ Map.intersectionWith (,) requested $ Map.fromList $ fmap (\x -> (TaskId $ _task_id x, First $ Just x)) tasks
91 pure $ V $ Map.intersectionWith (,) requested $ Map.fromList $ fmap (\x -> (TaskId $ _task_id x, First $ Just x)) tasks
65 onSubscribe :: forall a. VS (Id Task) Task a -> ReadDb (V (Id Task) Task a)
92 onSubscribe :: forall a. VS (Id Task) Task a -> ReadDb (V (Id Task) Task a)
@@ -116,11 +143,136 @@ instance MonadBeamWrite WriteDb where
116 runUpdate' = unsafeRunPgWriteDb . runUpdate
143 runUpdate' = unsafeRunPgWriteDb . runUpdate
117 runDelete' = unsafeRunPgWriteDb . runDelete
144 runDelete' = unsafeRunPgWriteDb . runDelete
118
145
119−decodeTransaction :: Transaction -> Db (Const (Seq Change))
146+class GPatchDatabase x where
147+ gDatabasePatchDecoder :: x
148+
149+instance GPatchDatabase (x p) => GPatchDatabase (D1 f x p) where
150+ gDatabasePatchDecoder = M1 gDatabasePatchDecoder
151+
152+instance GPatchDatabase (x p) => GPatchDatabase (C1 f x p) where
153+ gDatabasePatchDecoder = M1 gDatabasePatchDecoder
154+
155+instance (GPatchDatabase (x p), GPatchDatabase (y p)) => GPatchDatabase ((x :*: y) p) where
156+ gDatabasePatchDecoder = gDatabasePatchDecoder :*: gDatabasePatchDecoder
157+
158+instance PatchEntity be a => GPatchDatabase (S1 f (K1 Generic.R (EntityPatchDecoder be a)) p) where
159+ gDatabasePatchDecoder = M1 (K1 patchEntity)
160+
161+class GPatchFields x where
162+ gPatchFields :: x
163+
164+instance GPatchFields (x p) => GPatchFields (D1 f x p) where
165+ gPatchFields = M1 gPatchFields
166+
167+instance GPatchFields (x p) => GPatchFields (C1 f x p) where
168+ gPatchFields = M1 gPatchFields
169+
170+instance (GPatchFields (x p), GPatchFields (y p)) => GPatchFields ((x :*: y) p) where
171+ gPatchFields = gPatchFields :*: gPatchFields
172+
173+instance BackendFromField be a => GPatchFields (S1 f (K1 Generic.R (FieldDecoder be a)) p) where
174+ gPatchFields = M1 (K1 FieldDecoder)
175+
176+class PatchEntity be e where
177+ patchEntity :: EntityPatchDecoder be e
178+
179+from' :: Generic x => x -> Rep x ()
180+from' = Generic.from
181+
182+to' :: Generic x => Rep x () -> x
183+to' = Generic.to
184+
185+instance ( Generic (tbl (FieldDecoder be))
186+ , GPatchFields (Rep (tbl (FieldDecoder be)) ())
187+ , Table tbl
188+ ) => PatchEntity be (TableEntity tbl) where
189+ patchEntity = EntityPatchDecoder_Table $ to' gPatchFields
190+
191+defaultDatabasePatchDecoder
192+ :: forall be db
193+ . ( Generic (db (EntityPatchDecoder be))
194+ , GPatchDatabase (Rep (db (EntityPatchDecoder be)) ())
195+ )
196+ => db (EntityPatchDecoder be)
197+defaultDatabasePatchDecoder = to' gDatabasePatchDecoder
198+
199+class PatchableEntity be e where
200+ patchDecoder :: EntityPatchDecoder be e
201+
202+data EntityPatchDecoder be e where
203+ EntityPatchDecoder_NotDecodable :: EntityPatchDecoder be e
204+ EntityPatchDecoder_Table :: Table tbl => tbl (FieldDecoder be) -> EntityPatchDecoder be (TableEntity tbl)
205+
206+data FieldDecoder be a where
207+ FieldDecoder :: BackendFromField be a => FieldDecoder be a
208+
209+data EntityPatch e where
210+ EntityPatch_NotPatchable :: EntityPatch e
211+ EntityPatch_Table :: Seq (These (PrimaryKey tbl Maybe) (tbl Maybe)) -> EntityPatch (TableEntity tbl) --TODO: Add old keys / deletes
212+
213+deriving instance (Show (PrimaryKey tbl Maybe), Show (tbl Maybe)) => Show (EntityPatch (TableEntity tbl))
214+
215+deriving instance FromField a => FromField (SqlSerial a)
216+
217+decodeRow
218+ :: forall tbl tbl'
219+ . Beamable tbl
220+ => Connection
221+ -> tbl (TableField tbl')
222+ -> tbl (FieldDecoder Postgres)
223+ -> HashMap Identifier (TypeName, Maybe Literal)
224+ -> IO (tbl Maybe)
225+decodeRow conn settings decoders r = zipBeamFieldsM d settings decoders
226+ where d :: forall a. Columnar' (TableField tbl') a -> Columnar' (FieldDecoder Postgres) a -> IO (Columnar' Maybe a)
227+ d (Columnar' fieldSettings) (Columnar' FieldDecoder) = do
228+ let n = _fieldName fieldSettings
229+ result <- case HashMap.lookup (Identifier n) r of
230+ Just (TypeName typeName, v) -> do
231+ let f = Field
232+ { typeOid = case typeName of
233+ "text" -> TI.typoid TI.text
234+ "int8" -> TI.typoid TI.int8
235+ "bigint" -> TI.typoid TI.int8
236+ "character varying" -> TI.typoid TI.text
237+ "boolean" -> TI.typoid TI.bool
238+ _ -> error $ "Unrecognized type: " <> show typeName --TODO: Retrieve types properly
239+ , format = PQ.Text --TODO: Is this right?
240+ , tableOid = Nothing
241+ , name = Just $ encodeUtf8 n
242+ }
243+ case v of
244+ Just (Literal_Present l) -> Just <$> runConversion (fromField f $ Just l) conn
245+ Nothing -> Just <$> runConversion (fromField f Nothing) conn
246+ Just Literal_UnchangedToastDatum -> pure Nothing
247+ Nothing -> pure Nothing
248+ case result of
249+ Just (Ok a) -> pure $ Columnar' $ Just a
250+ Just (Errors e) -> fail $ show e
251+ Nothing -> pure $ Columnar' Nothing
252+
253+decodeChange
254+ :: forall tbl
255+ . Table tbl
256+ => Connection
257+ -> tbl (TableField tbl)
258+ -> tbl (FieldDecoder Postgres)
259+ -> Change
260+ -> IO (These (PrimaryKey tbl Maybe) (tbl Maybe))
261+decodeChange conn settings decoders = \case
262+ Change_Insert r -> That <$> decodeRow conn settings decoders r
263+ Change_Update k r -> These <$> decodeRow conn (primaryKey settings) (primaryKey decoders) (fmap (fmap Just) k) <*> decodeRow conn settings decoders r
264+ Change_Delete k -> This <$> decodeRow conn (primaryKey settings) (primaryKey decoders) (fmap (fmap Just) k)
265+
266+decodeTransaction :: Connection -> Transaction -> IO (Db EntityPatch)
120−decodeTransaction (_, tableChanges) = runIdentity $ zipTables (Proxy :: Proxy Postgres) f db db
267+decodeTransaction conn (_, tableChanges) = zipTables (Proxy :: Proxy Postgres) f db (defaultDatabasePatchDecoder @Postgres @Db)
121 where
268 where
122− f :: IsDatabaseEntity Postgres entityType => DatabaseEntity Postgres Db entityType -> DatabaseEntity Postgres Db entityType -> Identity (Const (Seq Change) entityType)
269+ f :: forall entityType. IsDatabaseEntity Postgres entityType => DatabaseEntity Postgres Db entityType -> EntityPatchDecoder Postgres entityType -> IO (EntityPatch entityType)
123− f (DatabaseEntity descriptor) _ = pure $ Const $
270+ f (DatabaseEntity descriptor) decoder =
124 let n = view dbEntityName descriptor
271 let n = view dbEntityName descriptor
125 s = Just "public" -- view dbEntitySchema descriptor --TODO: Beam has 'Nothing' while the replication connection reports 'Just "public"'
272 s = Just "public" -- view dbEntitySchema descriptor --TODO: Beam has 'Nothing' while the replication connection reports 'Just "public"'
126− in trace (show (s, n)) $ OldMap.findWithDefault mempty (QualifiedIdentifier s n) tableChanges
273+ vals = OldMap.findWithDefault mempty (QualifiedIdentifier s n) tableChanges
274+ x :: IO (EntityPatch entityType)
275+ x = case decoder of
276+ EntityPatchDecoder_NotDecodable -> pure EntityPatch_NotPatchable
277+ EntityPatchDecoder_Table decoders -> EntityPatch_Table <$> traverse (decodeChange conn (dbTableSettings descriptor) decoders) vals
278+ in x
skeleton/backend/src/Backend/Schema.hs+17−8
@@ -1,4 +1,5 @@
1{-# LANGUAGE StandaloneDeriving #-}
1{-# LANGUAGE StandaloneDeriving #-}
2+{-# LANGUAGE TypeApplications #-}
2{-# LANGUAGE UndecidableInstances #-}
3{-# LANGUAGE UndecidableInstances #-}
3{-# LANGUAGE ScopedTypeVariables #-}
4{-# LANGUAGE ScopedTypeVariables #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE MultiParamTypeClasses #-}
@@ -6,26 +7,34 @@
6{-# LANGUAGE FlexibleInstances #-}
7{-# LANGUAGE FlexibleInstances #-}
7{-# LANGUAGE DeriveGeneric #-}
8{-# LANGUAGE DeriveGeneric #-}
8{-# LANGUAGE TypeSynonymInstances #-}
9{-# LANGUAGE TypeSynonymInstances #-}
10+
11+{-# OPTIONS_GHC -fno-warn-orphans #-} --TODO: Eliminate this
9module Backend.Schema where
12module Backend.Schema where
10
13
11import Common.Schema
14import Common.Schema
12
15
16+import Control.Lens
17+import Data.List.NonEmpty (NonEmpty (..))
18+import Data.Proxy
19+import qualified Data.Text as T
13import Database.Beam
20import Database.Beam
21+import Database.Beam.Schema.Tables
14import Database.Beam.Migrate.Simple
22import Database.Beam.Migrate.Simple
15import Database.Beam.Migrate
23import Database.Beam.Migrate
16import Database.Beam.Postgres
24import Database.Beam.Postgres
17
25
18−data Db f = Db
19− { _db_tasks :: f (TableEntity TaskT)
20− }
21− deriving (Generic)
22−
23−deriving instance Show (f (TableEntity TaskT)) => Show (Db f)
24−
25instance Database Postgres Db
26instance Database Postgres Db
26
27
28+replaceAllFieldNames :: forall be db. Database be db => CheckedDatabaseSettings be db -> CheckedDatabaseSettings be db
29+replaceAllFieldNames d = runIdentity $ zipTables (Proxy @be) (\_ -> pure . f) d d
30+ where f :: IsDatabaseEntity be e => CheckedDatabaseEntity be db e -> CheckedDatabaseEntity be db e
31+ f (CheckedDatabaseEntity descriptor preds) =
32+ let renameEntity = unChecked . dbEntityName %~ T.drop 1 . T.dropWhile (/= '_')
33+ renameFields = unChecked %~ withFieldRenamer (renamingFields $ \(x :| _) -> T.drop 1 $ T.dropWhile (/= '_') $ T.drop 1 x)
34+ in CheckedDatabaseEntity (renameFields $ renameEntity descriptor) preds
35+
27checkedDb :: CheckedDatabaseSettings Postgres Db
36checkedDb :: CheckedDatabaseSettings Postgres Db
28−checkedDb = defaultMigratableDbSettings
37+checkedDb = replaceAllFieldNames defaultMigratableDbSettings
29
38
30db :: DatabaseSettings Postgres Db
39db :: DatabaseSettings Postgres Db
31db = unCheckDatabase checkedDb
40db = unCheckDatabase checkedDb
skeleton/common/src/Common/Schema.hs+13−1
@@ -7,6 +7,8 @@
7{-# LANGUAGE FlexibleInstances #-}
7{-# LANGUAGE FlexibleInstances #-}
8{-# LANGUAGE DeriveGeneric #-}
8{-# LANGUAGE DeriveGeneric #-}
9{-# LANGUAGE TypeSynonymInstances #-}
9{-# LANGUAGE TypeSynonymInstances #-}
10+
11+{-# OPTIONS_GHC -fno-warn-orphans #-} --TODO: Eliminate this
10module Common.Schema where
12module Common.Schema where
11
13
12import Data.Aeson
14import Data.Aeson
@@ -23,7 +25,8 @@ data TaskT f = Task
23 deriving (Generic)
25 deriving (Generic)
24type Task = TaskT Identity
26type Task = TaskT Identity
25
27
26−deriving instance Show Task
28+deriving instance (Show (Columnar f (SqlSerial Int64)), Show (Columnar f Text), Show (Columnar f Bool)) => Show (TaskT f)
29+
27deriving instance Eq Task
30deriving instance Eq Task
28
31
29instance Beamable TaskT
32instance Beamable TaskT
@@ -33,6 +36,8 @@ instance Table TaskT where
33 deriving (Generic)
36 deriving (Generic)
34 primaryKey = TaskId . _task_id
37 primaryKey = TaskId . _task_id
35
38
39+deriving instance Show (Columnar f (SqlSerial Int64)) => Show (PrimaryKey TaskT f)
40+
36deriving instance ToJSONKey a => ToJSONKey (SqlSerial a)
41deriving instance ToJSONKey a => ToJSONKey (SqlSerial a)
37deriving instance FromJSONKey a => FromJSONKey (SqlSerial a)
42deriving instance FromJSONKey a => FromJSONKey (SqlSerial a)
38
43
@@ -46,3 +51,10 @@ instance ToJSON (TaskT Identity)
46instance FromJSON (TaskT Identity)
51instance FromJSON (TaskT Identity)
47
52
48instance Beamable (PrimaryKey TaskT)
53instance Beamable (PrimaryKey TaskT)
54+
55+data Db f = Db
56+ { _db_tasks :: f (TableEntity TaskT)
57+ }
58+ deriving (Generic)
59+
60+deriving instance Show (f (TableEntity TaskT)) => Show (Db f)