@@ -1,4 +1,7 @@
11{-# LANGUAGE EmptyCase #-}
2+{-# LANGUAGE StandaloneDeriving #-}
3+{-# LANGUAGE TypeOperators #-}
4+{-# LANGUAGE UndecidableInstances #-}
25{-# LANGUAGE MultiParamTypeClasses #-}
36{-# LANGUAGE AllowAmbiguousTypes #-}
47{-# LANGUAGE TemplateHaskell #-}
@@ -16,13 +19,16 @@
1619{-# LANGUAGE OverloadedStrings #-}
1720{-# LANGUAGE RankNTypes #-}
1821{-# LANGUAGE TypeFamilies #-}
22+
23+{-# OPTIONS_GHC -fno-warn-orphans #-} --TODO: Eliminate this
1924module Backend where
2025
26+import Control.Exception
27+import Control.Concurrent
2128import Common.Api
2229import Common.Schema
2330import Backend.Schema
2431import Common.Route
25−import Data.Map.Monoidal (MonoidalMap (..))
2632import qualified Data.Map.Monoidal as Map
2733import Data.Pool
2834import Data.Semigroup
@@ -33,8 +39,19 @@ import Obelisk.Db
3339import Obelisk.Postgres.LogicalDecoding.Plugins.TestDecoding
3440import qualified Data.Map as OldMap
3541import 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
3646
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
3752import Database.Beam
53+import Database.Beam.Backend.Types
54+import Database.Beam.Backend.SQL.Types
3855import Database.Beam.Postgres
3956import Database.Beam.Postgres.Full (PgInsertReturning, returning, runPgInsertReturningList)
4057import Database.Beam.Postgres.Migrate
@@ -45,9 +62,10 @@ import Control.Monad.Reader
4562
4663import Data.Proxy
4764
48−import Data.Functor.Identity
49−import Data.Functor.Const
5065import Control.Lens
66+import qualified GHC.Generics as Generic
67+import GHC.Generics hiding (R, C)
68+import GHC.Types
5169
5270import Debug.Trace
5371
@@ -57,9 +75,18 @@ backend = Backend
5775 withDbUri "db" $ \dbUri -> do
5876 withConnectionPool dbUri $ \pool -> do
5977 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 ()
6183 let onNotify :: forall a. VS (Id Task) Task a -> Transaction -> ReadDb (V (Id Task) Task a)
6284 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
6390 tasks <- runSelectReturningList' $ select $ all_ $ _db_tasks db
6491 pure $ V $ Map.intersectionWith (,) requested $ Map.fromList $ fmap (\x -> (TaskId $ _task_id x, First $ Just x)) tasks
6592 onSubscribe :: forall a. VS (Id Task) Task a -> ReadDb (V (Id Task) Task a)
@@ -116,11 +143,136 @@ instance MonadBeamWrite WriteDb where
116143 runUpdate' = unsafeRunPgWriteDb . runUpdate
117144 runDelete' = unsafeRunPgWriteDb . runDelete
118145
119−decodeTransaction :: Transaction -> Db (Const (Seq Change))
120−decodeTransaction (_, tableChanges) = runIdentity $ zipTables (Proxy :: Proxy Postgres) f db db
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)
267+decodeTransaction conn (_, tableChanges) = zipTables (Proxy :: Proxy Postgres) f db (defaultDatabasePatchDecoder @Postgres @Db)
121268 where
122− f :: IsDatabaseEntity Postgres entityType => DatabaseEntity Postgres Db entityType -> DatabaseEntity Postgres Db entityType -> Identity (Const (Seq Change) entityType)
123− f (DatabaseEntity descriptor) _ = pure $ Const $
269+ f :: forall entityType. IsDatabaseEntity Postgres entityType => DatabaseEntity Postgres Db entityType -> EntityPatchDecoder Postgres entityType -> IO (EntityPatch entityType)
270+ f (DatabaseEntity descriptor) decoder =
124271 let n = view dbEntityName descriptor
125272 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