ihp-datasync (empty) → 1.5.0
raw patch · 23 files changed
+4036/−0 lines, 23 filesdep +aesondep +asyncdep +attoparsec
Dependencies added: aeson, async, attoparsec, base, bytestring, case-insensitive, classy-prelude, containers, deepseq, haskell-src-exts, haskell-src-meta, hasql, hasql-dynamic-statements, hasql-mapping, hasql-pool, hasql-postgresql-types, hasql-transaction, hspec, http-media, http-types, ihp, ihp-hsx, ihp-log, interpolate, mono-traversable, mtl, postgresql-types, safe-exceptions, scientific, stm, template-haskell, text, time, transformers, typerep-map, unliftio, unordered-containers, uuid, vault, vector, wai, wai-websockets, warp, websockets
Files
- IHP/DataSync/ChangeNotifications.hs +303/−0
- IHP/DataSync/Controller.hs +24/−0
- IHP/DataSync/ControllerImpl.hs +562/−0
- IHP/DataSync/DynamicQuery.hs +241/−0
- IHP/DataSync/DynamicQueryCompiler.hs +234/−0
- IHP/DataSync/Hasql.hs +43/−0
- IHP/DataSync/Pool.hs +32/−0
- IHP/DataSync/REST/Controller.hs +219/−0
- IHP/DataSync/REST/Routes.hs +54/−0
- IHP/DataSync/REST/Types.hs +12/−0
- IHP/DataSync/Role.hs +99/−0
- IHP/DataSync/RowLevelSecurity.hs +273/−0
- IHP/DataSync/TypedEncoder.hs +251/−0
- IHP/DataSync/Types.hs +76/−0
- LICENSE +21/−0
- Test/DataSync/ChangeNotifications.hs +328/−0
- Test/DataSync/DataSyncIntegrationSpec.hs +535/−0
- Test/DataSync/DynamicQueryCompiler.hs +291/−0
- Test/DataSync/RLSIntegrationSpec.hs +122/−0
- Test/DataSync/TypedEncoder.hs +127/−0
- Test/Main.hs +17/−0
- changelog.md +18/−0
- ihp-datasync.cabal +154/−0
+ IHP/DataSync/ChangeNotifications.hs view
@@ -0,0 +1,303 @@+module IHP.DataSync.ChangeNotifications+( channelName+, ChangeNotification (..)+, Change (..)+, ChangeSet (..)+, createNotificationFunction+, installTableChangeTriggers+, makeCachedInstallTableChangeTriggers+, makeInstallTableChangeTriggers+, retrieveChanges+, installTableChangeTriggersSession+, retrieveChangesSession+) where++import IHP.Prelude+import qualified Hasql.Pool+import Data.String.Interpolate.IsString (i)+import qualified Data.Text as Text+import Data.Aeson+import Data.Aeson.TH+import qualified IHP.DataSync.RowLevelSecurity as RLS+import qualified Data.Map.Strict as Map+import Control.Concurrent.MVar+import qualified Data.UUID as UUID+import qualified Hasql.Decoders as Decoders+import qualified Hasql.Encoders as Encoders+import qualified Hasql.Statement as Statement+import qualified Hasql.Session as Session+import IHP.DataSync.Hasql (runSession)+import IHP.PGVersion (defaultUuidFunction)+import IHP.Environment (Environment(..))+import System.IO.Unsafe (unsafePerformIO)++data ChangeNotification+ = DidInsert { id :: !UUID }+ | DidUpdate { id :: !UUID, changeSet :: !ChangeSet }+ | DidUpdateLarge { id :: !UUID, payloadId :: !UUID }+ | DidDelete { id :: !UUID }+ deriving (Eq, Show)++data ChangeSet+ = InlineChangeSet { changeSet :: ![Change] } -- | When the patch fits into the 8000 bytes limit of @pg_notify@+ | ExternalChangeSet { largePgNotificationId :: !UUID } -- | The patch is over 8000 bytes, so we have stored it in the @large_pg_notifications@ table+ deriving (Eq, Show)++data Change+ = Change { col :: !Text, new :: !Value }+ | AppendChange { col :: !Text, append :: !Text }+ deriving (Eq, Show)++-- | Returns the sql code to set up a database trigger. Mainly used by 'watchInsertOrUpdateTable'.+--+-- The function body is always updated via @CREATE OR REPLACE FUNCTION@ (no table lock needed).+-- The trigger DDL (@CREATE TRIGGER@) requires @ShareRowExclusiveLock@ on the table,+-- which conflicts with @RowExclusiveLock@ from writers (INSERT\/UPDATE\/DELETE\/COPY FROM).+-- It is only executed when the triggers don't already exist, with a short @lock_timeout@+-- to fail fast rather than block behind long-running writers.+createNotificationFunction :: Text -> RLS.TableWithRLS -> Text+createNotificationFunction uuidFunction table = [i|+ DO $$+ BEGIN+ -- Always update the function body. CREATE OR REPLACE FUNCTION only locks+ -- the function in pg_proc, not the table, so this is safe to run+ -- unconditionally and ensures the function stays up to date.+ CREATE OR REPLACE FUNCTION "#{functionName}"() RETURNS TRIGGER AS $BODY$+ DECLARE+ payload TEXT;+ large_pg_notification_id UUID;+ changeset JSON;+ BEGIN+ CASE TG_OP+ WHEN 'UPDATE' THEN+ SELECT coalesce(json_agg(+ CASE+ WHEN jsonb_typeof(pre.value) = 'string'+ AND jsonb_typeof(post.value) = 'string'+ AND length(post.value #>> '{}') > length(pre.value #>> '{}')+ AND starts_with(post.value #>> '{}', pre.value #>> '{}')+ THEN json_build_object(+ 'col', pre.key,+ 'append', substring(post.value #>> '{}' from length(pre.value #>> '{}') + 1)+ )+ ELSE json_build_object('col', pre.key, 'new', post.value)+ END+ ), '[]'::json)+ FROM jsonb_each(to_jsonb(OLD)) AS pre+ CROSS JOIN jsonb_each(to_jsonb(NEW)) AS post+ WHERE pre.key = post.key AND pre.value IS DISTINCT FROM post.value+ INTO changeset;+ payload := json_build_object(+ 'UPDATE', NEW.id::text,+ 'CHANGESET', changeset+ )::text;+ IF octet_length(payload) > 7800 THEN+ INSERT INTO large_pg_notifications (payload) VALUES (changeset) RETURNING id INTO large_pg_notification_id;+ payload := json_build_object(+ 'UPDATE', NEW.id::text,+ 'CHANGESET', large_pg_notification_id::text+ )::text;+ DELETE FROM large_pg_notifications WHERE created_at < CURRENT_TIMESTAMP - interval '30s';+ END IF;+ PERFORM pg_notify(+ '#{channelName table}',+ payload+ );+ WHEN 'DELETE' THEN+ PERFORM pg_notify(+ '#{channelName table}',+ (json_build_object('DELETE', OLD.id)::text)+ );+ WHEN 'INSERT' THEN+ PERFORM pg_notify(+ '#{channelName table}',+ json_build_object('INSERT', NEW.id)::text+ );+ END CASE;+ RETURN new;+ END;+ $BODY$ language plpgsql;++ -- Only install triggers if they don't already exist. CREATE TRIGGER+ -- takes ShareRowExclusiveLock on the table, which conflicts with+ -- RowExclusiveLock from INSERT/UPDATE/DELETE. Skipping this when+ -- triggers are already in place avoids blocking behind long-running+ -- writers or COPY FROM.+ --+ -- When triggers ARE missing (first install, new table), use a short+ -- lock_timeout so we fail fast rather than blocking writers+ -- indefinitely. The Haskell-side MVar cache removes its entry on+ -- failure, so the next WebSocket connection will retry.+ IF NOT EXISTS (+ SELECT 1 FROM pg_trigger WHERE tgname = '#{insertTriggerName}'+ AND tgrelid = '#{tableName}'::regclass+ ) THEN+ -- Use a short lock_timeout so we fail fast if a long-running+ -- writer holds RowExclusiveLock on the table (which conflicts+ -- with CREATE TRIGGER's ShareRowExclusiveLock). The error+ -- propagates to Haskell where the MVar cache removes its entry,+ -- allowing the next WebSocket connection to retry.+ SET LOCAL lock_timeout = '5s';+ BEGIN+ CREATE TRIGGER "#{insertTriggerName}" AFTER INSERT ON "#{tableName}" FOR EACH ROW EXECUTE PROCEDURE "#{functionName}"();+ CREATE TRIGGER "#{updateTriggerName}" AFTER UPDATE ON "#{tableName}" FOR EACH ROW EXECUTE PROCEDURE "#{functionName}"();+ CREATE TRIGGER "#{deleteTriggerName}" AFTER DELETE ON "#{tableName}" FOR EACH ROW EXECUTE PROCEDURE "#{functionName}"();+ EXCEPTION+ WHEN duplicate_object THEN null;+ END;+ END IF;++ BEGIN+ IF NOT EXISTS (+ SELECT FROM pg_catalog.pg_class c+ JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace+ WHERE c.relname = 'large_pg_notifications'+ AND n.nspname = 'public'+ ) THEN+ CREATE UNLOGGED TABLE large_pg_notifications (+ id UUID DEFAULT #{uuidFunction}() PRIMARY KEY NOT NULL,+ payload TEXT DEFAULT NULL,+ created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL+ );+ CREATE INDEX large_pg_notifications_created_at_index ON large_pg_notifications (created_at);+ END IF;+ EXCEPTION+ WHEN duplicate_table THEN null;+ END;+ END; $$+|]++ where+ tableName = Text.replace "\"" "\"\"" table.tableName++ functionName = "notify_did_change_" <> tableName+ insertTriggerName = "did_insert_" <> tableName+ updateTriggerName = "did_update_" <> tableName+ deleteTriggerName = "did_delete_" <> tableName++-- Statements++retrieveChangesStatement :: Statement.Statement UUID Text+retrieveChangesStatement = Statement.preparable+ "SELECT payload FROM large_pg_notifications WHERE id = $1 LIMIT 1"+ (Encoders.param (Encoders.nonNullable Encoders.uuid))+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text)))++-- Sessions++installTableChangeTriggersSession :: Text -> RLS.TableWithRLS -> Session.Session ()+installTableChangeTriggersSession uuidFunction table =+ Session.script (createNotificationFunction uuidFunction table)++retrieveChangesSession :: UUID -> Session.Session Text+retrieveChangesSession uuid = Session.statement uuid retrieveChangesStatement++-- IO API (thin wrappers)++installTableChangeTriggers :: Hasql.Pool.Pool -> RLS.TableWithRLS -> IO ()+installTableChangeTriggers pool tableNameRLS = do+ uuidFunction <- defaultUuidFunction+ runSession pool (installTableChangeTriggersSession uuidFunction tableNameRLS)+ pure ()++-- | In development, always re-run trigger SQL because @make db@ drops and+-- recreates the database, destroying previously installed triggers.+-- In production, cache per table to avoid unnecessary work.+makeInstallTableChangeTriggers :: Environment -> Hasql.Pool.Pool -> IO (RLS.TableWithRLS -> IO ())+makeInstallTableChangeTriggers Development pool = pure (installTableChangeTriggers pool)+makeInstallTableChangeTriggers Production pool = makeCachedInstallTableChangeTriggers pool++-- | Process-global lock map for trigger installation. Each table gets an MVar+-- that serializes installation in Haskell-land, so only one connection per+-- table hits the database while others wait cheaply without consuming a pool+-- connection.+-- See: https://github.com/digitallyinduced/ihp/issues/2467+{-# NOINLINE globalTriggerInstallLocks #-}+globalTriggerInstallLocks :: MVar (Map.Map RLS.TableWithRLS (MVar ()))+globalTriggerInstallLocks = unsafePerformIO (newMVar Map.empty)++-- | Wraps 'installTableChangeTriggers' with a process-global per-table lock+-- so each table's triggers are only installed once per process lifetime.+-- Concurrent callers for the same table block on an MVar in Haskell (not on+-- a database connection), preventing pool exhaustion from DDL lock waits.+-- If installation fails, the lock is removed so future connections can retry.+makeCachedInstallTableChangeTriggers :: Hasql.Pool.Pool -> IO (RLS.TableWithRLS -> IO ())+makeCachedInstallTableChangeTriggers pool = do+ pure \tableName -> do+ -- Atomically check if this table already has a lock entry.+ -- If not, create an empty MVar and register as the installer.+ (lock, weAreInstaller) <- modifyMVar globalTriggerInstallLocks \locks ->+ case Map.lookup tableName locks of+ Just existingLock ->+ pure (locks, (existingLock, False))+ Nothing -> do+ newLock <- newEmptyMVar+ pure (Map.insert tableName newLock locks, (newLock, True))++ if weAreInstaller+ then do+ -- We won the race — do the actual install.+ -- On success, signal waiters. On failure, remove the lock+ -- so future connections can retry, then re-throw.+ installTableChangeTriggers pool tableName+ `catch` \e -> do+ modifyMVar_ globalTriggerInstallLocks \locks ->+ pure (Map.delete tableName locks)+ throwIO (e :: SomeException)+ putMVar lock ()+ else+ -- Another connection is installing (or has installed).+ -- This blocks until the installer calls putMVar, then+ -- immediately returns the () without taking it.+ readMVar lock++-- | Returns the event name of the event that the pg notify trigger dispatches+channelName :: RLS.TableWithRLS -> ByteString+channelName table = "did_change_" <> (cs $ Text.replace "\"" "\"\"" table.tableName)+++instance FromJSON ChangeNotification where+ parseJSON = withObject "ChangeNotification" $ \values -> insert values <|> update values <|> delete values+ where+ insert values = do+ id <- values .: "INSERT"+ pure DidInsert { id }+ update values = do+ id <- values .: "UPDATE"+ changeSet <- values .: "CHANGESET"+ pure $ DidUpdate id changeSet+ delete values = DidDelete <$> values .: "DELETE"++instance FromJSON ChangeSet where+ parseJSON array@(Array v) = do+ changeSet <- parseJSON array+ pure InlineChangeSet { changeSet }+ parseJSON (String id) = do+ case UUID.fromText id of+ Just largePgNotificationId -> pure ExternalChangeSet { largePgNotificationId }+ Nothing -> fail "Invalid UUID"+ parseJSON invalid = fail $ cs ("Expected Array or String for ChangeSet, got: " <> tshow invalid)++instance FromJSON Change where+ parseJSON = withObject "Change" $ \values -> do+ col <- values .: "col"+ (Change col <$> values .: "new")+ <|> (AppendChange col <$> values .: "append")+-- | The @pg_notify@ function has a payload limit of 8000 bytes. When a record update is larger than the payload size+-- we store the patch in the @large_pg_notifications@ table and pass over the id to the patch.+--+-- This function retrieves the patch from the @large_pg_notifications@ table, or directly returns the patch+-- when it's less than 8000 bytes.+retrieveChanges :: Hasql.Pool.Pool -> ChangeSet -> IO [Change]+retrieveChanges _pool InlineChangeSet { changeSet } = pure changeSet+retrieveChanges pool ExternalChangeSet { largePgNotificationId } = do+ payload <- runSession pool (retrieveChangesSession largePgNotificationId)+ case eitherDecodeStrictText payload of+ Left e -> fail e+ Right changes -> pure changes++instance ToJSON Change where+ toJSON Change { col, new } = object ["col" .= col, "new" .= new]+ toJSON AppendChange { col, append } = object ["col" .= col, "append" .= append]+$(deriveToJSON defaultOptions 'InlineChangeSet)+$(deriveToJSON defaultOptions 'DidInsert)
+ IHP/DataSync/Controller.hs view
@@ -0,0 +1,24 @@+{-# LANGUAGE UndecidableInstances #-}+module IHP.DataSync.Controller where++import IHP.ControllerPrelude hiding (OrderByClause)++import IHP.DataSync.Types+import IHP.DataSync.RowLevelSecurity+import qualified IHP.DataSync.ChangeNotifications as ChangeNotifications+import IHP.DataSync.ControllerImpl (runDataSyncController)+import IHP.DataSync.DynamicQueryCompiler (camelCaseRenamer)++instance (+ Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => WSApp DataSyncController where+ initialState = DataSyncController++ run = do+ let hasqlPool = ?modelContext.hasqlPool+ ensureRLSEnabled <- makeCachedEnsureRLSEnabled hasqlPool+ installTableChangeTriggers <- ChangeNotifications.makeInstallTableChangeTriggers ?context.frameworkConfig.environment hasqlPool+ runDataSyncController hasqlPool ensureRLSEnabled installTableChangeTriggers (receiveData @ByteString) sendJSON (\_ _ -> pure ()) (\_ -> camelCaseRenamer)
+ IHP/DataSync/ControllerImpl.hs view
@@ -0,0 +1,562 @@+{-# LANGUAGE UndecidableInstances #-}+module IHP.DataSync.ControllerImpl where++import IHP.ControllerPrelude hiding (OrderByClause, sqlQuery, sqlExec, sqlQueryScalar)+import qualified Control.Exception.Safe as Exception+import qualified IHP.Log as Log+import qualified Data.Aeson as Aeson+import qualified Data.Aeson.Key as Aeson++import Data.Aeson.TH+import qualified Hasql.Decoders as Decoders+import qualified Hasql.Statement as Hasql+import qualified Hasql.Pool+import qualified Hasql.Session as Session+import IHP.DataSync.Hasql (runSession, runSessionOnConnection, withDedicatedConnection)+import qualified Data.HashMap.Strict as HashMap+import qualified Data.UUID.V4 as UUID+import qualified Control.Concurrent.MVar as MVar+import IHP.DataSync.Types+import IHP.DataSync.RowLevelSecurity+import IHP.DataSync.DynamicQuery+import IHP.DataSync.DynamicQueryCompiler+import IHP.DataSync.TypedEncoder (makeCachedColumnTypeLookup, typedAesonValueToSnippet, lookupColumnType)+import qualified Hasql.DynamicStatements.Snippet as Snippet+import Hasql.DynamicStatements.Snippet (Snippet)+import qualified IHP.DataSync.ChangeNotifications as ChangeNotifications+import qualified IHP.PGListener as PGListener+import qualified Data.Set as Set+import GHC.Conc (ThreadId, myThreadId, atomically)+import Control.Concurrent.QSemN+import Control.Concurrent.STM.TVar+import qualified Data.List as List++$(deriveFromJSON defaultOptions ''DataSyncMessage)+$(deriveToJSON defaultOptions { omitNothingFields = True } 'DataSyncResult)++type EnsureRLSEnabledFn = Text -> IO TableWithRLS+type InstallTableChangeTriggerFn = TableWithRLS -> IO ()+type SendJSONFn = DataSyncResponse -> IO ()+type HandleCustomMessageFn = (DataSyncResponse -> IO ()) -> DataSyncMessage -> IO ()++runDataSyncController ::+ ( HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ , ?context :: ControllerContext+ , ?modelContext :: ModelContext+ , ?request :: Request+ , ?state :: IORef DataSyncController+ , Typeable CurrentUserRecord+ , HasNewSessionUrl CurrentUserRecord+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ ) => Hasql.Pool.Pool -> EnsureRLSEnabledFn -> InstallTableChangeTriggerFn -> IO ByteString -> SendJSONFn -> HandleCustomMessageFn -> (Text -> Renamer) -> IO ()+runDataSyncController hasqlPool ensureRLSEnabled installTableChangeTriggers receiveData sendJSON handleCustomMessage renamer = do+ setState DataSyncReady { subscriptions = HashMap.empty, transactions = HashMap.empty }++ columnTypeLookup <- makeCachedColumnTypeLookup hasqlPool+ handleMessage :: DataSyncMessage -> IO () <- buildMessageHandler hasqlPool ensureRLSEnabled installTableChangeTriggers sendJSON handleCustomMessage renamer columnTypeLookup+++ sem <- newQSemN (maxSubscriptionsPerConnection * 2) -- needs to be larger than the subscriptions limit to trigger an error on overload. otherwise an overflow of connections might queue up silently++ -- Track Asyncs so we can cancel/wait on socket close+ childrenVar <- newTVarIO (HashMap.empty :: HashMap ThreadId (Async ()))++ let spawnWorker decodedMessage = do+ tidReady <- MVar.newEmptyMVar+ a <- asyncWithUnmask \unmask -> do+ -- Register myself+ tid <- myThreadId+ MVar.putMVar tidReady tid+ -- Take/release concurrency slot entirely inside the worker+ Exception.bracket_ (waitQSemN sem 1) (signalQSemN sem 1) do+ Exception.finally+ (unmask do+ result <- Exception.try (handleMessage decodedMessage)+ case result of+ Left (e :: Exception.SomeException) -> do+ let requestId = decodedMessage.requestId+ let errorMessage = cs (displayException e)+ Log.error (tshow e)+ sendJSON DataSyncError { requestId, errorMessage }+ Right _ -> pure ()+ )+ -- Self-deregister+ (do+ tid' <- myThreadId+ atomically $ modifyTVar' childrenVar (HashMap.delete tid')+ )+ -- Parent stores the Async by ThreadId (no race thanks to tidReady)+ tid <- MVar.takeMVar tidReady+ atomically $ modifyTVar' childrenVar (HashMap.insert tid a)++ let loop = forever do+ msg <- Aeson.eitherDecodeStrict' <$> receiveData+ case msg of+ Right decoded -> spawnWorker decoded+ Left err -> sendJSON FailedToDecodeMessageError { errorMessage = cs err }++ -- On websocket close: cancel and drain all children+ loop `Exception.finally` do+ m <- readTVarIO childrenVar+ let handles = HashMap.elems m+ mapM_ cancel handles+ mapM_ (const (pure ())) =<< mapM waitCatch handles+{-# INLINE runDataSyncController #-}+++buildMessageHandler ::+ ( HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ , ?context :: ControllerContext+ , ?modelContext :: ModelContext+ , ?request :: Request+ , ?state :: IORef DataSyncController+ , Typeable CurrentUserRecord+ , HasNewSessionUrl CurrentUserRecord+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ )+ => Hasql.Pool.Pool -> EnsureRLSEnabledFn -> InstallTableChangeTriggerFn -> SendJSONFn -> HandleCustomMessageFn -> (Text -> Renamer) -> (Text -> IO ColumnTypeInfo) -> IO (DataSyncMessage -> IO ())+buildMessageHandler hasqlPool ensureRLSEnabled installTableChangeTriggers sendJSON handleCustomMessage renamer columnTypeLookup = do+ getRLSColumns <- makeCachedRLSPolicyColumns hasqlPool+ pure (handleMessage getRLSColumns)+ where+ pgListener = ?request.pgListener+ handleMessage :: (Text -> IO (Set.Set Text)) -> DataSyncMessage -> IO ()+ handleMessage getRLSColumns DataSyncQuery { query, requestId, transactionId } = do+ ensureRLSEnabled (query.table)++ columnTypes <- columnTypeLookup query.table+ let querySnippet = compileQueryTyped (renamer query.table) columnTypes query+ let stmt = compiledQueryStatement querySnippet++ result :: [[Field]] <- sqlQueryWithRLSAndTransactionId hasqlPool transactionId stmt++ sendJSON DataSyncResult { result, requestId }++ handleMessage getRLSColumns CreateDataSubscription { query, requestId } = do+ ensureBelowSubscriptionsLimit++ tableNameRLS <- ensureRLSEnabled (query.table)++ subscriptionId <- UUID.nextRandom++ -- Allocate the close handle as early as possible+ -- to make DeleteDataSubscription calls succeed even when the DataSubscription is+ -- not fully set up yet+ close <- MVar.newEmptyMVar+ atomicModifyIORef'' ?state (\state -> state |> modify #subscriptions (HashMap.insert subscriptionId close))++ columnTypes <- columnTypeLookup query.table+ let querySnippet = compileQueryTyped (renamer query.table) columnTypes query+ let stmt = compiledQueryStatement querySnippet++ result :: [[Field]] <- sqlQueryWithRLS hasqlPool stmt++ let tableName = query.table++ -- Compute "sensitive columns": the union of columns referenced in the+ -- WHERE clause and columns referenced in RLS policies. When an UPDATE+ -- only touches columns outside this set, the record cannot leave the+ -- result set or change RLS visibility, so we can skip the EXISTS check.+ rlsCols <- getRLSColumns tableName+ let whereColumns = maybe Set.empty conditionColumns (query.whereCondition)+ let whereColumnsDb = Set.map (renamer tableName).fieldToColumn whereColumns+ let sensitiveColumns = Set.union whereColumnsDb rlsCols++ -- We need to keep track of all the ids of entities we're watching to make+ -- sure that we only send update notifications to clients that can actually+ -- access the record (e.g. if a RLS policy denies access)+ let watchedRecordIds = recordIds result++ -- Store it in IORef as an INSERT requires us to add an id+ watchedRecordIdsRef <- newIORef (Set.fromList watchedRecordIds)++ -- Make sure the database triggers are there+ installTableChangeTriggers tableNameRLS++ let handleUpdate id getChanges = do+ isWatchingRecord <- Set.member id <$> readIORef watchedRecordIdsRef+ when isWatchingRecord do+ changes <- getChanges+ let changedCols = Set.fromList (map (.col) changes)+ let affectsFilterOrRLS = not (Set.disjoint changedCols sensitiveColumns)+ let (changeSetVal, appendSetVal) = changesToValue (renamer tableName) changes+ if affectsFilterOrRLS+ then do+ let existsSnippet = Snippet.sql "SELECT EXISTS(SELECT * FROM (" <> querySnippet <> Snippet.sql ") AS records WHERE records.id = " <> uuidParam id <> Snippet.sql " LIMIT 1)"+ let existsStmt = Snippet.toPreparableStatement existsSnippet (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))+ isRecordInResultSet :: Bool <- sqlQueryScalarWithRLS hasqlPool existsStmt+ if isRecordInResultSet+ then sendJSON DidUpdate { subscriptionId, id, changeSet = changeSetVal, appendSet = appendSetVal }+ else do+ modifyIORef' watchedRecordIdsRef (Set.delete id)+ sendJSON DidDelete { subscriptionId, id }+ else+ sendJSON DidUpdate { subscriptionId, id, changeSet = changeSetVal, appendSet = appendSetVal }++ let callback notification = case notification of+ ChangeNotifications.DidInsert { id } -> do+ -- The new record could not be accessible to the current user with a RLS policy+ -- E.g. it could be a new record in a 'projects' table, but the project belongs+ -- to a different user, and thus the current user should not be able to see it.+ --+ -- The new record could also be not part of the WHERE condition of the initial query.+ -- Therefore we need to use the subscriptions WHERE condition to fetch the new record here.+ --+ -- To honor the RLS policies we therefore need to fetch the record as the current user+ -- If the result set is empty, we know the record is not accesible to us+ let filterSnippet = Snippet.sql "SELECT * FROM (" <> querySnippet <> Snippet.sql ") AS records WHERE records.id = " <> uuidParam id <> Snippet.sql " LIMIT 1"+ let filterStmt = Snippet.toPreparableStatement (wrapDynamicQuery filterSnippet) dynamicRowDecoder+ newRecord :: [[Field]] <- sqlQueryWithRLS hasqlPool filterStmt++ case headMay newRecord of+ Just record -> do+ -- Add the new record to 'watchedRecordIdsRef'+ -- Otherwise the updates and deletes will not be dispatched to the client+ modifyIORef' watchedRecordIdsRef (Set.insert id)++ sendJSON DidInsert { subscriptionId, record }+ Nothing -> pure ()+ ChangeNotifications.DidUpdate { id, changeSet } ->+ handleUpdate id (ChangeNotifications.retrieveChanges hasqlPool changeSet)+ ChangeNotifications.DidUpdateLarge { id, payloadId } ->+ handleUpdate id (ChangeNotifications.retrieveChanges hasqlPool (ChangeNotifications.ExternalChangeSet { largePgNotificationId = payloadId }))+ ChangeNotifications.DidDelete { id } -> do+ -- Only send the notifcation if the deleted record was part of the initial+ -- results set+ isWatchingRecord <- Set.member id <$> readIORef watchedRecordIdsRef+ when isWatchingRecord do+ sendJSON DidDelete { subscriptionId, id }++ let subscribe = PGListener.subscribeJSON (ChangeNotifications.channelName tableNameRLS) callback pgListener+ let unsubscribe subscription = PGListener.unsubscribe subscription pgListener++ Exception.bracket subscribe unsubscribe \channelSubscription -> do+ sendJSON DidCreateDataSubscription { subscriptionId, requestId, result }++ MVar.takeMVar close++ handleMessage getRLSColumns CreateCountSubscription { query, requestId } = do+ ensureBelowSubscriptionsLimit++ tableNameRLS <- ensureRLSEnabled query.table++ subscriptionId <- UUID.nextRandom++ -- Allocate the close handle as early as possible+ -- to make DeleteDataSubscription calls succeed even when the CountSubscription is+ -- not fully set up yet+ close <- MVar.newEmptyMVar+ atomicModifyIORef'' ?state (\state -> state |> modify #subscriptions (HashMap.insert subscriptionId close))++ columnTypes <- columnTypeLookup query.table+ let querySnippet = compileQueryTyped (renamer query.table) columnTypes query++ let countSnippet = Snippet.sql "SELECT COUNT(*) FROM (" <> querySnippet <> Snippet.sql ") AS _inner"+ let countDecoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable (fromIntegral <$> Decoders.int8)))+ let countStmt = Snippet.toPreparableStatement countSnippet countDecoder++ count :: Int <- sqlQueryScalarWithRLS hasqlPool countStmt+ countRef <- newIORef count++ installTableChangeTriggers tableNameRLS++ let+ callback :: ChangeNotifications.ChangeNotification -> IO ()+ callback _ = do+ newCount :: Int <- sqlQueryScalarWithRLS hasqlPool countStmt+ lastCount <- readIORef countRef++ when (newCount /= lastCount) do+ writeIORef countRef newCount+ sendJSON DidChangeCount { subscriptionId, count = newCount }++ let subscribe = PGListener.subscribeJSON (ChangeNotifications.channelName tableNameRLS) callback pgListener+ let unsubscribe subscription = PGListener.unsubscribe subscription pgListener++ Exception.bracket subscribe unsubscribe \channelSubscription -> do+ sendJSON DidCreateCountSubscription { subscriptionId, requestId, count }++ MVar.takeMVar close++ handleMessage getRLSColumns DeleteDataSubscription { requestId, subscriptionId } = do+ DataSyncReady { subscriptions } <- getState+ case HashMap.lookup subscriptionId subscriptions of+ Just closeSignalMVar -> do+ -- Cancel table watcher+ MVar.putMVar closeSignalMVar ()++ atomicModifyIORef'' ?state (\state -> state |> modify #subscriptions (HashMap.delete subscriptionId))++ sendJSON DidDeleteDataSubscription { subscriptionId, requestId }+ Nothing -> sendJSON DataSyncError { requestId, errorMessage = "Failed to delete DataSubscription, could not find DataSubscription with id " <> tshow subscriptionId }++ handleMessage getRLSColumns CreateRecordMessage { table, record, requestId, transactionId } = do+ ensureRLSEnabled table++ columnTypes <- columnTypeLookup table++ let pairsList = record+ |> HashMap.toList+ |> map (\(fieldName, val) ->+ let col = (renamer table).fieldToColumn fieldName+ in (col, lookupColumnType columnTypes col, val)+ )++ let columns = map (\(c,_,_) -> c) pairsList+ let valueSnippets = map (\(_, colType, val) -> typedAesonValueToSnippet colType val) pairsList+ let insertResult = compileInsert table columns valueSnippets (renamer table) columnTypes+ let stmt = compiledQueryStatement insertResult++ result :: [[Field]] <- sqlQueryWriteWithRLSAndTransactionId hasqlPool transactionId stmt++ case result of+ [record] ->+ sendJSON DidCreateRecord { requestId, record }+ otherwise -> sendJSON DataSyncError { requestId, errorMessage = "Unexpected result in CreateRecordMessage handler" }++ pure ()++ handleMessage getRLSColumns CreateRecordsMessage { table, records, requestId, transactionId } = do+ ensureRLSEnabled table++ columnTypes <- columnTypeLookup table++ case head records of+ Nothing -> sendJSON DataSyncError { requestId, errorMessage = "At least one record is required" }+ Just firstRecord -> do+ let fieldNames = HashMap.keys firstRecord+ let columns = map (renamer table).fieldToColumn fieldNames++ let encodeRow object = map+ (\(fieldName, col) -> typedAesonValueToSnippet (lookupColumnType columnTypes col) (fromMaybe Aeson.Null (HashMap.lookup fieldName object)))+ (zip fieldNames columns)+ let valueRows = map encodeRow records++ let insertResult = compileInsertMany table columns valueRows (renamer table) columnTypes+ let stmt = compiledQueryStatement insertResult++ records :: [[Field]] <- sqlQueryWriteWithRLSAndTransactionId hasqlPool transactionId stmt++ sendJSON DidCreateRecords { requestId, records }++ pure ()++ handleMessage getRLSColumns UpdateRecordMessage { table, id, patch, requestId, transactionId } = do+ ensureRLSEnabled table++ columnTypes <- columnTypeLookup table++ let setSql = encodePatchToSetSql (renamer table) columnTypes patch+ let updateResult = compileUpdate table setSql (Snippet.sql "id = " <> uuidParam id) (renamer table) columnTypes+ let stmt = compiledQueryStatement updateResult++ result :: [[Field]] <- sqlQueryWriteWithRLSAndTransactionId hasqlPool transactionId stmt++ case result of+ [record] ->+ sendJSON DidUpdateRecord { requestId, record }+ otherwise -> sendJSON DataSyncError { requestId, errorMessage = "Could not apply the update to the given record. Are you sure the record ID you passed is correct? If the record ID is correct, likely the row level security policy is not making the record visible to the UPDATE operation." }++ pure ()++ handleMessage getRLSColumns UpdateRecordsMessage { table, ids, patch, requestId, transactionId } = do+ ensureRLSEnabled table++ columnTypes <- columnTypeLookup table++ let setSql = encodePatchToSetSql (renamer table) columnTypes patch+ let inList = mconcat $ List.intersperse (Snippet.sql ", ") (map uuidParam ids)+ let updateResult = compileUpdate table setSql (Snippet.sql "id IN (" <> inList <> Snippet.sql ")") (renamer table) columnTypes+ let stmt = compiledQueryStatement updateResult++ records <- sqlQueryWriteWithRLSAndTransactionId hasqlPool transactionId stmt++ sendJSON DidUpdateRecords { requestId, records }++ pure ()++ handleMessage getRLSColumns DeleteRecordMessage { table, id, requestId, transactionId } = do+ ensureRLSEnabled table++ let deleteSnippet = Snippet.sql ("DELETE FROM " <> quoteIdentifier table <> " WHERE id = ") <> uuidParam id+ let stmt = Snippet.toPreparableStatement deleteSnippet Decoders.noResult+ sqlExecWithRLSAndTransactionId hasqlPool transactionId stmt++ sendJSON DidDeleteRecord { requestId }++ handleMessage getRLSColumns DeleteRecordsMessage { table, ids, requestId, transactionId } = do+ ensureRLSEnabled table++ let inList = mconcat $ List.intersperse (Snippet.sql ", ") (map uuidParam ids)+ let stmt = Snippet.toPreparableStatement (Snippet.sql ("DELETE FROM " <> quoteIdentifier table <> " WHERE id IN (") <> inList <> Snippet.sql ")") Decoders.noResult+ sqlExecWithRLSAndTransactionId hasqlPool transactionId stmt++ sendJSON DidDeleteRecords { requestId }++ handleMessage getRLSColumns StartTransaction { requestId } = do+ ensureBelowTransactionLimit++ transactionId <- UUID.nextRandom++ withDedicatedConnection ?context.frameworkConfig.databaseUrl \connection -> do+ transactionSignal <- MVar.newEmptyMVar++ runSessionOnConnection connection $ do+ Session.script "BEGIN"+ setRLSConfigSession++ let transaction = DataSyncTransaction+ { id = transactionId+ , connection+ , close = transactionSignal+ }++ atomicModifyIORef'' ?state (\state -> state |> modify #transactions (HashMap.insert transactionId transaction))++ sendJSON DidStartTransaction { requestId, transactionId }++ MVar.takeMVar transactionSignal++ atomicModifyIORef'' ?state (\state -> state |> modify #transactions (HashMap.delete transactionId))++ handleMessage getRLSColumns RollbackTransaction { requestId, id } = do+ DataSyncTransaction { id, close, connection } <- findTransactionById id++ runSessionOnConnection connection (Session.script "ROLLBACK")+ MVar.putMVar close ()++ sendJSON DidRollbackTransaction { requestId, transactionId = id }++ handleMessage getRLSColumns CommitTransaction { requestId, id } = do+ DataSyncTransaction { id, close, connection } <- findTransactionById id++ runSessionOnConnection connection (Session.script "COMMIT")+ MVar.putMVar close ()++ sendJSON DidCommitTransaction { requestId, transactionId = id }++ handleMessage _getRLSColumns otherwise = handleCustomMessage sendJSON otherwise++changesToValue :: Renamer -> [ChangeNotifications.Change] -> (Maybe Value, Maybe Value)+changesToValue renamer changes = (maybeObject replacePairs, maybeObject appendPairs)+ where+ maybeObject [] = Nothing+ maybeObject pairs = Just (object pairs)+ replacePairs = mapMaybe toReplacePair changes+ appendPairs = mapMaybe toAppendPair changes+ toReplacePair ChangeNotifications.Change { col, new } =+ Just $ (Aeson.fromText $ renamer.columnToField col) .= new+ toReplacePair _ = Nothing+ toAppendPair ChangeNotifications.AppendChange { col, append } =+ Just $ (Aeson.fromText $ renamer.columnToField col) .= append+ toAppendPair _ = Nothing++findTransactionById :: (?state :: IORef DataSyncController) => UUID -> IO DataSyncTransaction+findTransactionById transactionId = do+ transactions <- (.transactions) <$> readIORef ?state+ case HashMap.lookup transactionId transactions of+ Just transaction -> pure transaction+ Nothing -> Exception.throwIO (userError ("No transaction with id " <> cs (tshow transactionId)))++-- | Allow max 10 concurrent transactions per connection to avoid running out of database connections+--+-- Each transaction removes a database connection from the connection pool. If we don't limit the transactions,+-- a single user could take down the application by starting more than the pool size (HASQL_POOL_SIZE)+-- concurrent transactions. Then all database connections are removed from the connection pool and further database+-- queries for other users will fail.+--+ensureBelowTransactionLimit :: (?state :: IORef DataSyncController, ?context :: ControllerContext) => IO ()+ensureBelowTransactionLimit = do+ transactions <- (.transactions) <$> readIORef ?state+ let transactionCount = HashMap.size transactions+ when (transactionCount >= maxTransactionsPerConnection) do+ Exception.throwIO (userError ("You've reached the transaction limit of " <> cs (tshow maxTransactionsPerConnection) <> " transactions"))++ensureBelowSubscriptionsLimit :: (?state :: IORef DataSyncController, ?context :: ControllerContext) => IO ()+ensureBelowSubscriptionsLimit = do+ subscriptions <- (.subscriptions) <$> readIORef ?state+ let subscriptionsCount = HashMap.size subscriptions+ when (subscriptionsCount >= maxSubscriptionsPerConnection) do+ Exception.throwIO (userError ("You've reached the subscriptions limit of " <> cs (tshow maxSubscriptionsPerConnection) <> " subscriptions"))++maxTransactionsPerConnection :: (?context :: ControllerContext) => Int+maxTransactionsPerConnection =+ case getAppConfig @DataSyncMaxTransactionsPerConnection of+ DataSyncMaxTransactionsPerConnection value -> value++maxSubscriptionsPerConnection :: (?context :: ControllerContext) => Int+maxSubscriptionsPerConnection =+ case getAppConfig @DataSyncMaxSubscriptionsPerConnection of+ DataSyncMaxSubscriptionsPerConnection value -> value++-- | Encode a JSON patch (field name -> value) into a SQL SET clause 'Snippet' like @"col1" = $1, "col2" = $2@.+encodePatchToSetSql :: Renamer -> ColumnTypeInfo -> HashMap Text Value -> Snippet+encodePatchToSetSql ren columnTypes patch =+ let pairsList = patch+ |> HashMap.toList+ |> map (\(fieldName, val) ->+ let col = ren.fieldToColumn fieldName+ in (col, lookupColumnType columnTypes col, val)+ )+ encodeSetClause (col, colType, val) =+ Snippet.sql (quoteIdentifier col <> " = ") <> typedAesonValueToSnippet colType val+ setSnippets = map encodeSetClause pairsList+ in mconcat $ List.intersperse (Snippet.sql ", ") setSnippets++sqlQueryWithRLSAndTransactionId ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ , ?state :: IORef DataSyncController+ ) => Hasql.Pool.Pool -> Maybe UUID -> Hasql.Statement () [result] -> IO [result]+sqlQueryWithRLSAndTransactionId _pool (Just transactionId) statement = do+ -- RLS role and user id were already set when the transaction was started+ DataSyncTransaction { connection } <- findTransactionById transactionId+ runSessionOnConnection connection+ (Session.statement () statement)+sqlQueryWithRLSAndTransactionId pool Nothing statement = runSession pool (sqlQueryWithRLSSession statement)++-- | Like 'sqlQueryWithRLSAndTransactionId', but uses a write transaction when no transaction ID is provided.+--+-- Use this for INSERT, UPDATE, or DELETE statements with RETURNING that need+-- to return results (e.g. wrapped with 'wrapDynamicQuery').+sqlQueryWriteWithRLSAndTransactionId ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ , ?state :: IORef DataSyncController+ ) => Hasql.Pool.Pool -> Maybe UUID -> Hasql.Statement () [result] -> IO [result]+sqlQueryWriteWithRLSAndTransactionId _pool (Just transactionId) statement = do+ -- RLS role and user id were already set when the transaction was started+ DataSyncTransaction { connection } <- findTransactionById transactionId+ runSessionOnConnection connection+ (Session.statement () statement)+sqlQueryWriteWithRLSAndTransactionId pool Nothing statement = runSession pool (sqlQueryWriteWithRLSSession statement)++sqlExecWithRLSAndTransactionId ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ , ?state :: IORef DataSyncController+ ) => Hasql.Pool.Pool -> Maybe UUID -> Hasql.Statement () () -> IO ()+sqlExecWithRLSAndTransactionId _pool (Just transactionId) statement = do+ -- RLS role and user id were already set when the transaction was started+ DataSyncTransaction { connection } <- findTransactionById transactionId+ runSessionOnConnection connection+ (Session.statement () statement)+sqlExecWithRLSAndTransactionId pool Nothing statement = runSession pool (sqlExecWithRLSSession statement)+++instance SetField "subscriptions" DataSyncController (HashMap UUID (MVar.MVar ())) where+ setField subscriptions record = record { subscriptions }++instance SetField "transactions" DataSyncController (HashMap UUID DataSyncTransaction) where+ setField transactions record = record { transactions }++atomicModifyIORef'' ref updateFn = atomicModifyIORef' ref (\value -> (updateFn value, ()))
+ IHP/DataSync/DynamicQuery.hs view
@@ -0,0 +1,241 @@+{-# LANGUAGE DeriveAnyClass #-}+{-|+Module: IHP.DataSync.DynamicQuery+Description: The normal IHP query functionality is type-safe. This module provides type-unsafe access to the database.+Copyright: (c) digitally induced GmbH, 2021+-}+module IHP.DataSync.DynamicQuery where++import IHP.ControllerPrelude hiding (OrderByClause)+import qualified Data.Aeson as Aeson+import qualified IHP.QueryBuilder as QueryBuilder+import qualified Hasql.Decoders as Decoders+import qualified Hasql.Encoders as Encoders+import qualified Hasql.Statement as Hasql+import Data.Aeson.TH+import qualified GHC.Generics+import qualified Control.DeepSeq as DeepSeq+import qualified Data.Aeson.KeyMap as Aeson+import qualified Data.Aeson.Key as Aeson+import qualified Data.Scientific as Scientific+import qualified Data.UUID as UUID+import qualified Data.Vector as Vector+import qualified Data.List as List+import qualified Data.HashMap.Strict as HashMap+import qualified Data.Set as Set+import qualified Data.Text as Text+import qualified Hasql.DynamicStatements.Snippet as Snippet+import Hasql.DynamicStatements.Snippet (Snippet)++data Field = Field { fieldName :: Text, fieldValue :: Value }++newtype UndecodedJSON = UndecodedJSON ByteString+ deriving (Show, Eq)+++-- | Similiar to IHP.QueryBuilder.SQLQuery, but is designed to be accessed by external users+--+-- When compiling to SQL we have to be extra careful to escape all identifers and variables in the query.+-- The normal IHP.QueryBuilder doesn't need to be that careful as parts of the input are derived from+-- generated code from the Schema.sql.+--+data DynamicSQLQuery = DynamicSQLQuery+ { table :: !Text+ , selectedColumns :: SelectedColumns+ , whereCondition :: !(Maybe ConditionExpression)+ , orderByClause :: ![OrderByClause]+ , distinctOnColumn :: !(Maybe ByteString)+ , limit :: !(Maybe Int)+ , offset :: !(Maybe Int)+ } deriving (Show, Eq)++data OrderByClause+ = OrderByClause+ { orderByColumn :: !ByteString+ , orderByDirection :: !OrderByDirection }+ | OrderByTSRank { tsvector :: Text, tsquery :: !Text }+ deriving (Show, Eq, GHC.Generics.Generic, DeepSeq.NFData)++-- | Represents a WHERE conditions of a 'DynamicSQLQuery'+data ConditionExpression+ = ColumnExpression { field :: !Text }+ | InfixOperatorExpression+ { left :: !ConditionExpression+ , op :: !ConditionOperator+ , right :: !ConditionExpression+ }+ | LiteralExpression { value :: !Value }+ | CallExpression { functionCall :: !FunctionCall }+ | ListExpression { values :: ![Value] }+ deriving (Show, Eq)++data FunctionCall+ = ToTSQuery { text :: !Text } -- ^ to_tsquery('english', text)+ deriving (Show, Eq, GHC.Generics.Generic, DeepSeq.NFData)++-- | Operators available in WHERE conditions+data ConditionOperator+ = OpEqual -- ^ a = b+ | OpGreaterThan -- ^ a > b+ | OpLessThan -- ^ a < b+ | OpGreaterThanOrEqual -- ^ a >= b+ | OpLessThanOrEqual -- ^ a <= b+ | OpNotEqual -- ^ a <> b+ | OpAnd -- ^ a AND b+ | OpOr -- ^ a OR b+ | OpIs -- ^ a IS b+ | OpIsNot -- ^ a IS NOT b+ | OpTSMatch -- ^ tsvec_a @@ tsvec_b+ | OpIn -- ^ a IN b+ deriving (Show, Eq)++data SelectedColumns+ = SelectAll -- ^ SELECT * FROM table+ | SelectSpecific [Text] -- ^ SELECT a, b, c FROM table+ deriving (Show, Eq)++instance FromJSON ByteString where+ parseJSON (String v) = pure $ cs v+ parseJSON invalid = fail $ cs ("Expected String for ByteString, got: " <> tshow invalid)++instance {-# OVERLAPS #-} ToJSON [Field] where+ toJSON fields = object (map (\Field { fieldName, fieldValue } -> (cs fieldName) .= fieldValue) fields)+ toEncoding fields = pairs $ foldl' (<>) mempty encodedFields+ where+ encodedFields = (map (\Field { fieldName, fieldValue } -> (cs fieldName) .= fieldValue) fields)++-- | Decoder for dynamic query results wrapped with 'wrapDynamicQuery'.+--+-- Each row comes back as a single JSON column (from @row_to_json@), which is then+-- parsed into a list of @Field@ values.+dynamicRowDecoder :: Decoders.Result [[Field]]+dynamicRowDecoder = Decoders.rowList dynamicRowJsonDecoder++-- | Decodes a single row from a @row_to_json@ wrapped query.+-- The row consists of a single JSONB column containing the original row as a JSON object.+dynamicRowJsonDecoder :: Decoders.Row [Field]+dynamicRowJsonDecoder =+ Decoders.column (Decoders.nonNullable Decoders.jsonb)+ |> fmap (\jsonValue ->+ case jsonToFields jsonValue of+ Right fields -> fields+ Left err -> error ("dynamicRowJsonDecoder: Failed to decode JSON row: " <> cs err)+ )++-- | Converts a JSON object (from @row_to_json@) into a list of Fields+jsonToFields :: Value -> Either String [Field]+jsonToFields (Object obj) = Right $ map toField (Aeson.toList obj)+ where+ toField (key, value) = Field+ { fieldName = Aeson.toText key+ , fieldValue = value+ }+jsonToFields other = Left (cs ("Expected JSON object but got: " <> show other))+++-- | Returns a list of all id's in a result+recordIds :: [[Field]] -> [UUID]+recordIds result = result+ |> concat+ |> filter (\Field { fieldName } -> fieldName == "id")+ |> map (.fieldValue)+ |> mapMaybe \case+ Aeson.String text -> UUID.fromText text+ otherwise -> Nothing++++-- | A map from column name to PostgreSQL type name (e.g. @"uuid"@, @"int4"@, @"timestamptz"@)+type ColumnTypeMap = HashMap.HashMap Text Text++-- | Column type information with both O(1) type lookup and database column ordering.+--+-- The 'typeMap' provides fast type lookups for WHERE clause compilation,+-- while 'orderedColumns' preserves the order columns were defined in the schema+-- (from @pg_attribute.attnum@), ensuring @id@ naturally appears first when+-- expanding @SELECT *@.+data ColumnTypeInfo = ColumnTypeInfo+ { typeMap :: !ColumnTypeMap+ , orderedColumns :: ![Text]+ } deriving (Show, Eq)++-- | Encode an Aeson 'Value' as a parameterized SQL 'Snippet'.+--+-- Used for expressions without column context (e.g. bare literals in WHERE clauses).+-- When column types are known, prefer 'IHP.DataSync.TypedEncoder.typedValueParam'+-- for correctly typed parameters.+dynamicValueParam :: Value -> Snippet+dynamicValueParam Aeson.Null = Snippet.sql "NULL"+dynamicValueParam (Aeson.Bool b) = Snippet.encoderAndParam (Encoders.nonNullable Encoders.bool) b+dynamicValueParam (Aeson.String t) = Snippet.encoderAndParam (Encoders.nonNullable Encoders.text) t+dynamicValueParam (Aeson.Number n) =+ case Scientific.floatingOrInteger n of+ Left (d :: Double) -> Snippet.encoderAndParam (Encoders.nonNullable Encoders.float8) d+ Right (i :: Integer) -> Snippet.encoderAndParam (Encoders.nonNullable Encoders.int8) (fromIntegral i :: Int64)+dynamicValueParam (Aeson.Array arr) =+ Snippet.sql "ARRAY[" <> mconcat (List.intersperse (Snippet.sql ", ") (map dynamicValueParam (Vector.toList arr))) <> Snippet.sql "]"+dynamicValueParam (Aeson.Object obj) = Snippet.encoderAndParam (Encoders.nonNullable Encoders.text) (cs (encode (Object obj)) :: Text)++-- | Extracts all column names referenced in a 'ConditionExpression'+conditionColumns :: ConditionExpression -> Set.Set Text+conditionColumns (ColumnExpression field) = Set.singleton field+conditionColumns (InfixOperatorExpression left _ right) = Set.union (conditionColumns left) (conditionColumns right)+conditionColumns (LiteralExpression _) = Set.empty+conditionColumns (CallExpression _) = Set.empty+conditionColumns (ListExpression _) = Set.empty++-- | Wraps a SQL query so that each row is returned as a JSON object.+--+-- Uses a CTE (Common Table Expression) which works for both SELECT queries+-- and DML statements (INSERT, UPDATE, DELETE) with RETURNING.+wrapDynamicQuery :: Snippet -> Snippet+wrapDynamicQuery innerQuery =+ Snippet.sql "WITH _ihp_dynamic_result AS (" <> innerQuery <> Snippet.sql ") SELECT row_to_json(t)::jsonb FROM _ihp_dynamic_result AS t"++-- | Quote a SQL identifier (table name, column name) to prevent SQL injection.+--+-- Wraps the identifier in double quotes and escapes any embedded double quotes+-- by doubling them, following the SQL standard.+quoteIdentifier :: Text -> Text+quoteIdentifier name = "\"" <> Text.replace "\"" "\"\"" name <> "\""++-- | Convert a query 'Snippet' into a prepared statement that returns dynamic JSON rows.+-- Wraps the query SQL with the CTE that produces @row_to_json@ output.+compiledQueryStatement :: Snippet -> Hasql.Statement () [[Field]]+compiledQueryStatement snippet = Snippet.toPreparableStatement (wrapDynamicQuery snippet) dynamicRowDecoder+{-# INLINE compiledQueryStatement #-}++-- | Encode a UUID as a parameterized SQL 'Snippet'.+uuidParam :: UUID -> Snippet+uuidParam id = Snippet.encoderAndParam (Encoders.nonNullable Encoders.uuid) id+{-# INLINE uuidParam #-}++$(deriveFromJSON defaultOptions ''FunctionCall)+$(deriveFromJSON defaultOptions ''QueryBuilder.OrderByDirection)+$(deriveFromJSON defaultOptions 'SelectAll)+$(deriveFromJSON defaultOptions ''ConditionOperator)+$(deriveFromJSON defaultOptions ''ConditionExpression)++instance FromJSON DynamicSQLQuery where+ parseJSON = withObject "DynamicSQLQuery" $ \v -> DynamicSQLQuery+ <$> v .: "table"+ <*> v .: "selectedColumns"+ <*> v .: "whereCondition"+ <*> v .: "orderByClause"+ <*> v .:? "distinctOnColumn" -- distinctOnColumn can be absent in older versions of ihp-datasync.js+ <*> v .:? "limit" -- Limit can be absent in older versions of ihp-datasync.js+ <*> v .:? "offset" -- Offset can be absent in older versions of ihp-datasync.js+++instance FromJSON OrderByClause where+ parseJSON = withObject "OrderByClause" $ \v -> do+ let oldFormat = OrderByClause+ <$> v .: "orderByColumn"+ <*> v .: "orderByDirection"+ let tagged = do+ tag <- v .: "tag"+ case tag of+ "OrderByClause" -> oldFormat+ "OrderByTSRank" -> OrderByTSRank <$> v .: "tsvector" <*> v .: "tsquery"+ otherwise -> error ("Invalid tag: " <> otherwise)+ tagged <|> oldFormat
+ IHP/DataSync/DynamicQueryCompiler.hs view
@@ -0,0 +1,234 @@+{-|+Module: IHP.DataSync.DynamicQueryCompiler+Description: Compiles a DynamicQuery to SQL+Copyright: (c) digitally induced GmbH, 2021+-}+module IHP.DataSync.DynamicQueryCompiler where++import IHP.Prelude+import IHP.DataSync.DynamicQuery+import IHP.DataSync.TypedEncoder (typedValueParam)+import qualified IHP.QueryBuilder as QueryBuilder+import qualified Hasql.Encoders as Encoders+import qualified Data.List as List+import qualified Data.HashMap.Strict as HashMap+import qualified Data.Aeson as Aeson+import qualified Hasql.DynamicStatements.Snippet as Snippet+import Hasql.DynamicStatements.Snippet (Snippet)++data Renamer = Renamer+ { fieldToColumn :: Text -> Text+ , columnToField :: Text -> Text+ }++-- | Compile a 'DynamicSQLQuery' to a 'Snippet' with typed parameter encoding.+--+-- Column types must be provided via 'ColumnTypeInfo' (from 'makeCachedColumnTypeLookup').+-- Missing column types in WHERE conditions will error at runtime.+--+-- This function:+-- 1. Converts field names from camelCase to snake_case for the query+-- 2. Generates SQL column aliases so results come back with camelCase names+-- 3. For 'SelectAll', expands to all columns from 'ColumnTypeInfo' in database schema order+compileQueryTyped :: Renamer -> ColumnTypeInfo -> DynamicSQLQuery -> Snippet+compileQueryTyped renamer columnInfo query =+ compileQueryMappedTyped renamer columnInfo (mapColumnNames renamer.fieldToColumn query)++-- | Default renamer used by DataSync.+--+-- Transforms JS inputs in @camelCase@ to snake_case for the database+-- and DB outputs in @snake_case@ back to @camelCase@+camelCaseRenamer :: Renamer+camelCaseRenamer =+ Renamer+ { fieldToColumn = fieldNameToColumnName+ , columnToField = columnNameToFieldName+ }++-- | When a Field is retrieved from the database, it's all in @snake_case@. This turns it into @camelCase@+renameField :: Renamer -> Field -> Field+renameField renamer field =+ field { fieldName = renamer.columnToField field.fieldName }++compileQueryMappedTyped :: Renamer -> ColumnTypeInfo -> DynamicSQLQuery -> Snippet+compileQueryMappedTyped renamer columnInfo DynamicSQLQuery { .. } =+ selectPart <> wherePart <> orderByPart <> limitPart <> offsetPart+ where+ selectPart = Snippet.sql ("SELECT"+ <> distinctOnText+ <> compileSelectedColumns renamer columnInfo selectedColumns+ <> " FROM "+ <> quoteIdentifier table)++ distinctOnText = case distinctOnColumn of+ Just column -> " DISTINCT ON (" <> quoteIdentifier (cs column) <> ") "+ Nothing -> " "++ wherePart = case whereCondition of+ Just condition -> Snippet.sql " WHERE " <> compileConditionTyped columnInfo.typeMap condition+ Nothing -> mempty++ orderByPart = case orderByClause of+ [] -> mempty+ clauses -> Snippet.sql " ORDER BY " <> mconcat (List.intersperse (Snippet.sql ", ") (map compileOrderByClauseText clauses))++ limitPart = case limit of+ Just l -> Snippet.sql " LIMIT " <> Snippet.encoderAndParam (Encoders.nonNullable Encoders.int4) (fromIntegral l :: Int32)+ Nothing -> mempty++ offsetPart = case offset of+ Just o -> Snippet.sql " OFFSET " <> Snippet.encoderAndParam (Encoders.nonNullable Encoders.int4) (fromIntegral o :: Int32)+ Nothing -> mempty++-- | Used to transform column names from @camelCase@ to @snake_case@+mapColumnNames :: (Text -> Text) -> DynamicSQLQuery -> DynamicSQLQuery+mapColumnNames rename query =+ query+ { selectedColumns = mapSelectedColumns query.selectedColumns+ , whereCondition = mapConditionExpression <$> query.whereCondition+ , orderByClause = map mapOrderByClause query.orderByClause+ , distinctOnColumn = (cs . rename . cs) <$> query.distinctOnColumn+ }+ where+ mapSelectedColumns :: SelectedColumns -> SelectedColumns+ mapSelectedColumns SelectAll = SelectAll+ mapSelectedColumns (SelectSpecific columns) = SelectSpecific (map rename columns)++ mapConditionExpression :: ConditionExpression -> ConditionExpression+ mapConditionExpression ColumnExpression { field } = ColumnExpression { field = rename field }+ mapConditionExpression InfixOperatorExpression { left, op, right } = InfixOperatorExpression { left = mapConditionExpression left, op, right = mapConditionExpression right }+ mapConditionExpression otherwise = otherwise++ mapOrderByClause :: OrderByClause -> OrderByClause+ mapOrderByClause OrderByClause { orderByColumn, orderByDirection } = OrderByClause { orderByColumn = cs (rename (cs orderByColumn)), orderByDirection }+ mapOrderByClause otherwise = otherwise++compileOrderByClauseText :: OrderByClause -> Snippet+compileOrderByClauseText OrderByClause { orderByColumn, orderByDirection } =+ Snippet.sql (quoteIdentifier (cs orderByColumn)+ <> if orderByDirection == QueryBuilder.Desc then " DESC" else "")+compileOrderByClauseText OrderByTSRank { tsvector, tsquery } =+ Snippet.sql ("ts_rank(" <> quoteIdentifier tsvector <> ", to_tsquery('english', ")+ <> Snippet.encoderAndParam (Encoders.nonNullable Encoders.text) tsquery+ <> Snippet.sql "))"++-- | Compile selected columns, generating SQL aliases when the camelCase field name+-- differs from the snake_case column name.+--+-- For example, if the column is @user_id@ but the client expects @userId@, this will+-- generate @"user_id" AS "userId"@ instead of just @"user_id"@.+--+-- For 'SelectAll', expands to all columns from 'ColumnTypeInfo' in the order they+-- were defined in the database schema (from @pg_attribute.attnum@).+compileSelectedColumns :: Renamer -> ColumnTypeInfo -> SelectedColumns -> Text+compileSelectedColumns renamer columnInfo SelectAll =+ compileSelectedColumns renamer columnInfo (SelectSpecific columnInfo.orderedColumns)+compileSelectedColumns renamer _columnInfo (SelectSpecific columns) =+ mconcat (List.intersperse ", " (map compileColumn columns))+ where+ compileColumn col =+ let alias = renamer.columnToField col+ in if alias == col+ then quoteIdentifier col+ else quoteIdentifier col <> " AS " <> quoteIdentifier alias++-- | Compile a SQL @RETURNING@ clause with aliased columns.+--+-- For INSERT/UPDATE/DELETE statements that return data, this generates+-- @RETURNING "col_a" AS "colA", "col_b" AS "colB", ...@ with proper camelCase aliases.+compileReturningClause :: Renamer -> ColumnTypeInfo -> Text+compileReturningClause renamer columnInfo =+ " RETURNING " <> compileSelectedColumns renamer columnInfo SelectAll++-- | Build an INSERT statement with RETURNING clause.+compileInsert :: Text -> [Text] -> [Snippet] -> Renamer -> ColumnTypeInfo -> Snippet+compileInsert table columns valueSnippets renamer columnTypes =+ Snippet.sql ("INSERT INTO " <> quoteIdentifier table+ <> " (" <> columnSql <> ") VALUES (")+ <> valueSql+ <> Snippet.sql (")" <> compileReturningClause renamer columnTypes)+ where+ columnSql = mconcat $ List.intersperse ", " (map quoteIdentifier columns)+ valueSql = mconcat $ List.intersperse (Snippet.sql ", ") valueSnippets++-- | Build an INSERT statement for multiple rows with RETURNING clause.+compileInsertMany :: Text -> [Text] -> [[Snippet]] -> Renamer -> ColumnTypeInfo -> Snippet+compileInsertMany table columns valueRows renamer columnTypes =+ Snippet.sql ("INSERT INTO " <> quoteIdentifier table+ <> " (" <> columnSql <> ") VALUES ")+ <> valuesSnippet+ <> Snippet.sql (compileReturningClause renamer columnTypes)+ where+ columnSql = mconcat $ List.intersperse ", " (map quoteIdentifier columns)+ valueRowSnippets = map (\row -> Snippet.sql "(" <> mconcat (List.intersperse (Snippet.sql ", ") row) <> Snippet.sql ")") valueRows+ valuesSnippet = mconcat $ List.intersperse (Snippet.sql ", ") valueRowSnippets++-- | Build an UPDATE statement with RETURNING clause.+compileUpdate :: Text -> Snippet -> Snippet -> Renamer -> ColumnTypeInfo -> Snippet+compileUpdate table setSql whereSql renamer columnTypes =+ Snippet.sql ("UPDATE " <> quoteIdentifier table <> " SET ")+ <> setSql+ <> Snippet.sql " WHERE "+ <> whereSql+ <> Snippet.sql (compileReturningClause renamer columnTypes)++-- | Compile a condition expression to a 'Snippet' with typed parameter encoding.+compileConditionTyped :: ColumnTypeMap -> ConditionExpression -> Snippet+compileConditionTyped _ (ColumnExpression column) = Snippet.sql (quoteIdentifier column)+compileConditionTyped types (InfixOperatorExpression a OpEqual (LiteralExpression Aeson.Null)) = compileConditionTyped types (InfixOperatorExpression a OpIs (LiteralExpression Aeson.Null))+compileConditionTyped types (InfixOperatorExpression a OpNotEqual (LiteralExpression Aeson.Null)) = compileConditionTyped types (InfixOperatorExpression a OpIsNot (LiteralExpression Aeson.Null))+compileConditionTyped _types (InfixOperatorExpression _a OpIn (ListExpression { values = [] })) = Snippet.sql "FALSE"+compileConditionTyped types (InfixOperatorExpression a OpIn (ListExpression { values })) | (Aeson.Null `List.elem` values) =+ let condition =+ case partition ((/=) Aeson.Null) values of+ ([], _nullValues) -> InfixOperatorExpression a OpIs (LiteralExpression Aeson.Null)+ (nonNullValues, _nullValues) -> InfixOperatorExpression (InfixOperatorExpression a OpIn (ListExpression { values = nonNullValues })) OpOr (InfixOperatorExpression a OpIs (LiteralExpression Aeson.Null))+ in+ compileConditionTyped types condition+-- When comparing a column to a literal or list, look up the column's type for typed encoding.+compileConditionTyped types (InfixOperatorExpression (ColumnExpression col) operator (LiteralExpression literal)) =+ let opText = compileOperator operator+ rightSnippet = case literal of+ Aeson.Null -> Snippet.sql "NULL"+ _ -> let colType = HashMap.lookup col types+ in Snippet.sql "(" <> typedValueParam colType literal <> Snippet.sql ")"+ in Snippet.sql ("(" <> quoteIdentifier col <> ") " <> opText <> " ") <> rightSnippet+compileConditionTyped types (InfixOperatorExpression (ColumnExpression col) operator (ListExpression { values })) =+ let opText = compileOperator operator+ colType = HashMap.lookup col types+ valSnippets = map (typedValueParam colType) values+ in Snippet.sql ("(" <> quoteIdentifier col <> ") " <> opText <> " (") <> mconcat (List.intersperse (Snippet.sql ", ") valSnippets) <> Snippet.sql ")"+compileConditionTyped types (InfixOperatorExpression a operator b) =+ let aSnippet = compileConditionTyped types a+ opText = compileOperator operator+ bSnippet = compileConditionTyped types b+ rightSnippet = if rightParentheses+ then Snippet.sql "(" <> bSnippet <> Snippet.sql ")"+ else bSnippet++ rightParentheses :: Bool+ rightParentheses =+ case b of+ LiteralExpression Aeson.Null -> False+ ListExpression {} -> False+ _ -> True+ in Snippet.sql "(" <> aSnippet <> Snippet.sql (") " <> opText <> " ") <> rightSnippet+compileConditionTyped _types (LiteralExpression literal) = dynamicValueParam literal+compileConditionTyped _ (CallExpression { functionCall = ToTSQuery { text } }) =+ Snippet.sql "to_tsquery('english', " <> Snippet.encoderAndParam (Encoders.nonNullable Encoders.text) text <> Snippet.sql ")"+compileConditionTyped _types (ListExpression { values }) =+ Snippet.sql "(" <> mconcat (List.intersperse (Snippet.sql ", ") (map dynamicValueParam values)) <> Snippet.sql ")"++compileOperator :: ConditionOperator -> Text+compileOperator OpEqual = "="+compileOperator OpGreaterThan = ">"+compileOperator OpLessThan = "<"+compileOperator OpGreaterThanOrEqual = ">="+compileOperator OpLessThanOrEqual = "<="+compileOperator OpNotEqual = "<>"+compileOperator OpAnd = "AND"+compileOperator OpOr = "OR"+compileOperator OpIs = "IS"+compileOperator OpIsNot = "IS NOT"+compileOperator OpTSMatch = "@@"+compileOperator OpIn = "IN"
+ IHP/DataSync/Hasql.hs view
@@ -0,0 +1,43 @@+module IHP.DataSync.Hasql+( runSession+, runSessionOnConnection+, withDedicatedConnection+) where++import IHP.Prelude+import qualified Control.Exception.Safe as Exception+import qualified Hasql.Pool+import qualified Hasql.Connection as Hasql+import qualified Hasql.Connection.Settings as HasqlSettings+import qualified Hasql.Session as Session+import qualified Hasql.Errors as Hasql++-- | Run a composed Session against the pool. Throws 'Hasql.Pool.UsageError' on failure.+runSession :: Hasql.Pool.Pool -> Session.Session a -> IO a+runSession pool session = do+ result <- Hasql.Pool.use pool session+ case result of+ Left err -> Exception.throwIO err+ Right val -> pure val+{-# INLINE runSession #-}++-- | Run a composed Session on a bare connection. Throws on failure.+runSessionOnConnection :: Hasql.Connection -> Session.Session a -> IO a+runSessionOnConnection conn session = do+ result <- Hasql.use conn session+ case result of+ Left err -> error (Hasql.toDetailedText err)+ Right val -> pure val+{-# INLINE runSessionOnConnection #-}++-- | Acquire a dedicated connection for the duration of an IO action.+--+-- Used for long-lived transactions in DataSync where a connection must be held+-- open across multiple message handlers. The connection is released when the+-- action completes (normally or via exception).+withDedicatedConnection :: ByteString -> (Hasql.Connection -> IO a) -> IO a+withDedicatedConnection databaseUrl action = do+ connResult <- Hasql.acquire (HasqlSettings.connectionString (cs databaseUrl))+ case connResult of+ Right conn -> action conn `Exception.finally` Hasql.release conn+ Left err -> error (Hasql.toDetailedText err)
+ IHP/DataSync/Pool.hs view
@@ -0,0 +1,32 @@+module IHP.DataSync.Pool+( initHasqlPool+, initHasqlPoolWithRLS+) where++import IHP.Prelude+import IHP.FrameworkConfig (findOptionOrNothing, addInitializer)+import IHP.FrameworkConfig.Types (RLSAuthenticatedRole(..))+import qualified Control.Monad.Trans.State.Strict as State+import qualified Data.TMap as TMap+import qualified IHP.DataSync.Role as Role++-- | No-op for backwards compatibility. The hasql pool is now always created+-- as part of the ModelContext. You can remove this call from your Config.hs.+initHasqlPool :: State.StateT TMap.TMap IO ()+initHasqlPool = pure ()++-- | Ensures the RLS authenticated role exists in postgres at startup.+-- Reads 'RLSAuthenticatedRole' from the config (default: @ihp_authenticated@).+--+-- Use this in your @Config.hs@:+--+-- > config :: ConfigBuilder+-- > config = do+-- > initHasqlPoolWithRLS+initHasqlPoolWithRLS :: State.StateT TMap.TMap IO ()+initHasqlPoolWithRLS = do+ rlsRole <- findOptionOrNothing @RLSAuthenticatedRole >>= \case+ Just (RLSAuthenticatedRole role) -> pure role+ Nothing -> pure "ihp_authenticated"+ addInitializer do+ Role.ensureAuthenticatedRoleExistsWithRole ?modelContext.hasqlPool rlsRole
+ IHP/DataSync/REST/Controller.hs view
@@ -0,0 +1,219 @@+{-# LANGUAGE UndecidableInstances #-}+module IHP.DataSync.REST.Controller where++import IHP.ControllerPrelude hiding (OrderByClause)+import IHP.DataSync.REST.Types+import Data.Aeson+import qualified Data.Vector as Vector+import qualified Data.ByteString.Char8 as ByteString+import qualified Control.Exception.Safe as Exception+import IHP.DataSync.RowLevelSecurity+import IHP.DataSync.DynamicQuery+import IHP.DataSync.Types+import Network.HTTP.Types (status400)+import IHP.DataSync.DynamicQueryCompiler+import IHP.DataSync.TypedEncoder (makeCachedColumnTypeLookup, typedAesonValueToSnippet, lookupColumnType)+import qualified Hasql.DynamicStatements.Snippet as Snippet+import Hasql.DynamicStatements.Snippet (Snippet)+import qualified Data.Text as Text+import qualified Data.List as List++import qualified Data.ByteString.Builder as ByteString+import qualified Data.Aeson.Encoding.Internal as Aeson+import qualified Data.Aeson.KeyMap as Aeson+import qualified Data.Aeson.Key as Aeson+import qualified Hasql.Decoders as Decoders++instance (+ Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Controller ApiController where+ action CreateRecordAction { table } = do+ let hasqlPool = ?modelContext.hasqlPool+ ensureRLSEnabled hasqlPool table++ columnTypeLookup <- makeCachedColumnTypeLookup hasqlPool+ columnTypes <- columnTypeLookup table++ payload <- requestBodyJSON++ case payload of+ Object hashMap -> do+ let pairsList = hashMap+ |> Aeson.toList+ |> map (\(key, val) ->+ let col = fieldNameToColumnName (Aeson.toText key)+ in (col, lookupColumnType columnTypes col, val)+ )++ let columns = map (\(c,_,_) -> c) pairsList+ let valueSnippets = map (\(_, colType, val) -> typedAesonValueToSnippet colType val) pairsList++ let insertResult = compileInsert table columns valueSnippets camelCaseRenamer columnTypes+ let stmt = compiledQueryStatement insertResult++ result :: Either SomeException [[Field]] <- Exception.try do+ sqlQueryWriteWithRLS hasqlPool stmt++ case result of+ Left e -> renderErrorJson (show e :: Text)+ Right result -> renderJson result++ Array objects -> do+ let objectList = Vector.toList objects+ case objectList of+ [] -> renderErrorJson ("At least one record is required" :: Text)+ (firstElement:_) -> case firstElement of+ Object firstHashMap -> do+ let columns = firstHashMap+ |> Aeson.keys+ |> map (fieldNameToColumnName . Aeson.toText)++ let parseObject value = case value of+ Object hashMap -> Right hashMap+ _otherwise -> Left ("Expected object, got: " <> show value)++ case mapM parseObject objectList of+ Left err -> renderErrorJson (cs err :: Text)+ Right hashMaps -> do+ let encodeRow hashMap = map+ (\col ->+ let fieldName = columnNameToFieldName col+ val = fromMaybe Data.Aeson.Null (Aeson.lookup (Aeson.fromText fieldName) hashMap)+ in typedAesonValueToSnippet (lookupColumnType columnTypes col) val)+ columns+ let valueRows = map encodeRow hashMaps++ let insertResult = compileInsertMany table columns valueRows camelCaseRenamer columnTypes+ let stmt = compiledQueryStatement insertResult++ result :: [[Field]] <- sqlQueryWriteWithRLS hasqlPool stmt+ renderJson result+ _otherwise -> renderErrorJson ("Expected object" :: Text)++ _ -> error "Expected JSON object or array"++ action UpdateRecordAction { table, id } = do+ let hasqlPool = ?modelContext.hasqlPool+ ensureRLSEnabled hasqlPool table++ columnTypeLookup <- makeCachedColumnTypeLookup hasqlPool+ columnTypes <- columnTypeLookup table++ payload <- requestBodyJSON+ let hashMap = case payload of+ Object hm -> hm+ _ -> error "Expected JSON object"++ let setSql = encodeKeyMapToSetSql columnTypes hashMap+ let updateResult = compileUpdate table setSql (Snippet.sql "id = " <> uuidParam id) camelCaseRenamer columnTypes+ let stmt = compiledQueryStatement updateResult++ result :: [[Field]] <- sqlQueryWriteWithRLS hasqlPool stmt++ renderJson (head result)++ -- DELETE /api/:table/:id+ action DeleteRecordAction { table, id } = do+ let hasqlPool = ?modelContext.hasqlPool+ ensureRLSEnabled hasqlPool table++ let deleteSnippet = Snippet.sql ("DELETE FROM " <> quoteIdentifier table <> " WHERE id = ") <> uuidParam id+ let stmt = Snippet.toPreparableStatement deleteSnippet Decoders.noResult+ sqlExecWithRLS hasqlPool stmt++ renderJson True++ -- GET /api/:table/:id+ action ShowRecordAction { table, id } = do+ let hasqlPool = ?modelContext.hasqlPool+ ensureRLSEnabled hasqlPool table++ columnTypeLookup <- makeCachedColumnTypeLookup hasqlPool+ columnTypes <- columnTypeLookup table+ let selectColumns = compileSelectedColumns camelCaseRenamer columnTypes SelectAll+ let selectSnippet = Snippet.sql ("SELECT " <> selectColumns <> " FROM " <> quoteIdentifier table <> " WHERE id = ") <> uuidParam id+ let stmt = Snippet.toPreparableStatement (wrapDynamicQuery selectSnippet) dynamicRowDecoder+ result :: [[Field]] <- sqlQueryWithRLS hasqlPool stmt++ renderJson (head result)++ -- GET /api/:table+ -- GET /api/:table?orderBy=createdAt+ -- GET /api/:table?fields=id,title+ action ListRecordsAction { table } = do+ let hasqlPool = ?modelContext.hasqlPool+ ensureRLSEnabled hasqlPool table++ columnTypeLookup <- makeCachedColumnTypeLookup hasqlPool+ columnTypes <- columnTypeLookup table+ let querySnippet = compileQueryTyped camelCaseRenamer columnTypes (buildDynamicQueryFromRequest table)+ let stmt = compiledQueryStatement querySnippet+ result :: [[Field]] <- sqlQueryWithRLS hasqlPool stmt++ renderJson result++ action GraphQLQueryAction = do+ error "GraphQLQueryAction is handled by the GraphQL middleware"++buildDynamicQueryFromRequest table = DynamicSQLQuery+ { table+ , selectedColumns = paramOrDefault SelectAll "fields"+ , whereCondition = Nothing+ , orderByClause = paramList "orderBy"+ , distinctOnColumn = paramOrNothing "distinctOnColumn"+ , limit = paramOrNothing "limit"+ , offset = paramOrNothing "offset"+ }++instance ParamReader SelectedColumns where+ readParameter byteString = pure $+ byteString+ |> cs+ |> Text.split (\char -> char == ',')+ |> SelectSpecific++instance ParamReader OrderByClause where+ readParameter byteString = case ByteString.split ',' byteString of+ [orderByColumn, order] -> do+ orderByDirection <- parseOrder order+ pure OrderByClause { orderByColumn, orderByDirection }+ [orderByColumn] -> pure OrderByClause { orderByColumn, orderByDirection = Asc }+ _ -> Left "Invalid order by clause"+ where+ parseOrder "asc" = Right Asc+ parseOrder "desc" = Right Desc+ parseOrder otherwise = Left ("Invalid order " <> cs otherwise)++-- | Encode an Aeson KeyMap (from REST JSON payload) into a SQL SET clause 'Snippet'.+encodeKeyMapToSetSql :: ColumnTypeInfo -> Aeson.KeyMap Value -> Snippet+encodeKeyMapToSetSql columnTypes hashMap =+ let pairsList = hashMap+ |> Aeson.toList+ |> map (\(key, val) ->+ let col = fieldNameToColumnName (Aeson.toText key)+ in (col, lookupColumnType columnTypes col, val)+ )+ encodeSetClause (col, colType, val) =+ Snippet.sql (quoteIdentifier col <> " = ") <> typedAesonValueToSnippet colType val+ setSnippets = map encodeSetClause pairsList+ in mconcat $ List.intersperse (Snippet.sql ", ") setSnippets++renderErrorJson :: (?context :: ControllerContext, ?request :: Request) => ToJSON json => json -> IO ()+renderErrorJson json = renderJsonWithStatusCode status400 json+{-# INLINABLE renderErrorJson #-}++instance ToJSON GraphQLResult where+ toJSON GraphQLResult { requestId, graphQLResult } = object [ "tag" .= ("GraphQLResult" :: Text), "requestId" .= requestId, "graphQLResult" .= ("" :: Text) ]+ toEncoding GraphQLResult { requestId, graphQLResult } = Aeson.econcat+ [ Aeson.unsafeToEncoding "{\"tag\":\"GraphQLResult\",\"requestId\":"+ , Aeson.int requestId+ , Aeson.unsafeToEncoding ",\"graphQLResult\":"+ , toEncoding graphQLResult+ , Aeson.unsafeToEncoding "}"+ ]+instance ToJSON UndecodedJSON where+ toJSON (UndecodedJSON _) = error "Not implemented"+ toEncoding (UndecodedJSON json) = Aeson.unsafeToEncoding (ByteString.byteString json)
+ IHP/DataSync/REST/Routes.hs view
@@ -0,0 +1,54 @@+module IHP.DataSync.REST.Routes where++import IHP.RouterPrelude+import IHP.DataSync.REST.Types++instance CanRoute ApiController where+ parseRoute' = do+ string "/api/"++ let+ graphQLQueryAction = do+ string "graphql"+ endOfInput+ onlyAllowMethods [POST]+ pure GraphQLQueryAction++ createRecordAction table = do+ endOfInput+ onlyAllowMethods [POST]+ pure CreateRecordAction { table }++ updateOrDeleteRecordAction table = do+ string "/"+ id <- parseUUID++ endOfInput++ method <- getMethod+ case method of+ PATCH -> pure UpdateRecordAction { table, id }+ GET -> pure ShowRecordAction { table, id }+ DELETE -> pure DeleteRecordAction { table, id }+ _ -> error "updateOrDeleteRecordAction: unsupported method"++ listRecordsAction table = do+ endOfInput+ method <- getMethod+ case method of+ GET -> pure ListRecordsAction { table }+ _ -> error "listRecordsAction: unsupported method"++ crud = do+ table <- parseText+ updateOrDeleteRecordAction table <|> createRecordAction table <|> listRecordsAction table++ graphQLQueryAction <|> crud++instance HasPath ApiController where+ pathTo CreateRecordAction { table } = "/api/" <> table+ pathTo UpdateRecordAction { table, id } = "/api/" <> table <> "/" <> tshow id+ pathTo DeleteRecordAction { table, id } = "/api/" <> table <> "/" <> tshow id+ pathTo ShowRecordAction { table, id } = "/api/" <> table <> "/" <> tshow id+ pathTo ListRecordsAction { table } = "/api/" <> table+ pathTo GraphQLQueryAction = "/api/graphql"
+ IHP/DataSync/REST/Types.hs view
@@ -0,0 +1,12 @@+module IHP.DataSync.REST.Types where++import IHP.Prelude++data ApiController+ = CreateRecordAction { table :: Text } -- ^ POST /api/books+ | UpdateRecordAction { table :: Text, id :: UUID }+ | DeleteRecordAction { table :: Text, id :: UUID }+ | ShowRecordAction { table :: Text, id :: UUID } -- ^ GET /api/books/9ba0ffbc-bfc1-4d7d-8152-30a6648806f7+ | ListRecordsAction { table :: Text } -- ^ GET /api/books+ | GraphQLQueryAction+ deriving (Eq, Show, Data)
+ IHP/DataSync/Role.hs view
@@ -0,0 +1,99 @@+{-|+Module: IHP.DataSync.Role+Description: Postgres role management for RLS+Copyright: (c) digitally induced GmbH, 2021++The default user that creates a table in postgres always+has access to all rows inside the table. The default user is not restricted+to the RLS policies.++Therefore we need to use a second role whenever we want to+make a query with RLS enabled. Basically for every query we do, we'll+wrap it in a transaction and then use 'SET LOCAL ROLE ..' to switch to+our second role for the duration of the transaction.++-}+module IHP.DataSync.Role where++import IHP.Prelude+import IHP.FrameworkConfig+import qualified Data.Text as Text+import qualified Hasql.Pool+import qualified Hasql.Decoders as Decoders+import qualified Hasql.Encoders as Encoders+import qualified Hasql.Statement as Statement+import qualified Hasql.Session as Session+import IHP.DataSync.Hasql (runSession)++-- Statements++doesRoleExistsStatement :: Statement.Statement Text Bool+doesRoleExistsStatement = Statement.preparable+ "SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname = $1 LIMIT 1)"+ (Encoders.param (Encoders.nonNullable Encoders.text))+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))++-- Sessions++ensureAuthenticatedRoleSession :: (?context :: context, ConfigProvider context) => Session.Session ()+ensureAuthenticatedRoleSession = do+ let role = authenticatedRole+ roleExists <- Session.statement role doesRoleExistsStatement+ unless roleExists (createAuthenticatedRoleSession role)+ grantPermissionsSession role++createAuthenticatedRoleSession :: Text -> Session.Session ()+createAuthenticatedRoleSession role = do+ -- The role is only going to be used from 'SET ROLE ..' calls+ -- Therefore we can disallow direct connection with NOLOGIN+ Session.statement () (Statement.unpreparable+ ("CREATE ROLE " <> quoteIdentifier role <> " NOLOGIN")+ Encoders.noParams+ Decoders.noResult)++grantPermissionsSession :: Text -> Session.Session ()+grantPermissionsSession role = do+ -- From SO https://stackoverflow.com/a/17355059/14144232+ --+ -- GRANTs on different objects are separate. GRANTing on a database doesn't GRANT rights to the schema within. Similiarly, GRANTing on a schema doesn't grant rights on the tables within.+ --+ -- If you have rights to SELECT from a table, but not the right to see it in the schema that contains it then you can't access the table.+ --+ -- The rights tests are done in order:+ --+ -- Do you have `USAGE` on the schema?+ -- No: Reject access.+ -- Yes: Do you also have the appropriate rights on the table?+ -- No: Reject access.+ -- Yes: Check column privileges.++ let exec sql = Session.statement () (Statement.unpreparable sql Encoders.noParams Decoders.noResult)++ -- The role should have access to all existing tables in our schema+ exec ("GRANT USAGE ON SCHEMA public TO " <> quoteIdentifier role)++ -- The role should have access to all existing tables in our schema+ exec ("GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO " <> quoteIdentifier role)++ -- Also grant access to all tables created in the future+ exec ("ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL PRIVILEGES ON TABLES TO " <> quoteIdentifier role)++ensureAuthenticatedRoleExists :: (?context :: context, ConfigProvider context) => Hasql.Pool.Pool -> IO ()+ensureAuthenticatedRoleExists pool = runSession pool ensureAuthenticatedRoleSession++-- | Like 'ensureAuthenticatedRoleExists', but takes the role name directly+-- instead of reading it from the framework config. Used by 'initHasqlPoolWithRLS'+-- at config-build time when 'FrameworkConfig' is not yet available.+ensureAuthenticatedRoleExistsWithRole :: Hasql.Pool.Pool -> Text -> IO ()+ensureAuthenticatedRoleExistsWithRole pool role = runSession pool $ do+ roleExists <- Session.statement role doesRoleExistsStatement+ unless roleExists (createAuthenticatedRoleSession role)+ grantPermissionsSession role++authenticatedRole :: (?context :: context, ConfigProvider context) => Text+authenticatedRole = ?context.frameworkConfig.rlsAuthenticatedRole++-- | Quote a SQL identifier (role name, table name, etc.) to prevent SQL injection.+-- Escapes embedded double quotes by doubling them per SQL standard.+quoteIdentifier :: Text -> Text+quoteIdentifier name = "\"" <> Text.replace "\"" "\"\"" name <> "\""
+ IHP/DataSync/RowLevelSecurity.hs view
@@ -0,0 +1,273 @@+module IHP.DataSync.RowLevelSecurity+( ensureRLSEnabled+, hasRLSEnabled+, TableWithRLS (..)+, makeCachedEnsureRLSEnabled+, sqlQueryWithRLS+, sqlQueryWriteWithRLS+, sqlExecWithRLS+, sqlQueryScalarWithRLS+, hasRLSEnabledSession+, ensureRLSEnabledSession+, sqlQueryWithRLSSession+, sqlQueryWriteWithRLSSession+, sqlExecWithRLSSession+, sqlQueryScalarWithRLSSession+, setRLSConfigStatement+, setRLSConfigSession+, rlsPolicyColumns+, makeCachedRLSPolicyColumns+)+where++import IHP.ControllerPrelude hiding (sqlQuery, sqlExec, sqlQueryScalar)+import qualified Hasql.Pool+import qualified Hasql.Decoders as Decoders+import qualified Hasql.Encoders as Encoders+import qualified Hasql.Statement as Statement+import qualified Hasql.Session as Session+import qualified Hasql.Transaction as Tx+import qualified Hasql.Transaction.Sessions as Tx+import qualified IHP.DataSync.Role as Role+import qualified Data.Set as Set+import qualified Data.HashMap.Strict as HashMap+import IHP.DataSync.Hasql (runSession)++-- Statements++hasRLSEnabledStatement :: Statement.Statement Text Bool+hasRLSEnabledStatement = Statement.preparable+ "SELECT relrowsecurity FROM pg_class WHERE oid = quote_ident($1)::regclass"+ (Encoders.param (Encoders.nonNullable Encoders.text))+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))++-- Sessions++hasRLSEnabledSession :: Text -> Session.Session Bool+hasRLSEnabledSession table = Session.statement table hasRLSEnabledStatement++ensureRLSEnabledSession :: Text -> Session.Session TableWithRLS+ensureRLSEnabledSession table = do+ rlsEnabled <- hasRLSEnabledSession table+ unless rlsEnabled (error "Row level security is required for accessing this table")+ pure (TableWithRLS table)++-- | Set RLS config (role and user id) in the current transaction.+--+-- This is a Session-level action for use in user-managed transactions+-- (e.g. after a manual @BEGIN@).+setRLSConfigSession ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Session.Session ()+setRLSConfigSession = Session.statement (Role.authenticatedRole, encodedUserId) setRLSConfigStatement+ where+ encodedUserId = case (.id) <$> currentUserOrNothing of+ Just userId -> tshow userId+ Nothing -> ""++sqlQueryWithRLSSession ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Statement.Statement () [result] -> Session.Session [result]+sqlQueryWithRLSSession statement =+ Tx.transaction Tx.ReadCommitted Tx.Read $ do+ Tx.statement (Role.authenticatedRole, encodedUserId) setRLSConfigStatement+ Tx.statement () statement+ where+ encodedUserId = case (.id) <$> currentUserOrNothing of+ Just userId -> tshow userId+ Nothing -> ""+{-# INLINE sqlQueryWithRLSSession #-}++-- | Like 'sqlQueryWithRLSSession', but uses a write transaction.+--+-- Use this for INSERT, UPDATE, or DELETE statements with RETURNING that need+-- to return results (e.g. wrapped with 'wrapDynamicQuery').+sqlQueryWriteWithRLSSession ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Statement.Statement () [result] -> Session.Session [result]+sqlQueryWriteWithRLSSession statement =+ Tx.transaction Tx.ReadCommitted Tx.Write $ do+ Tx.statement (Role.authenticatedRole, encodedUserId) setRLSConfigStatement+ Tx.statement () statement+ where+ encodedUserId = case (.id) <$> currentUserOrNothing of+ Just userId -> tshow userId+ Nothing -> ""+{-# INLINE sqlQueryWriteWithRLSSession #-}++sqlExecWithRLSSession ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Statement.Statement () () -> Session.Session ()+sqlExecWithRLSSession statement =+ Tx.transaction Tx.ReadCommitted Tx.Write $ do+ Tx.statement (Role.authenticatedRole, encodedUserId) setRLSConfigStatement+ Tx.statement () statement+ where+ encodedUserId = case (.id) <$> currentUserOrNothing of+ Just userId -> tshow userId+ Nothing -> ""+{-# INLINE sqlExecWithRLSSession #-}++sqlQueryScalarWithRLSSession ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Statement.Statement () result -> Session.Session result+sqlQueryScalarWithRLSSession statement =+ Tx.transaction Tx.ReadCommitted Tx.Read $ do+ Tx.statement (Role.authenticatedRole, encodedUserId) setRLSConfigStatement+ Tx.statement () statement+ where+ encodedUserId = case (.id) <$> currentUserOrNothing of+ Just userId -> tshow userId+ Nothing -> ""+{-# INLINE sqlQueryScalarWithRLSSession #-}++-- IO API (thin wrappers)++sqlQueryWithRLS ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Hasql.Pool.Pool -> Statement.Statement () [result] -> IO [result]+sqlQueryWithRLS pool statement = runSession pool (sqlQueryWithRLSSession statement)+{-# INLINE sqlQueryWithRLS #-}++-- | Like 'sqlQueryWithRLS', but uses a write transaction.+--+-- Use this for INSERT, UPDATE, or DELETE statements with RETURNING that need+-- to return results (e.g. wrapped with 'wrapDynamicQuery').+sqlQueryWriteWithRLS ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Hasql.Pool.Pool -> Statement.Statement () [result] -> IO [result]+sqlQueryWriteWithRLS pool statement = runSession pool (sqlQueryWriteWithRLSSession statement)+{-# INLINE sqlQueryWriteWithRLS #-}++sqlExecWithRLS ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Hasql.Pool.Pool -> Statement.Statement () () -> IO ()+sqlExecWithRLS pool statement = runSession pool (sqlExecWithRLSSession statement)+{-# INLINE sqlExecWithRLS #-}++sqlQueryScalarWithRLS ::+ ( ?context :: ControllerContext+ , Show (PrimaryKey (GetTableName CurrentUserRecord))+ , HasNewSessionUrl CurrentUserRecord+ , Typeable CurrentUserRecord+ , HasField "id" CurrentUserRecord (Id' (GetTableName CurrentUserRecord))+ ) => Hasql.Pool.Pool -> Statement.Statement () result -> IO result+sqlQueryScalarWithRLS pool statement = runSession pool (sqlQueryScalarWithRLSSession statement)+{-# INLINE sqlQueryScalarWithRLS #-}++-- | Returns a proof that RLS is enabled for a table+ensureRLSEnabled :: Hasql.Pool.Pool -> Text -> IO TableWithRLS+ensureRLSEnabled pool table = runSession pool (ensureRLSEnabledSession table)++-- | Returns a factory for 'ensureRLSEnabled' that memoizes when a table has RLS enabled.+--+-- When a table doesn't have RLS enabled yet, the result is not memoized.+--+-- __Example:__+--+-- > -- Setup+-- > ensureRLSEnabled <- makeCachedEnsureRLSEnabled hasqlPool+-- >+-- > ensureRLSEnabled "projects" -- Runs a database query to check if row level security is enabled for the projects table+-- >+-- > -- Asuming 'ensureRLSEnabled "projects"' proceeded without errors:+-- >+-- > ensureRLSEnabled "projects" -- Now this will instantly return True and don't fire any SQL queries anymore+--+makeCachedEnsureRLSEnabled :: Hasql.Pool.Pool -> IO (Text -> IO TableWithRLS)+makeCachedEnsureRLSEnabled pool = do+ tables <- newIORef Set.empty+ pure \tableName -> do+ rlsEnabled <- Set.member tableName <$> readIORef tables++ if rlsEnabled+ then pure TableWithRLS { tableName }+ else do+ proof <- ensureRLSEnabled pool tableName+ modifyIORef' tables (Set.insert tableName)+ pure proof++-- | Returns 'True' if row level security has been enabled on a table+--+-- RLS can be enabled with this SQL statement:+--+-- > ALTER TABLE my_table ENABLE ROW LEVEL SECURITY;+--+-- After this 'hasRLSEnabled' will return true:+--+-- >>> hasRLSEnabled pool "my_table"+-- True+hasRLSEnabled :: Hasql.Pool.Pool -> Text -> IO Bool+hasRLSEnabled pool table = runSession pool (hasRLSEnabledSession table)++-- | Can be constructed using 'ensureRLSEnabled'+--+-- > tableWithRLS <- ensureRLSEnabled "my_table"+--+-- Useful to carry a proof that the RLS is actually enabled+newtype TableWithRLS = TableWithRLS { tableName :: Text } deriving (Eq, Ord)++-- | Prepared statement to query which columns a table's RLS policies reference.+--+-- Checks both @USING@ (@polqual@) and @WITH CHECK@ (@polwithcheck@) expressions.+rlsPolicyColumnsStatement :: Statement.Statement Text [Text]+rlsPolicyColumnsStatement = Statement.preparable+ "SELECT DISTINCT a.attname::text FROM pg_policy p JOIN pg_class c ON c.oid = p.polrelid JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum > 0 WHERE c.relname = $1 AND (pg_get_expr(p.polqual, p.polrelid) LIKE '%' || a.attname || '%' OR pg_get_expr(p.polwithcheck, p.polrelid) LIKE '%' || a.attname || '%')"+ (Encoders.param (Encoders.nonNullable Encoders.text))+ (Decoders.rowList (Decoders.column (Decoders.nonNullable Decoders.text)))++-- | Returns the set of column names referenced in a table's RLS policies.+--+-- >>> rlsPolicyColumns pool "messages"+-- fromList ["user_id"]+rlsPolicyColumns :: Hasql.Pool.Pool -> Text -> IO (Set.Set Text)+rlsPolicyColumns pool table = do+ results <- runSession pool (Session.statement table rlsPolicyColumnsStatement)+ pure (Set.fromList results)++-- | Returns a cached version of 'rlsPolicyColumns'.+--+-- Queries once per table, caches forever for the connection lifetime.+makeCachedRLSPolicyColumns :: Hasql.Pool.Pool -> IO (Text -> IO (Set.Set Text))+makeCachedRLSPolicyColumns pool = do+ cache <- newIORef HashMap.empty+ pure \tableName -> do+ cached <- HashMap.lookup tableName <$> readIORef cache+ case cached of+ Just columns -> pure columns+ Nothing -> do+ columns <- rlsPolicyColumns pool tableName+ modifyIORef' cache (HashMap.insert tableName columns)+ pure columns
+ IHP/DataSync/TypedEncoder.hs view
@@ -0,0 +1,251 @@+{-|+Module: IHP.DataSync.TypedEncoder+Description: Schema-aware parameter encoding for DataSync queries+Copyright: (c) digitally induced GmbH, 2025++Queries column types from @pg_attribute@/@pg_type@ at runtime, caches per-table,+and uses typed encoders (e.g. 'Encoders.uuid' for UUID columns) via 'Snippet'.+This avoids type mismatches like @operator does not exist: uuid = text@+that occur in the extended query protocol when sending text-typed parameters for non-text columns.+-}+module IHP.DataSync.TypedEncoder+( ColumnTypeMap+, ColumnTypeInfo(..)+, makeCachedColumnTypeLookup+, typedValueParam+, typedAesonValueToSnippet+, lookupColumnType+) where++import IHP.Prelude+import IHP.DataSync.DynamicQuery (ColumnTypeMap, ColumnTypeInfo(..), quoteIdentifier)+import PostgresqlTypes.Point (Point, fromCoordinates)+import qualified Hasql.Mapping.IsScalar as Mapping+import Hasql.PostgresqlTypes ()+import qualified Data.HashMap.Strict as HashMap+import qualified Hasql.Pool+import qualified Hasql.Session as Session+import qualified Hasql.Statement as Statement+import qualified Hasql.Encoders as Encoders+import qualified Hasql.Decoders as Decoders+import IHP.DataSync.Hasql (runSession)+import Data.Aeson (Value(..))+import qualified Data.Aeson as Aeson+import qualified Data.Aeson.KeyMap as Aeson+import qualified Data.UUID as UUID+import qualified Data.Scientific as Scientific+import Data.Int (Int16)+import qualified Data.Time.Format.ISO8601 as ISO8601+import qualified Data.Attoparsec.Text as Attoparsec+import qualified Data.Text as Text+import qualified Data.List as List+import qualified Data.Vector as Vector+import qualified Hasql.DynamicStatements.Snippet as Snippet+import Hasql.DynamicStatements.Snippet (Snippet)++-- | Creates a cached lookup function that queries column types from @pg_attribute@/@pg_type@+-- and caches the result per table name.+--+-- Returns 'ColumnTypeInfo' which includes both a type map for O(1) lookups and+-- an ordered column list matching the database schema order (from @attnum@).+--+-- Follows the same caching pattern as 'makeCachedEnsureRLSEnabled'.+makeCachedColumnTypeLookup :: Hasql.Pool.Pool -> IO (Text -> IO ColumnTypeInfo)+makeCachedColumnTypeLookup pool = do+ cache <- newIORef HashMap.empty+ pure \tableName -> do+ cached <- HashMap.lookup tableName <$> readIORef cache+ case cached of+ Just types -> pure types+ Nothing -> do+ types <- runSession pool (Session.statement tableName columnTypesStatement)+ modifyIORef' cache (HashMap.insert tableName types)+ pure types++-- | Prepared statement that queries column types for a table.+--+-- Uses @pg_attribute@ joined with @pg_type@ to get clean type names.+-- Filters out dropped columns and system columns (@attnum > 0@).+-- Results are ordered by @attnum@ to preserve the database schema column order.+columnTypesStatement :: Statement.Statement Text ColumnTypeInfo+columnTypesStatement = Statement.preparable+ "SELECT a.attname::text, t.typname::text FROM pg_attribute a JOIN pg_type t ON a.atttypid = t.oid WHERE a.attrelid = quote_ident($1)::regclass AND a.attnum > 0 AND NOT a.attisdropped ORDER BY a.attnum"+ (Encoders.param (Encoders.nonNullable Encoders.text))+ (buildColumnTypeInfo <$> Decoders.rowList columnTypeDecoder)+ where+ buildColumnTypeInfo rows = ColumnTypeInfo+ { typeMap = HashMap.fromList rows+ , orderedColumns = map fst rows+ }+ columnTypeDecoder =+ (,) <$> Decoders.column (Decoders.nonNullable Decoders.text)+ <*> Decoders.column (Decoders.nonNullable Decoders.text)++-- | Encode a value with the given encoder as a 'Snippet'.+encodeValue :: Encoders.NullableOrNot Encoders.Value a -> a -> Snippet+encodeValue enc val = Snippet.encoderAndParam enc val+{-# INLINE encodeValue #-}++-- | Encode an Aeson 'Value' as a typed parameter 'Snippet'.+--+-- When a column type is known (from 'ColumnTypeMap'), uses the correct typed encoder.+-- Errors when no type info is available, since 'makeCachedColumnTypeLookup' should+-- always provide column types.+typedValueParam :: Maybe Text -> Value -> Snippet+typedValueParam _ Aeson.Null = Snippet.sql "NULL"+typedValueParam colType (Object values)+ | colType == Just "point" =+ let+ tryDecodeAsPoint :: Maybe Point+ tryDecodeAsPoint = do+ xValue <- Aeson.lookup "x" values+ yValue <- Aeson.lookup "y" values+ x <- case xValue of+ Aeson.Number number -> pure (Scientific.toRealFloat number)+ _ -> Nothing+ y <- case yValue of+ Aeson.Number number -> pure (Scientific.toRealFloat number)+ _ -> Nothing+ pure (fromCoordinates x y)+ in+ if Aeson.size values == 2+ then case tryDecodeAsPoint of+ Just point -> encodeValue (Encoders.nonNullable Mapping.encoder) point+ Nothing -> error "Cannot decode as Point"+ else error "Cannot decode as Point: expected {x, y} object"+typedValueParam pgType (Array arr) =+ let elemType = case pgType of+ Just t | "_" `Text.isPrefixOf` t -> Just (Text.drop 1 t)+ _ -> pgType+ in Snippet.sql "ARRAY[" <> mconcat (List.intersperse (Snippet.sql ", ") (map (typedValueParam elemType) (Vector.toList arr))) <> Snippet.sql "]"+typedValueParam (Just pgType) value = encodeWithType pgType value+typedValueParam Nothing value = error ("typedValueParam: No column type available for value: " <> show value)++-- | Encode an Aeson 'Value' using a specific PostgreSQL type encoder.+encodeWithType :: Text -> Value -> Snippet+encodeWithType _ Aeson.Null = Snippet.sql "NULL"+encodeWithType "uuid" val = encodeValue (Encoders.nonNullable Encoders.uuid) (toUUID val)+encodeWithType "text" val = encodeValue (Encoders.nonNullable Encoders.text) (toText val)+encodeWithType "varchar" val = encodeValue (Encoders.nonNullable Encoders.text) (toText val)+encodeWithType "bpchar" val = encodeValue (Encoders.nonNullable Encoders.text) (toText val)+encodeWithType "name" val = encodeValue (Encoders.nonNullable Encoders.text) (toText val)+encodeWithType "int4" val = encodeValue (Encoders.nonNullable Encoders.int4) (toInt32 val)+encodeWithType "int8" val = encodeValue (Encoders.nonNullable Encoders.int8) (toInt64 val)+encodeWithType "int2" val = encodeValue (Encoders.nonNullable Encoders.int2) (toInt16 val)+encodeWithType "bool" val = encodeValue (Encoders.nonNullable Encoders.bool) (toBool val)+encodeWithType "timestamptz" val = encodeValue (Encoders.nonNullable Encoders.timestamptz) (toUTCTime val)+encodeWithType "timestamp" val = encodeValue (Encoders.nonNullable Encoders.timestamp) (toLocalTime val)+encodeWithType "date" val = encodeValue (Encoders.nonNullable Encoders.date) (toDay val)+encodeWithType "float8" val = encodeValue (Encoders.nonNullable Encoders.float8) (toDouble val)+encodeWithType "float4" val = encodeValue (Encoders.nonNullable Encoders.float4) (toFloat val)+encodeWithType "numeric" val = encodeValue (Encoders.nonNullable Encoders.numeric) (toScientific val)+encodeWithType "jsonb" val = encodeValue (Encoders.nonNullable Encoders.jsonb) val+encodeWithType "json" val = encodeValue (Encoders.nonNullable Encoders.json) val+encodeWithType "bytea" val = encodeValue (Encoders.nonNullable Encoders.bytea) (toByteString val)+-- Interval uses text+cast for dynamic DataSync queries where values come as JSON text.+encodeWithType "interval" val = encodeValue (Encoders.nonNullable Encoders.text) (toText val) <> Snippet.sql "::interval"+encodeWithType pgType val = encodeValue (Encoders.nonNullable Encoders.text) (toText val) <> Snippet.sql ("::" <> quoteIdentifier pgType)++-- | Encode an Aeson 'Value' as a typed parameter for INSERT/UPDATE operations.+--+-- Delegates directly to 'typedValueParam'.+typedAesonValueToSnippet :: Maybe Text -> Value -> Snippet+typedAesonValueToSnippet = typedValueParam++-- | Look up a column's type from ColumnTypeInfo.+lookupColumnType :: ColumnTypeInfo -> Text -> Maybe Text+lookupColumnType info col = HashMap.lookup col info.typeMap++-- Conversion helpers (from Aeson Value to Haskell types)++toUUID :: Value -> UUID+toUUID (String t) = fromMaybe (error ("Invalid UUID: " <> cs t)) (UUID.fromText t)+toUUID v = error ("Cannot convert to UUID: " <> show v)++toText :: Value -> Text+toText (String t) = t+toText (Number n) = case Scientific.floatingOrInteger n of+ Left (d :: Double) -> tshow d+ Right (i :: Integer) -> tshow i+toText (Bool b) = if b then "true" else "false"+toText v = error ("Cannot convert to Text: " <> show v)++toInt32 :: Value -> Int32+toInt32 (Number n) = case Scientific.floatingOrInteger n of+ Left (d :: Double) -> round d+ Right (i :: Integer) -> fromIntegral i+toInt32 (String t) = case Attoparsec.parseOnly (Attoparsec.signed Attoparsec.decimal) t of+ Right i -> i+ Left _ -> error ("Cannot parse Int32 from text: " <> cs t)+toInt32 v = error ("Cannot convert to Int32: " <> show v)++toInt64 :: Value -> Int64+toInt64 (Number n) = case Scientific.floatingOrInteger n of+ Left (d :: Double) -> round d+ Right (i :: Integer) -> fromIntegral i+toInt64 (String t) = case Attoparsec.parseOnly (Attoparsec.signed Attoparsec.decimal) t of+ Right i -> i+ Left _ -> error ("Cannot parse Int64 from text: " <> cs t)+toInt64 v = error ("Cannot convert to Int64: " <> show v)++toInt16 :: Value -> Int16+toInt16 (Number n) = case Scientific.floatingOrInteger n of+ Left (d :: Double) -> round d+ Right (i :: Integer) -> fromIntegral i+toInt16 (String t) = case Attoparsec.parseOnly (Attoparsec.signed Attoparsec.decimal) t of+ Right i -> i+ Left _ -> error ("Cannot parse Int16 from text: " <> cs t)+toInt16 v = error ("Cannot convert to Int16: " <> show v)++toBool :: Value -> Bool+toBool (Bool b) = b+toBool (String "true") = True+toBool (String "false") = False+toBool (String "t") = True+toBool (String "f") = False+toBool (Number 0) = False+toBool (Number _) = True+toBool v = error ("Cannot convert to Bool: " <> show v)++toUTCTime :: Value -> UTCTime+toUTCTime (String t) = case ISO8601.iso8601ParseM (cs t) of+ Just utc -> utc+ Nothing -> error ("Cannot parse UTCTime from text: " <> cs t)+toUTCTime v = error ("Cannot convert to UTCTime: " <> show v)++toLocalTime :: Value -> LocalTime+toLocalTime (String t) = case ISO8601.iso8601ParseM (cs t) of+ Just lt -> lt+ Nothing -> error ("Cannot parse LocalTime from text: " <> cs t)+toLocalTime v = error ("Cannot convert to LocalTime: " <> show v)++toDay :: Value -> Day+toDay (String t) = case ISO8601.iso8601ParseM (cs t) of+ Just d -> d+ Nothing -> error ("Cannot parse Day from text: " <> cs t)+toDay v = error ("Cannot convert to Day: " <> show v)++toDouble :: Value -> Double+toDouble (Number n) = Scientific.toRealFloat n+toDouble (String t) = case Attoparsec.parseOnly Attoparsec.double t of+ Right d -> d+ Left _ -> error ("Cannot parse Double from text: " <> cs t)+toDouble v = error ("Cannot convert to Double: " <> show v)++toFloat :: Value -> Float+toFloat (Number n) = Scientific.toRealFloat n+toFloat (String t) = case Attoparsec.parseOnly Attoparsec.double t of+ Right d -> realToFrac d+ Left _ -> error ("Cannot parse Float from text: " <> cs t)+toFloat v = error ("Cannot convert to Float: " <> show v)++toScientific :: Value -> Scientific.Scientific+toScientific (Number n) = n+toScientific (String t) = case Attoparsec.parseOnly Attoparsec.scientific t of+ Right s -> s+ Left _ -> error ("Cannot parse Scientific from text: " <> cs t)+toScientific v = error ("Cannot convert to Scientific: " <> show v)++toByteString :: Value -> ByteString+toByteString (String t) = cs t+toByteString v = cs (toText v)
+ IHP/DataSync/Types.hs view
@@ -0,0 +1,76 @@+module IHP.DataSync.Types where++import IHP.Prelude+import Data.Aeson+import IHP.DataSync.DynamicQuery+import qualified Hasql.Connection as Hasql+import Control.Concurrent.MVar as MVar+++data DataSyncMessage+ = DataSyncQuery { query :: !DynamicSQLQuery, requestId :: !Int, transactionId :: !(Maybe UUID) }+ | CreateDataSubscription { query :: !DynamicSQLQuery, requestId :: !Int }+ | CreateCountSubscription { query :: !DynamicSQLQuery, requestId :: !Int }+ | DeleteDataSubscription { subscriptionId :: !UUID, requestId :: !Int }+ | CreateRecordMessage { table :: !Text, record :: !(HashMap Text Value), requestId :: !Int, transactionId :: !(Maybe UUID) }+ | CreateRecordsMessage { table :: !Text, records :: ![HashMap Text Value], requestId :: !Int, transactionId :: !(Maybe UUID) }+ | UpdateRecordMessage { table :: !Text, id :: !UUID, patch :: !(HashMap Text Value), requestId :: !Int, transactionId :: !(Maybe UUID) }+ | UpdateRecordsMessage { table :: !Text, ids :: ![UUID], patch :: !(HashMap Text Value), requestId :: !Int, transactionId :: !(Maybe UUID) }+ | DeleteRecordMessage { table :: !Text, id :: !UUID, requestId :: !Int, transactionId :: !(Maybe UUID) }+ | DeleteRecordsMessage { table :: !Text, ids :: ![UUID], requestId :: !Int, transactionId :: !(Maybe UUID) }+ | StartTransaction { requestId :: !Int }+ | RollbackTransaction { requestId :: !Int, id :: !UUID }+ | CommitTransaction { requestId :: !Int, id :: !UUID }+ | LoginWithEmailAndPassword { requestId :: !Int, email :: !Text, password :: !Text }+ | LoginWithJWT { requestId :: !Int, jwt :: !Text }+ | CreateUser { requestId :: !Int, email :: !Text, password :: !Text }+ | ConfirmUser { requestId :: !Int, userId :: !UUID, token :: !Text }+ deriving (Eq, Show)++data DataSyncResponse+ = DataSyncResult { result :: ![[Field]], requestId :: !Int }+ | DataSyncError { requestId :: !Int, errorMessage :: !Text }+ | FailedToDecodeMessageError { errorMessage :: !Text }+ | DidCreateDataSubscription { requestId :: !Int, subscriptionId :: !UUID, result :: ![[Field]] }+ | DidCreateCountSubscription { requestId :: !Int, subscriptionId :: !UUID, count :: !Int }+ | DidDeleteDataSubscription { requestId :: !Int, subscriptionId :: !UUID }+ | DidInsert { subscriptionId :: !UUID, record :: ![Field] }+ | DidUpdate { subscriptionId :: !UUID, id :: UUID, changeSet :: !(Maybe Value), appendSet :: !(Maybe Value) }+ | DidDelete { subscriptionId :: !UUID, id :: !UUID }+ | DidChangeCount { subscriptionId :: !UUID, count :: !Int }+ | DidCreateRecord { requestId :: !Int, record :: ![Field] } -- ^ Response to 'CreateRecordMessage'+ | DidCreateRecords { requestId :: !Int, records :: ![[Field]] } -- ^ Response to 'CreateRecordsMessage'+ | DidUpdateRecord { requestId :: !Int, record :: ![Field] } -- ^ Response to 'UpdateRecordMessage'+ | DidUpdateRecords { requestId :: !Int, records :: ![[Field]] } -- ^ Response to 'UpdateRecordsMessage'+ | DidDeleteRecord { requestId :: !Int }+ | DidDeleteRecords { requestId :: !Int }+ | DidStartTransaction { requestId :: !Int, transactionId :: !UUID }+ | DidRollbackTransaction { requestId :: !Int, transactionId :: !UUID }+ | DidCommitTransaction { requestId :: !Int, transactionId :: !UUID }++ | LoginSuccessful { requestId :: !Int, userId :: !UUID, jwt :: !Text }+ | UserLocked { requestId :: !Int }+ | UserUnconfirmed { requestId :: !Int }+ | InvalidCredentials { requestId :: !Int }++ | DidCreateUser { requestId :: !Int, userId :: !UUID, emailConfirmationRequired :: !Bool, jwt :: !Text }+ | CreateUserFailed { requestId :: !Int, validationFailures :: [(Text, Text)] }+ | DidConfirmUser { requestId :: !Int, jwt :: !Text }+ | DidConfirmUserAlready { requestId :: !Int }+ | ConfirmUserFailed { requestId :: !Int }++data GraphQLResult = GraphQLResult { graphQLResult :: !UndecodedJSON, requestId :: !Int }++data DataSyncTransaction+ = DataSyncTransaction+ { id :: !UUID+ , connection :: !Hasql.Connection+ , close :: MVar ()+ }++data DataSyncController+ = DataSyncController+ | DataSyncReady+ { subscriptions :: !(HashMap UUID (MVar.MVar ()))+ , transactions :: !(HashMap UUID DataSyncTransaction)+ }
+ LICENSE view
@@ -0,0 +1,21 @@+The MIT License (MIT)++Copyright (c) 2020 digitally induced GmbH++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.
+ Test/DataSync/ChangeNotifications.hs view
@@ -0,0 +1,328 @@+module Test.DataSync.ChangeNotifications where++import Test.Hspec+import IHP.Prelude+import Data.Aeson+import IHP.DataSync.ChangeNotifications (Change(..), makeCachedInstallTableChangeTriggers, installTableChangeTriggers)+import IHP.DataSync.ControllerImpl (changesToValue)+import IHP.DataSync.DynamicQueryCompiler (Renamer(..))+import IHP.DataSync.DynamicQuery (ConditionExpression(..), ConditionOperator(..), FunctionCall(..), conditionColumns)+import IHP.DataSync.RowLevelSecurity (TableWithRLS(..))+import qualified Data.Set as Set+import qualified Prelude+import qualified Hasql.Pool+import qualified Hasql.Pool.Config as Hasql.Pool.Config+import qualified Hasql.Connection.Settings as HasqlSettings+import qualified Hasql.Session as Session+import IHP.DataSync.Hasql (runSession)+import qualified Hasql.Decoders as Decoders+import qualified Hasql.Encoders as Encoders+import qualified Hasql.Statement as Statement+import qualified Control.Exception as Exception+import System.Environment (lookupEnv)+import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, readMVar)+import Control.Monad (replicateM)+import qualified Data.UUID.V4 as UUID+import qualified Data.UUID as UUID+import qualified Data.Text as Text++tests = do+ describe "IHP.DataSync.ChangeNotifications" do+ describe "FromJSON Change" do+ it "parses a regular Change with 'new' key" do+ let json = "{\"col\":\"body\",\"new\":\"Hello\"}"+ let expected = Change { col = "body", new = String "Hello" }+ eitherDecode json `shouldBe` Right expected++ it "parses an AppendChange with 'append' key" do+ let json = "{\"col\":\"body\",\"append\":\" World\"}"+ let expected = AppendChange { col = "body", append = " World" }+ eitherDecode json `shouldBe` Right expected++ describe "ToJSON Change" do+ it "serializes a regular Change" do+ let change = Change { col = "body", new = String "Hello" }+ let result = toJSON change+ result `shouldBe` object ["col" .= ("body" :: Text), "new" .= ("Hello" :: Text)]++ it "serializes an AppendChange" do+ let change = AppendChange { col = "body", append = " World" }+ let result = toJSON change+ result `shouldBe` object ["col" .= ("body" :: Text), "append" .= (" World" :: Text)]++ describe "ToJSON/FromJSON round-trip" do+ it "round-trips a regular Change" do+ let change = Change { col = "title", new = String "New Title" }+ (eitherDecode (encode change)) `shouldBe` Right change++ it "round-trips an AppendChange" do+ let change = AppendChange { col = "body", append = " appended text" }+ (eitherDecode (encode change)) `shouldBe` Right change++ describe "changesToValue" do+ let identityRenamer = Renamer { fieldToColumn = Prelude.id, columnToField = Prelude.id }++ it "splits mixed changes into changeSet and appendSet" do+ let changes =+ [ Change { col = "title", new = String "New Title" }+ , AppendChange { col = "body", append = " more text" }+ ]+ let (changeSet, appendSet) = changesToValue identityRenamer changes+ changeSet `shouldBe` Just (object ["title" .= ("New Title" :: Text)])+ appendSet `shouldBe` Just (object ["body" .= (" more text" :: Text)])++ it "returns empty appendSet when all changes are regular" do+ let changes =+ [ Change { col = "title", new = String "New Title" }+ , Change { col = "body", new = String "Full body" }+ ]+ let (changeSet, appendSet) = changesToValue identityRenamer changes+ changeSet `shouldBe` Just (object ["title" .= ("New Title" :: Text), "body" .= ("Full body" :: Text)])+ appendSet `shouldBe` Nothing++ it "returns empty changeSet when all changes are appends" do+ let changes =+ [ AppendChange { col = "body", append = " suffix" }+ ]+ let (changeSet, appendSet) = changesToValue identityRenamer changes+ changeSet `shouldBe` Nothing+ appendSet `shouldBe` Just (object ["body" .= (" suffix" :: Text)])++ it "applies renamer to column names" do+ let renamer = Renamer { fieldToColumn = Prelude.id, columnToField = \col -> case col of+ "user_name" -> "userName"+ other -> other+ }+ let changes =+ [ Change { col = "user_name", new = String "Alice" }+ , AppendChange { col = "user_name", append = " Smith" }+ ]+ let (changeSet, appendSet) = changesToValue renamer changes+ changeSet `shouldBe` Just (object ["userName" .= ("Alice" :: Text)])+ appendSet `shouldBe` Just (object ["userName" .= (" Smith" :: Text)])++ describe "conditionColumns" do+ it "returns a single column for a simple WHERE" do+ let condition = InfixOperatorExpression+ { left = ColumnExpression "conversationId"+ , op = OpEqual+ , right = LiteralExpression (String "00000000-0000-0000-0000-000000000000")+ }+ conditionColumns condition `shouldBe` Set.fromList ["conversationId"]++ it "returns multiple columns for a compound WHERE with AND" do+ let condition = InfixOperatorExpression+ { left = InfixOperatorExpression+ { left = ColumnExpression "conversationId"+ , op = OpEqual+ , right = LiteralExpression (String "00000000-0000-0000-0000-000000000000")+ }+ , op = OpAnd+ , right = InfixOperatorExpression+ { left = ColumnExpression "status"+ , op = OpEqual+ , right = LiteralExpression (String "active")+ }+ }+ conditionColumns condition `shouldBe` Set.fromList ["conversationId", "status"]++ it "extracts columns from nested AND/OR" do+ let condition = InfixOperatorExpression+ { left = InfixOperatorExpression+ { left = ColumnExpression "a"+ , op = OpEqual+ , right = LiteralExpression (Number 1)+ }+ , op = OpOr+ , right = InfixOperatorExpression+ { left = ColumnExpression "b"+ , op = OpAnd+ , right = ColumnExpression "c"+ }+ }+ conditionColumns condition `shouldBe` Set.fromList ["a", "b", "c"]++ it "returns empty set for CallExpression" do+ let condition = CallExpression (ToTSQuery "hello")+ conditionColumns condition `shouldBe` Set.empty++ it "returns empty set for ListExpression" do+ let condition = ListExpression [Number 1, Number 2]+ conditionColumns condition `shouldBe` Set.empty++ it "returns empty set for LiteralExpression" do+ let condition = LiteralExpression (String "hello")+ conditionColumns condition `shouldBe` Set.empty++ -- https://github.com/digitallyinduced/ihp/issues/2467+ describe "concurrent trigger installation" do+ it "does not exhaust the connection pool under concurrent trigger installation" do+ withDB \connStr -> do+ -- Tiny pool (2 connections) + short acquisition timeout to+ -- make pool exhaustion visible quickly.+ Exception.bracket (makePoolWithTimeout 2 3 connStr) Hasql.Pool.release \pool -> do+ execSQL pool "CREATE EXTENSION IF NOT EXISTS \"uuid-ossp\""++ -- Use a unique table name to avoid stale global MVar state+ -- across repeated test runs in the same ghci session+ tableUuid <- UUID.nextRandom+ let tableName = "test_concurrent_" <> Text.replace "-" "_" (UUID.toText tableUuid)+ execSQL pool (cs ("CREATE TABLE " <> tableName <> " (id UUID PRIMARY KEY DEFAULT gen_random_uuid(), body TEXT)"))++ let table = TableWithRLS tableName++ -- Create many independent cached installers, simulating+ -- concurrent WebSocket connections.+ -- On master: each gets its own IORef cache, so all of them+ -- call installTableChangeTriggers independently. The DDL+ -- requires AccessExclusiveLock, so they queue up behind+ -- each other at the DB level, each holding a pool+ -- connection while waiting → pool exhausted for seconds.+ -- On fix: all share the process-global MVar, so only 1 grabs+ -- a pool connection for DDL, the rest wait on the MVar+ -- in Haskell without consuming pool connections.+ installers <- replicateM 20000 (makeCachedInstallTableChangeTriggers pool)++ -- Barrier: ensure all 20000 threads are created and waiting+ -- before they rush the pool simultaneously.+ barrier <- newEmptyMVar++ result <- Exception.try $ do+ installsAsync <- async $+ mapConcurrently_ (\install -> readMVar barrier >> install table) installers++ -- Give threads time to start and block on barrier+ threadDelay 500_000++ -- Release all installers at once+ putMVar barrier ()++ wait installsAsync++ case result of+ Left (_ :: Hasql.Pool.UsageError) ->+ expectationFailure+ "Pool exhausted — concurrent trigger installation caused AcquisitionTimeoutUsageError"+ Right () -> pure ()++ describe "trigger installation with locked table" do+ it "succeeds on second call even when a writer holds RowExclusiveLock" do+ withDB \connStr -> do+ Exception.bracket (makePool connStr) Hasql.Pool.release \pool -> do+ execSQL pool "CREATE EXTENSION IF NOT EXISTS \"uuid-ossp\""++ tableUuid <- UUID.nextRandom+ let tableName = "test_lock_" <> Text.replace "-" "_" (UUID.toText tableUuid)+ execSQL pool (cs ("CREATE TABLE " <> tableName <> " (id UUID PRIMARY KEY DEFAULT gen_random_uuid(), body TEXT)"))++ let table = TableWithRLS tableName++ -- First call: install triggers (table is not locked)+ installTableChangeTriggers pool table++ -- Verify triggers were created+ triggerCount <- queryTriggerCount pool tableName+ triggerCount `shouldBe` 3++ -- Simulate a long-running writer holding RowExclusiveLock.+ -- RowExclusiveLock conflicts with ShareRowExclusiveLock+ -- (which CREATE TRIGGER takes), but NOT with the+ -- IF NOT EXISTS check (AccessShareLock on pg_trigger).+ -- An open SELECT only takes AccessShareLock which does+ -- NOT conflict with CREATE TRIGGER, so we must use a+ -- writer (INSERT) to properly test this.+ lockPool <- makePoolN 1 connStr+ execSQL lockPool (cs ("BEGIN; INSERT INTO " <> tableName <> " (body) VALUES ('lock holder')"))++ -- Second call: triggers already exist, so this should+ -- skip CREATE TRIGGER and succeed despite the writer lock.+ -- With the old code (DROP TRIGGER + CREATE TRIGGER always),+ -- this would block on the table lock.+ result <- Exception.try $ do+ execSQL pool "SET statement_timeout = '2s'"+ installTableChangeTriggers pool table+ execSQL pool "SET statement_timeout = '0'"++ Hasql.Pool.release lockPool++ case result of+ Left (e :: Exception.SomeException) ->+ expectationFailure+ (cs ("Second trigger install blocked or failed with locked table: " <> show e))+ Right () -> pure ()++-- DB helpers (same pattern as DataSyncIntegrationSpec.hs)++getMasterDatabaseUrl :: IO Text+getMasterDatabaseUrl = do+ envUrl <- lookupEnv "DATABASE_URL"+ case envUrl of+ Just url -> pure (cs url)+ Nothing -> pure "postgresql:///postgres"++makePool :: Text -> IO Hasql.Pool.Pool+makePool = makePoolN 4++makePoolN :: Int -> Text -> IO Hasql.Pool.Pool+makePoolN poolSize connStr = Hasql.Pool.acquire $ Hasql.Pool.Config.settings+ [ Hasql.Pool.Config.size poolSize+ , Hasql.Pool.Config.staticConnectionSettings+ (HasqlSettings.connectionString connStr)+ ]++makePoolWithTimeout :: Int -> Int -> Text -> IO Hasql.Pool.Pool+makePoolWithTimeout poolSize timeoutSeconds connStr = Hasql.Pool.acquire $ Hasql.Pool.Config.settings+ [ Hasql.Pool.Config.size poolSize+ , Hasql.Pool.Config.acquisitionTimeout (fromIntegral timeoutSeconds)+ , Hasql.Pool.Config.staticConnectionSettings+ (HasqlSettings.connectionString connStr)+ ]++execSQL :: Hasql.Pool.Pool -> ByteString -> IO ()+execSQL pool sql = runSession pool (Session.script (cs sql))++canConnectToPostgres :: IO Bool+canConnectToPostgres = do+ masterUrl <- getMasterDatabaseUrl+ result <- Exception.try $ Exception.bracket (makePool masterUrl) Hasql.Pool.release+ (\pool -> execSQL pool "SELECT 1")+ case result of+ Left (_ :: Exception.SomeException) -> pure False+ Right _ -> pure True++withTestDatabase :: (Text -> IO a) -> IO a+withTestDatabase action = do+ masterUrl <- getMasterDatabaseUrl+ testDbName <- randomDatabaseName+ Exception.bracket (makePool masterUrl) Hasql.Pool.release \masterPool -> do+ execSQL masterPool (cs ("CREATE DATABASE " <> testDbName))+ let testConnStr = "dbname=" <> testDbName+ Exception.finally+ (action testConnStr)+ (execSQL masterPool (cs ("DROP DATABASE " <> testDbName <> " WITH (FORCE)")))++randomDatabaseName :: IO Text+randomDatabaseName = do+ uuid <- UUID.nextRandom+ let name = "ihp_test_cn_" <> (uuid |> UUID.toText |> Text.replace "-" "_")+ pure name++withHasqlPool :: Text -> (Hasql.Pool.Pool -> IO a) -> IO a+withHasqlPool connStr action =+ Exception.bracket (makePool connStr) Hasql.Pool.release action++queryTriggerCount :: Hasql.Pool.Pool -> Text -> IO Int+queryTriggerCount pool tableName = do+ let session = Session.statement (cs tableName) $ Statement.preparable+ "SELECT count(*)::int FROM pg_trigger WHERE tgrelid = $1::regclass AND tgname LIKE 'did_%'"+ (Encoders.param (Encoders.nonNullable Encoders.text))+ (Decoders.singleRow (Decoders.column (Decoders.nonNullable (fromIntegral <$> Decoders.int4))))+ runSession pool session++withDB :: (Text -> IO ()) -> IO ()+withDB action = do+ available <- canConnectToPostgres+ if available+ then withTestDatabase action+ else pendingWith "PostgreSQL not available (set DATABASE_URL or start a local Postgres)"
+ Test/DataSync/DataSyncIntegrationSpec.hs view
@@ -0,0 +1,535 @@+{-# LANGUAGE UndecidableInstances #-}+module Test.DataSync.DataSyncIntegrationSpec where++import Test.Hspec+import IHP.Prelude+import qualified Hasql.Pool+import qualified Hasql.Pool.Config as Hasql.Pool.Config+import qualified Hasql.Connection.Settings as HasqlSettings+import qualified Hasql.Session as Session+import IHP.DataSync.Hasql (runSession)+import IHP.DataSync.ControllerImpl (runDataSyncController)+import IHP.DataSync.Types+import IHP.DataSync.DynamicQuery (Field(..))+import IHP.DataSync.DynamicQueryCompiler (camelCaseRenamer)+import IHP.DataSync.RowLevelSecurity (makeCachedEnsureRLSEnabled)+import qualified IHP.DataSync.ChangeNotifications as ChangeNotifications+import IHP.RequestVault (pgListenerVaultKey, frameworkConfigVaultKey)+import IHP.Controller.Context (newControllerContext, putContext, freeze)+import IHP.LoginSupport.Types (HasNewSessionUrl(..), CurrentUserRecord)+import qualified IHP.ModelSupport as ModelSupport+import IHP.ModelSupport.Types (Id'(..), PrimaryKey)+import qualified IHP.PGListener as PGListener+import IHP.FrameworkConfig (buildFrameworkConfig)+import IHP.FrameworkConfig.Types++import qualified Data.Vault.Lazy as Vault+import qualified Data.UUID.V4 as UUID+import qualified Data.UUID as UUID+import qualified Data.Text as Text+import qualified Control.Exception as Exception+import System.Environment (lookupEnv)+import Network.Wai (defaultRequest, vault)+import Data.Aeson (Value(..), object, (.=))+import qualified Data.Aeson as Aeson+import Control.Concurrent.STM+import Control.Concurrent (threadDelay)+import qualified IHP.Log as Log++-- | Define CurrentUserRecord for this test module+data TestUser = TestUser { id :: Id' "test_users" }+ deriving (Show, Typeable)++type instance CurrentUserRecord = TestUser+type instance GetTableName TestUser = "test_users"+type instance PrimaryKey "test_users" = UUID++instance HasNewSessionUrl TestUser where+ newSessionUrl _ = "/"++-- | Get the master database URL from DATABASE_URL env var or use a sensible default+getMasterDatabaseUrl :: IO Text+getMasterDatabaseUrl = do+ envUrl <- lookupEnv "DATABASE_URL"+ case envUrl of+ Just url -> pure (cs url)+ Nothing -> pure "postgresql:///postgres"++-- | Create a hasql pool for the given connection string+makePool :: Text -> IO Hasql.Pool.Pool+makePool connStr = Hasql.Pool.acquire $ Hasql.Pool.Config.settings+ [ Hasql.Pool.Config.size 4+ , Hasql.Pool.Config.staticConnectionSettings+ (HasqlSettings.connectionString connStr)+ ]++-- | Run a raw SQL statement on a pool (for test setup)+execSQL :: Hasql.Pool.Pool -> ByteString -> IO ()+execSQL pool sql = runSession pool (Session.script (cs sql))++-- | Check if we can connect to Postgres+canConnectToPostgres :: IO Bool+canConnectToPostgres = do+ masterUrl <- getMasterDatabaseUrl+ result <- Exception.try $ Exception.bracket (makePool masterUrl) Hasql.Pool.release+ (\pool -> execSQL pool "SELECT 1")+ case result of+ Left (_ :: Exception.SomeException) -> pure False+ Right _ -> pure True++-- | Create a temporary test database, run the action, then drop it+withTestDatabase :: (Text -> IO a) -> IO a+withTestDatabase action = do+ masterUrl <- getMasterDatabaseUrl+ testDbName <- randomDatabaseName+ Exception.bracket (makePool masterUrl) Hasql.Pool.release \masterPool -> do+ execSQL masterPool (cs ("CREATE DATABASE " <> testDbName))+ let testConnStr = "dbname=" <> testDbName+ Exception.finally+ (action testConnStr)+ (execSQL masterPool (cs ("DROP DATABASE " <> testDbName <> " WITH (FORCE)")))++-- | Generate a random database name for test isolation+randomDatabaseName :: IO Text+randomDatabaseName = do+ uuid <- UUID.nextRandom+ let name = "ihp_test_datasync_" <> (uuid |> UUID.toText |> Text.replace "-" "_")+ pure name++-- | Create a hasql pool, run the action, then release+withHasqlPool :: Text -> (Hasql.Pool.Pool -> IO a) -> IO a+withHasqlPool connStr action =+ Exception.bracket (makePool connStr) Hasql.Pool.release action++-- | Run a database test, skipping if Postgres is not available+withDB :: (Text -> IO ()) -> IO ()+withDB action = do+ available <- canConnectToPostgres+ if available+ then withTestDatabase action+ else pendingWith "PostgreSQL not available (set DATABASE_URL or start a local Postgres)"++-- | Set up the test database schema+setupTestSchema :: Hasql.Pool.Pool -> IO ()+setupTestSchema pool = do+ execSQL pool "CREATE EXTENSION IF NOT EXISTS \"uuid-ossp\""+ execSQL pool "CREATE TABLE test_users (id UUID PRIMARY KEY DEFAULT gen_random_uuid())"+ execSQL pool "CREATE TABLE messages (id UUID PRIMARY KEY DEFAULT gen_random_uuid(), user_id UUID NOT NULL, body TEXT NOT NULL)"+ execSQL pool "ALTER TABLE messages ENABLE ROW LEVEL SECURITY"+ execSQL pool "CREATE POLICY messages_policy ON messages USING (user_id = current_setting('rls.ihp_user_id')::uuid)"+ -- Create the authenticated role and grant permissions+ execSQL pool "DO $$ BEGIN CREATE ROLE ihp_authenticated NOLOGIN; EXCEPTION WHEN duplicate_object THEN null; END $$"+ execSQL pool "GRANT USAGE ON SCHEMA public TO ihp_authenticated"+ execSQL pool "GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO ihp_authenticated"++-- | Insert test data and return the user ID and message ID+insertTestData :: Hasql.Pool.Pool -> IO (UUID, UUID)+insertTestData pool = do+ userId <- UUID.nextRandom+ messageId <- UUID.nextRandom+ execSQL pool (cs ("INSERT INTO test_users (id) VALUES ('" <> UUID.toText userId <> "')"))+ execSQL pool (cs ("INSERT INTO messages (id, user_id, body) VALUES ('" <> UUID.toText messageId <> "', '" <> UUID.toText userId <> "', 'Hello')"))+ pure (userId, messageId)++-- | Run the DataSync controller with TQueue-based I/O, yielding send/receive handles to the test.+withDataSyncController+ :: Text -- ^ database connection string+ -> UUID -- ^ test user ID+ -> ((ByteString -> IO (), IO DataSyncResponse, Async ()) -> IO a)+ -> IO a+withDataSyncController connStr testUserId action = do+ withHasqlPool connStr \hasqlPool -> do+ -- If connStr already has dbname= prefix, use it directly; otherwise format it+ let actualConnStr = if "dbname=" `Text.isPrefixOf` connStr+ then cs connStr+ else cs ("dbname=" <> connStr)+ logger <- Log.newLogger def { Log.level = Log.Error }+ ModelSupport.withModelContext actualConnStr logger \modelContext -> do+ PGListener.withPGListener actualConnStr logger \pgListener -> do+ frameworkConfig <- buildFrameworkConfig (pure ())+ let frameworkConfig' = frameworkConfig { databaseUrl = actualConnStr }++ let v = Vault.empty+ |> Vault.insert pgListenerVaultKey pgListener+ |> Vault.insert frameworkConfigVaultKey frameworkConfig'+ let request = defaultRequest { vault = v }++ -- Set up ControllerContext with the request and current user+ let ?request = request+ context <- newControllerContext+ let ?context = context++ -- Put the current user into context so currentUserOrNothing can find it+ putContext (Just (TestUser { id = Id testUserId }) :: Maybe TestUser)++ -- Freeze the context so it can be accessed from pure code+ frozenContext <- freeze ?context+ let ?context = frozenContext++ -- Create the DataSync state IORef+ stateRef <- newIORef DataSyncController+ let ?state = stateRef++ -- Create TQueues for communication+ inQueue <- newTQueueIO :: IO (TQueue ByteString)+ outQueue <- newTQueueIO :: IO (TQueue DataSyncResponse)++ let receiveData = atomically $ readTQueue inQueue+ let sendJSON response = atomically $ writeTQueue outQueue response++ -- Build the helper functions+ ensureRLSEnabled <- makeCachedEnsureRLSEnabled hasqlPool+ let installTableChangeTriggers = ChangeNotifications.installTableChangeTriggers hasqlPool++ -- Start the controller in an async thread+ let ?modelContext = modelContext+ controllerAsync <- async $+ runDataSyncController hasqlPool ensureRLSEnabled installTableChangeTriggers receiveData sendJSON (\_ _ -> pure ()) (\_ -> camelCaseRenamer)++ -- Run the test action, then clean up+ Exception.finally+ (action (\msg -> atomically $ writeTQueue inQueue msg, readResponseWithTimeout outQueue, controllerAsync))+ (cancel controllerAsync)++-- | Read the next DataSyncResponse with a timeout+readResponseWithTimeout :: TQueue DataSyncResponse -> IO DataSyncResponse+readResponseWithTimeout outQueue = do+ result <- race (threadDelay 5_000_000) (atomically $ readTQueue outQueue)+ case result of+ Left () -> error "readResponse: timed out waiting for DataSync response"+ Right response -> pure response++-- | Encode a DataSyncQuery message as JSON+encodeDataSyncQuery :: Text -> Int -> Maybe UUID -> ByteString+encodeDataSyncQuery table requestId transactionId = cs $ Aeson.encode $ object+ [ "tag" .= ("DataSyncQuery" :: Text)+ , "query" .= object+ [ "table" .= table+ , "selectedColumns" .= object ["tag" .= ("SelectAll" :: Text)]+ , "whereCondition" .= Null+ , "orderByClause" .= ([] :: [Value])+ , "distinctOnColumn" .= Null+ , "limit" .= Null+ , "offset" .= Null+ ]+ , "requestId" .= requestId+ , "transactionId" .= transactionId+ ]++-- | Encode a CreateRecordMessage as JSON+encodeCreateRecord :: Text -> [(Text, Value)] -> Int -> Maybe UUID -> ByteString+encodeCreateRecord table fields requestId transactionId = cs $ Aeson.encode $ object+ [ "tag" .= ("CreateRecordMessage" :: Text)+ , "table" .= table+ , "record" .= object (map (\(k, v) -> (cs k) .= v) fields)+ , "requestId" .= requestId+ , "transactionId" .= transactionId+ ]++-- | Encode an UpdateRecordMessage as JSON+encodeUpdateRecord :: Text -> UUID -> [(Text, Value)] -> Int -> Maybe UUID -> ByteString+encodeUpdateRecord table recordId patch requestId transactionId = cs $ Aeson.encode $ object+ [ "tag" .= ("UpdateRecordMessage" :: Text)+ , "table" .= table+ , "id" .= recordId+ , "patch" .= object (map (\(k, v) -> (cs k) .= v) patch)+ , "requestId" .= requestId+ , "transactionId" .= transactionId+ ]++-- | Encode a DeleteRecordMessage as JSON+encodeDeleteRecord :: Text -> UUID -> Int -> Maybe UUID -> ByteString+encodeDeleteRecord table recordId requestId transactionId = cs $ Aeson.encode $ object+ [ "tag" .= ("DeleteRecordMessage" :: Text)+ , "table" .= table+ , "id" .= recordId+ , "requestId" .= requestId+ , "transactionId" .= transactionId+ ]++-- | Encode a CreateDataSubscription as JSON+encodeCreateDataSubscription :: Text -> Int -> ByteString+encodeCreateDataSubscription table requestId = cs $ Aeson.encode $ object+ [ "tag" .= ("CreateDataSubscription" :: Text)+ , "query" .= object+ [ "table" .= table+ , "selectedColumns" .= object ["tag" .= ("SelectAll" :: Text)]+ , "whereCondition" .= Null+ , "orderByClause" .= ([] :: [Value])+ , "distinctOnColumn" .= Null+ , "limit" .= Null+ , "offset" .= Null+ ]+ , "requestId" .= requestId+ ]++-- | Encode a DeleteDataSubscription as JSON+encodeDeleteDataSubscription :: UUID -> Int -> ByteString+encodeDeleteDataSubscription subscriptionId requestId = cs $ Aeson.encode $ object+ [ "tag" .= ("DeleteDataSubscription" :: Text)+ , "subscriptionId" .= subscriptionId+ , "requestId" .= requestId+ ]++-- | Encode a StartTransaction as JSON+encodeStartTransaction :: Int -> ByteString+encodeStartTransaction requestId = cs $ Aeson.encode $ object+ [ "tag" .= ("StartTransaction" :: Text)+ , "requestId" .= requestId+ ]++-- | Encode a CommitTransaction as JSON+encodeCommitTransaction :: Int -> UUID -> ByteString+encodeCommitTransaction requestId transactionId = cs $ Aeson.encode $ object+ [ "tag" .= ("CommitTransaction" :: Text)+ , "requestId" .= requestId+ , "id" .= transactionId+ ]++-- | Encode a RollbackTransaction as JSON+encodeRollbackTransaction :: Int -> UUID -> ByteString+encodeRollbackTransaction requestId transactionId = cs $ Aeson.encode $ object+ [ "tag" .= ("RollbackTransaction" :: Text)+ , "requestId" .= requestId+ , "id" .= transactionId+ ]++-- | Find a field value by name in a list of Fields+findField :: Text -> [Field] -> Maybe Value+findField name fields = case filter (\f -> f.fieldName == name) fields of+ (f:_) -> Just f.fieldValue+ [] -> Nothing++tests :: Spec+tests = do+ describe "IHP.DataSync Integration" do+ describe "DataSyncQuery" do+ it "returns rows from a table with RLS" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, messageId) <- insertTestData pool++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeDataSyncQuery "messages" 1 Nothing)+ response <- recv+ case response of+ DataSyncResult { result, requestId } -> do+ requestId `shouldBe` 1+ length result `shouldBe` 1+ case result of+ (row:_) -> do+ findField "id" row `shouldBe` Just (String (UUID.toText messageId))+ findField "body" row `shouldBe` Just (String "Hello")+ [] -> expectationFailure "Expected at least one row"+ DataSyncError { errorMessage } ->+ expectationFailure (cs $ "Unexpected error: " <> errorMessage)+ _ -> expectationFailure "Expected DataSyncResult"++ describe "CreateRecordMessage" do+ it "creates a record and returns it" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, _) <- insertTestData pool++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeCreateRecord "messages"+ [ ("userId", String (UUID.toText userId))+ , ("body", String "New message")+ ] 2 Nothing)+ response <- recv+ case response of+ DidCreateRecord { record, requestId } -> do+ requestId `shouldBe` 2+ findField "body" record `shouldBe` Just (String "New message")+ DataSyncError { errorMessage } ->+ expectationFailure (cs $ "Unexpected error: " <> errorMessage)+ _ -> expectationFailure "Expected DidCreateRecord"++ describe "UpdateRecordMessage" do+ it "updates a record and returns the updated version" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, messageId) <- insertTestData pool++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeUpdateRecord "messages" messageId+ [ ("body", String "Updated message") ] 3 Nothing)+ response <- recv+ case response of+ DidUpdateRecord { record, requestId } -> do+ requestId `shouldBe` 3+ findField "body" record `shouldBe` Just (String "Updated message")+ DataSyncError { errorMessage } ->+ expectationFailure (cs $ "Unexpected error: " <> errorMessage)+ _ -> expectationFailure "Expected DidUpdateRecord"++ describe "DeleteRecordMessage" do+ it "deletes a record" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, messageId) <- insertTestData pool++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeDeleteRecord "messages" messageId 4 Nothing)+ response <- recv+ case response of+ DidDeleteRecord { requestId } ->+ requestId `shouldBe` 4+ DataSyncError { errorMessage } ->+ expectationFailure (cs $ "Unexpected error: " <> errorMessage)+ _ -> expectationFailure "Expected DidDeleteRecord"++ -- Verify the record is gone by querying+ send (encodeDataSyncQuery "messages" 5 Nothing)+ response2 <- recv+ case response2 of+ DataSyncResult { result } ->+ length result `shouldBe` 0+ _ -> expectationFailure "Expected DataSyncResult"++ describe "RLS enforcement" do+ it "only returns records visible to the current user" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, _) <- insertTestData pool++ -- Insert a message belonging to a different user+ otherUserId <- UUID.nextRandom+ otherMsgId <- UUID.nextRandom+ execSQL pool (cs ("INSERT INTO test_users (id) VALUES ('" <> UUID.toText otherUserId <> "')"))+ execSQL pool (cs ("INSERT INTO messages (id, user_id, body) VALUES ('" <> UUID.toText otherMsgId <> "', '" <> UUID.toText otherUserId <> "', 'Other user message')"))++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeDataSyncQuery "messages" 6 Nothing)+ response <- recv+ case response of+ DataSyncResult { result } ->+ -- Should only see the message belonging to our test user+ length result `shouldBe` 1+ DataSyncError { errorMessage } ->+ expectationFailure (cs $ "Unexpected error: " <> errorMessage)+ _ -> expectationFailure "Expected DataSyncResult"++ it "rejects queries on tables without RLS" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, _) <- insertTestData pool+ -- test_users does NOT have RLS enabled+ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeDataSyncQuery "test_users" 7 Nothing)+ response <- recv+ case response of+ DataSyncError {} -> pure ()+ _ -> expectationFailure "Expected DataSyncError for table without RLS"++ describe "Subscriptions" do+ it "creates and deletes a data subscription" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, _) <- insertTestData pool++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeCreateDataSubscription "messages" 8)+ response <- recv+ case response of+ DidCreateDataSubscription { requestId, subscriptionId, result } -> do+ requestId `shouldBe` 8+ length result `shouldBe` 1++ -- Now delete the subscription+ send (encodeDeleteDataSubscription subscriptionId 9)+ response2 <- recv+ case response2 of+ DidDeleteDataSubscription { requestId, subscriptionId = deletedId } -> do+ requestId `shouldBe` 9+ deletedId `shouldBe` subscriptionId+ _ -> expectationFailure "Expected DidDeleteDataSubscription"+ DataSyncError { errorMessage } ->+ expectationFailure (cs $ "Unexpected error: " <> errorMessage)+ _ -> expectationFailure "Expected DidCreateDataSubscription"++ describe "Transactions" do+ it "starts, uses, and commits a transaction" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, _) <- insertTestData pool++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeStartTransaction 10)+ response <- recv+ case response of+ DidStartTransaction { requestId, transactionId } -> do+ requestId `shouldBe` 10++ -- Create a record within the transaction+ send (encodeCreateRecord "messages"+ [ ("userId", String (UUID.toText userId))+ , ("body", String "Transactional message")+ ] 11 (Just transactionId))+ response2 <- recv+ case response2 of+ DidCreateRecord {} -> pure ()+ DataSyncError { errorMessage } ->+ expectationFailure (cs $ "Create in txn failed: " <> errorMessage)+ _ -> expectationFailure "Expected DidCreateRecord"++ -- Commit the transaction+ send (encodeCommitTransaction 12 transactionId)+ response3 <- recv+ case response3 of+ DidCommitTransaction { requestId } ->+ requestId `shouldBe` 12+ _ -> expectationFailure "Expected DidCommitTransaction"++ -- Verify the record persists after commit+ send (encodeDataSyncQuery "messages" 13 Nothing)+ response4 <- recv+ case response4 of+ DataSyncResult { result } ->+ -- Original + transactional message+ length result `shouldBe` 2+ _ -> expectationFailure "Expected DataSyncResult"+ _ -> expectationFailure "Expected DidStartTransaction"++ it "rolls back a transaction" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ setupTestSchema pool+ (userId, _) <- insertTestData pool++ withDataSyncController connStr userId \(send, recv, _) -> do+ send (encodeStartTransaction 20)+ response <- recv+ case response of+ DidStartTransaction { transactionId } -> do+ -- Create a record in the transaction+ send (encodeCreateRecord "messages"+ [ ("userId", String (UUID.toText userId))+ , ("body", String "Will be rolled back")+ ] 21 (Just transactionId))+ _ <- recv -- DidCreateRecord++ -- Rollback+ send (encodeRollbackTransaction 22 transactionId)+ response3 <- recv+ case response3 of+ DidRollbackTransaction { requestId } ->+ requestId `shouldBe` 22+ _ -> expectationFailure "Expected DidRollbackTransaction"++ -- Verify the record was NOT persisted+ send (encodeDataSyncQuery "messages" 23 Nothing)+ response4 <- recv+ case response4 of+ DataSyncResult { result } ->+ length result `shouldBe` 1 -- only the original+ _ -> expectationFailure "Expected DataSyncResult"+ _ -> expectationFailure "Expected DidStartTransaction"
+ Test/DataSync/DynamicQueryCompiler.hs view
@@ -0,0 +1,291 @@+{-|+Copyright: (c) digitally induced GmbH, 2021+-}+module Test.DataSync.DynamicQueryCompiler where++import Test.Hspec+import IHP.Prelude+import IHP.DataSync.DynamicQueryCompiler+import IHP.DataSync.DynamicQuery+import IHP.QueryBuilder hiding (OrderByClause)+import qualified Data.HashMap.Strict as HashMap+import qualified Data.Aeson as Aeson+import qualified Hasql.DynamicStatements.Snippet as Snippet++-- | Column types for the "posts" table used in tests.+-- Simulates database column order: id first, then other columns in schema definition order.+postsTypes :: ColumnTypeInfo+postsTypes = ColumnTypeInfo+ { typeMap = HashMap.fromList+ [ ("user_id", "uuid")+ , ("id", "uuid")+ , ("a", "text")+ , ("title", "text")+ , ("ts", "tsvector")+ , ("group_id", "uuid")+ ]+ , orderedColumns = ["id", "user_id", "title", "a", "ts", "group_id"]+ }++-- | Column types for the "products" table used in tests.+productsTypes :: ColumnTypeInfo+productsTypes = ColumnTypeInfo+ { typeMap = HashMap.fromList [("ts", "tsvector")]+ , orderedColumns = ["ts"]+ }++-- | Compile a query with the camelCase renamer and typed encoding.+compile :: ColumnTypeInfo -> DynamicSQLQuery -> Snippet.Snippet+compile = compileQueryTyped camelCaseRenamer++-- | Expected SELECT clause for postsTypes when using SelectAll+-- Columns appear in database schema order (from orderedColumns), with camelCase aliases+postsSelectAll :: Text+postsSelectAll = "\"id\", \"user_id\" AS \"userId\", \"title\", \"a\", \"ts\", \"group_id\" AS \"groupId\""++tests = do+ describe "IHP.DataSync.DynamicQueryCompiler" do+ describe "compileQueryTyped" do+ it "compile a basic select query" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Nothing+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ -- SelectAll expands to all columns with appropriate camelCase aliases (id first, then alphabetically)+ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\"")++ it "compile a select query with order by" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Nothing+ , distinctOnColumn = Nothing+ , orderByClause = [OrderByClause { orderByColumn = "title", orderByDirection = Desc }]+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" ORDER BY \"title\" DESC")++ it "compile a select query with multiple order bys" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Nothing+ , orderByClause = [+ OrderByClause { orderByColumn = "createdAt", orderByDirection = Desc },+ OrderByClause { orderByColumn = "title", orderByDirection = Asc }+ ]+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" ORDER BY \"created_at\" DESC, \"title\"")++ it "compile a basic select query with a where condition" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "userId") OpEqual (LiteralExpression (Aeson.String "b8553ce9-6a42-4a68-b5fc-259be3e2acdc"))+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"user_id\") = ($1)")++ it "compile a basic select query with a where condition and an order by" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "userId") OpEqual (LiteralExpression (Aeson.String "b8553ce9-6a42-4a68-b5fc-259be3e2acdc"))+ , orderByClause = [ OrderByClause { orderByColumn = "createdAt", orderByDirection = Desc } ]+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"user_id\") = ($1) ORDER BY \"created_at\" DESC")++ it "compile a basic select query with a limit" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "userId") OpEqual (LiteralExpression (Aeson.String "b8553ce9-6a42-4a68-b5fc-259be3e2acdc"))+ , distinctOnColumn = Nothing+ , orderByClause = []+ , limit = Just 50+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"user_id\") = ($1) LIMIT $2")++ it "compile a basic select query with an offset" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "userId") OpEqual (LiteralExpression (Aeson.String "b8553ce9-6a42-4a68-b5fc-259be3e2acdc"))+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Just 50+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"user_id\") = ($1) OFFSET $2")++ it "compile a basic select query with a limit and an offset" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "userId") OpEqual (LiteralExpression (Aeson.String "b8553ce9-6a42-4a68-b5fc-259be3e2acdc"))+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Just 25+ , offset = Just 50+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"user_id\") = ($1) LIMIT $2 OFFSET $3")++ it "compile 'field = NULL' conditions to 'field IS NULL'" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "userId") OpEqual (LiteralExpression Aeson.Null)+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"user_id\") IS NULL")++ it "compile 'field <> NULL' conditions to 'field IS NOT NULL'" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "userId") OpNotEqual (LiteralExpression Aeson.Null)+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"user_id\") IS NOT NULL")++ it "compile 'field IN (NULL)' conditions to 'field IS NULL'" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "a") OpIn (ListExpression { values = [Aeson.Null] })+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"a\") IS NULL")++ it "compile 'field IN (NULL, 'string')' conditions to 'field IS NULL OR field IN ('string')'" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "a") OpIn (ListExpression { values = [Aeson.Null, Aeson.String "test" ] })+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE ((\"a\") IN ($1)) OR ((\"a\") IS NULL)")++ it "compile queries with TS expressions" do+ let query = DynamicSQLQuery+ { table = "products"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "ts") OpTSMatch (CallExpression { functionCall = ToTSQuery { text = "test" }})+ , orderByClause = [ OrderByTSRank { tsvector = "ts", tsquery = "test" } ]+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ -- productsTypes only has "ts" column (no rename needed since it's already snake_case = camelCase)+ Snippet.toSql (compile productsTypes query) `shouldBe`+ "SELECT \"ts\" FROM \"products\" WHERE (\"ts\") @@ (to_tsquery('english', $1)) ORDER BY ts_rank(\"ts\", to_tsquery('english', $2))"++ it "compile a basic select query with distinctOn" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Nothing+ , orderByClause = []+ , distinctOnColumn = Just "groupId"+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT DISTINCT ON (\"group_id\") " <> postsSelectAll <> " FROM \"posts\"")++ it "compile a WHERE IN query" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "id") OpIn (ListExpression { values = [Aeson.String "a5d7772f-c63f-4444-be69-dd9afd902e9b", Aeson.String "bb88d55a-1ed0-44ad-be13-d768f4b3f9ca"] })+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE (\"id\") IN ($1, $2)")++ it "compile an empty WHERE IN query to FALSE" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectAll+ , whereCondition = Just $ InfixOperatorExpression (ColumnExpression "id") OpIn (ListExpression { values = [] })+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ Snippet.toSql (compile postsTypes query) `shouldBe`+ ("SELECT " <> postsSelectAll <> " FROM \"posts\" WHERE FALSE")++ it "compile SelectSpecific with camelCase to snake_case aliases" do+ let query = DynamicSQLQuery+ { table = "posts"+ , selectedColumns = SelectSpecific ["userId", "title"]+ , whereCondition = Nothing+ , orderByClause = []+ , distinctOnColumn = Nothing+ , limit = Nothing+ , offset = Nothing+ }++ -- SelectSpecific columns get renamed to snake_case and aliased back to camelCase+ Snippet.toSql (compile postsTypes query) `shouldBe`+ "SELECT \"user_id\" AS \"userId\", \"title\" FROM \"posts\""
+ Test/DataSync/RLSIntegrationSpec.hs view
@@ -0,0 +1,122 @@+module Test.DataSync.RLSIntegrationSpec where++import Test.Hspec+import IHP.Prelude hiding (head)+import qualified Hasql.Pool+import qualified Hasql.Pool.Config as Hasql.Pool.Config+import qualified Hasql.Connection.Settings as HasqlSettings+import qualified Hasql.Session as Session+import IHP.DataSync.RowLevelSecurity (rlsPolicyColumns)+import IHP.DataSync.Hasql (runSession)+import qualified Data.Set as Set+import qualified Data.UUID.V4 as UUID+import qualified Data.UUID as UUID+import qualified Data.Text as Text+import qualified Control.Exception as Exception+import System.Environment (lookupEnv)++-- | Get the master database URL from DATABASE_URL env var or use a sensible default+getMasterDatabaseUrl :: IO Text+getMasterDatabaseUrl = do+ envUrl <- lookupEnv "DATABASE_URL"+ case envUrl of+ Just url -> pure (cs url)+ Nothing -> pure "postgresql:///postgres"++-- | Create a hasql pool for the given connection string+makePool :: Text -> IO Hasql.Pool.Pool+makePool connStr = Hasql.Pool.acquire $ Hasql.Pool.Config.settings+ [ Hasql.Pool.Config.size 2+ , Hasql.Pool.Config.staticConnectionSettings+ (HasqlSettings.connectionString connStr)+ ]++-- | Run a raw SQL statement on a pool (for test setup)+execSQL :: Hasql.Pool.Pool -> ByteString -> IO ()+execSQL pool sql = runSession pool (Session.script (cs sql))++-- | Check if we can connect to Postgres. Returns True if a connection succeeds.+canConnectToPostgres :: IO Bool+canConnectToPostgres = do+ masterUrl <- getMasterDatabaseUrl+ result <- Exception.try $ Exception.bracket (makePool masterUrl) Hasql.Pool.release+ (\pool -> execSQL pool "SELECT 1")+ case result of+ Left (_ :: Exception.SomeException) -> pure False+ Right _ -> pure True++-- | Create a temporary test database, run the action, then drop the database+withTestDatabase :: (Text -> IO a) -> IO a+withTestDatabase action = do+ masterUrl <- getMasterDatabaseUrl+ testDbName <- randomDatabaseName+ Exception.bracket (makePool masterUrl) Hasql.Pool.release \masterPool -> do+ execSQL masterPool (cs ("CREATE DATABASE " <> testDbName))+ let testConnStr = "dbname=" <> testDbName+ Exception.finally+ (action testConnStr)+ (execSQL masterPool (cs ("DROP DATABASE " <> testDbName <> " WITH (FORCE)")))++-- | Generate a random database name for test isolation+randomDatabaseName :: IO Text+randomDatabaseName = do+ uuid <- UUID.nextRandom+ let name = "ihp_test_rls_" <> (uuid |> UUID.toText |> Text.replace "-" "_")+ pure name++-- | Create a hasql pool for the given connection string, run the action, then release+withHasqlPool :: Text -> (Hasql.Pool.Pool -> IO a) -> IO a+withHasqlPool connStr action =+ Exception.bracket (makePool connStr) Hasql.Pool.release action++-- | Run a database test, skipping if Postgres is not available+withDB :: (Text -> IO ()) -> IO ()+withDB action = do+ available <- canConnectToPostgres+ if available+ then withTestDatabase action+ else pendingWith "PostgreSQL not available (set DATABASE_URL or start a local Postgres)"++tests :: Spec+tests = do+ describe "IHP.DataSync.RowLevelSecurity" do+ describe "rlsPolicyColumns" do+ it "returns columns referenced in a USING policy" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ execSQL pool "CREATE TABLE messages (id UUID PRIMARY KEY, user_id UUID NOT NULL, body TEXT)"+ execSQL pool "ALTER TABLE messages ENABLE ROW LEVEL SECURITY"+ execSQL pool "CREATE POLICY messages_policy ON messages USING (user_id = current_setting('rls.ihp_user_id')::uuid)"++ cols <- rlsPolicyColumns pool "messages"+ cols `shouldSatisfy` Set.member "user_id"++ it "returns columns from WITH CHECK policy" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ execSQL pool "CREATE TABLE projects (id UUID PRIMARY KEY, owner_id UUID NOT NULL, name TEXT)"+ execSQL pool "ALTER TABLE projects ENABLE ROW LEVEL SECURITY"+ execSQL pool "CREATE POLICY projects_insert ON projects FOR INSERT WITH CHECK (owner_id = current_setting('rls.ihp_user_id')::uuid)"++ cols <- rlsPolicyColumns pool "projects"+ cols `shouldSatisfy` Set.member "owner_id"++ it "returns empty set for a table without RLS policies" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ execSQL pool "CREATE TABLE no_rls_table (id UUID PRIMARY KEY, body TEXT)"++ cols <- rlsPolicyColumns pool "no_rls_table"+ cols `shouldBe` Set.empty++ it "returns columns from multiple policies" do+ withDB \connStr -> do+ withHasqlPool connStr \pool -> do+ execSQL pool "CREATE TABLE docs (id UUID PRIMARY KEY, user_id UUID NOT NULL, org_id UUID NOT NULL, body TEXT)"+ execSQL pool "ALTER TABLE docs ENABLE ROW LEVEL SECURITY"+ execSQL pool "CREATE POLICY docs_user ON docs USING (user_id = current_setting('rls.ihp_user_id')::uuid)"+ execSQL pool "CREATE POLICY docs_org ON docs USING (org_id = current_setting('rls.ihp_org_id')::uuid)"++ cols <- rlsPolicyColumns pool "docs"+ cols `shouldSatisfy` Set.member "user_id"+ cols `shouldSatisfy` Set.member "org_id"
+ Test/DataSync/TypedEncoder.hs view
@@ -0,0 +1,127 @@+{-|+Copyright: (c) digitally induced GmbH, 2025+-}+module Test.DataSync.TypedEncoder where++import Test.Hspec+import IHP.Prelude+import IHP.DataSync.DynamicQuery+import IHP.DataSync.TypedEncoder+import qualified Data.Aeson as Aeson+import qualified Data.Vector as Vector+import qualified Control.Exception as Exception+import qualified Hasql.DynamicStatements.Snippet as Snippet++tests = do+ describe "IHP.DataSync.TypedEncoder" do+ describe "typedValueParam" do+ it "encodes NULL regardless of type" do+ Snippet.toSql (typedValueParam Nothing Aeson.Null) `shouldBe` "NULL"+ Snippet.toSql (typedValueParam (Just "uuid") Aeson.Null) `shouldBe` "NULL"+ Snippet.toSql (typedValueParam (Just "text") Aeson.Null) `shouldBe` "NULL"++ it "encodes UUID with typed encoder" do+ Snippet.toSql (typedValueParam (Just "uuid") (Aeson.String "a5d7772f-c63f-4444-be69-dd9afd902e9b"))+ `shouldBe` "$1"++ it "encodes text with typed encoder" do+ Snippet.toSql (typedValueParam (Just "text") (Aeson.String "hello"))+ `shouldBe` "$1"++ it "encodes int4 with typed encoder" do+ Snippet.toSql (typedValueParam (Just "int4") (Aeson.Number 42))+ `shouldBe` "$1"++ it "encodes int8 with typed encoder" do+ Snippet.toSql (typedValueParam (Just "int8") (Aeson.Number 42))+ `shouldBe` "$1"++ it "encodes bool with typed encoder" do+ Snippet.toSql (typedValueParam (Just "bool") (Aeson.Bool True))+ `shouldBe` "$1"++ it "encodes float8 with typed encoder" do+ Snippet.toSql (typedValueParam (Just "float8") (Aeson.Number 3.14))+ `shouldBe` "$1"++ it "encodes custom enum with explicit cast" do+ Snippet.toSql (typedValueParam (Just "my_enum") (Aeson.String "active"))+ `shouldBe` "$1::\"my_enum\""++ it "errors when no type info is available" do+ Exception.evaluate (Snippet.toSql (typedValueParam Nothing (Aeson.String "hello")))+ `shouldThrow` anyErrorCall++ it "encodes Point with binary encoder" do+ let pointJson = Aeson.object ["x" Aeson..= (1.0 :: Double), "y" Aeson..= (2.0 :: Double)]+ Snippet.toSql (typedValueParam (Just "point") pointJson)+ `shouldBe` "$1"++ it "encodes Array with typed element encoding" do+ Snippet.toSql (typedValueParam (Just "_int4") (Aeson.Array (Vector.fromList [Aeson.Number 1, Aeson.Number 2])))+ `shouldBe` "ARRAY[$1, $2]"++ it "propagates array element type by stripping _ prefix" do+ Snippet.toSql (typedValueParam (Just "_uuid") (Aeson.Array (Vector.fromList [Aeson.String "a5d7772f-c63f-4444-be69-dd9afd902e9b"])))+ `shouldBe` "ARRAY[$1]"++ describe "typedAesonValueToSnippet" do+ it "encodes JSON null as NULL" do+ Snippet.toSql (typedAesonValueToSnippet Nothing Aeson.Null) `shouldBe` "NULL"+ Snippet.toSql (typedAesonValueToSnippet (Just "text") Aeson.Null) `shouldBe` "NULL"++ it "preserves JSONB values as-is" do+ Snippet.toSql (typedAesonValueToSnippet (Just "jsonb") (Aeson.object ["key" Aeson..= ("value" :: Text)]))+ `shouldBe` "$1"++ it "preserves JSON values as-is" do+ Snippet.toSql (typedAesonValueToSnippet (Just "json") (Aeson.object ["key" Aeson..= ("value" :: Text)]))+ `shouldBe` "$1"++ it "decodes {x, y} object as Point only when column type is point" do+ let pointJson = Aeson.object ["x" Aeson..= (1.0 :: Double), "y" Aeson..= (2.0 :: Double)]+ Snippet.toSql (typedAesonValueToSnippet (Just "point") pointJson)+ `shouldBe` "$1"++ it "does not decode {x, y} object as Point when column type is not point" do+ let pointJson = Aeson.object ["x" Aeson..= (1.0 :: Double), "y" Aeson..= (2.0 :: Double)]+ -- When column type is jsonb, the {x,y} object should be preserved as JSON+ Snippet.toSql (typedAesonValueToSnippet (Just "jsonb") pointJson)+ `shouldBe` "$1"++ it "encodes string values with typed encoder" do+ Snippet.toSql (typedAesonValueToSnippet (Just "text") (Aeson.String "hello"))+ `shouldBe` "$1"++ it "encodes number values with typed encoder" do+ Snippet.toSql (typedAesonValueToSnippet (Just "int4") (Aeson.Number 42))+ `shouldBe` "$1"++ it "encodes boolean values with typed encoder" do+ Snippet.toSql (typedAesonValueToSnippet (Just "bool") (Aeson.Bool True))+ `shouldBe` "$1"++ describe "dynamicValueParam" do+ it "encodes Number (integer) as native int parameter" do+ Snippet.toSql (dynamicValueParam (Aeson.Number 42))+ `shouldBe` "$1"++ it "encodes Number (double) as native float parameter" do+ Snippet.toSql (dynamicValueParam (Aeson.Number 3.14))+ `shouldBe` "$1"++ it "encodes String as native text parameter" do+ Snippet.toSql (dynamicValueParam (Aeson.String "hello"))+ `shouldBe` "$1"++ it "encodes Bool as native bool parameter" do+ Snippet.toSql (dynamicValueParam (Aeson.Bool True))+ `shouldBe` "$1"++ it "encodes Array with native element parameters" do+ Snippet.toSql (dynamicValueParam (Aeson.Array (Vector.fromList [Aeson.Number 1, Aeson.Number 2])))+ `shouldBe` "ARRAY[$1, $2]"++ it "encodes Null as SQL NULL" do+ Snippet.toSql (dynamicValueParam Aeson.Null)+ `shouldBe` "NULL"
+ Test/Main.hs view
@@ -0,0 +1,17 @@+module Main where++import Prelude+import Test.Hspec+import qualified Test.DataSync.DynamicQueryCompiler+import qualified Test.DataSync.TypedEncoder+import qualified Test.DataSync.ChangeNotifications+import qualified Test.DataSync.RLSIntegrationSpec+import qualified Test.DataSync.DataSyncIntegrationSpec++main :: IO ()+main = hspec do+ Test.DataSync.DynamicQueryCompiler.tests+ Test.DataSync.TypedEncoder.tests+ Test.DataSync.ChangeNotifications.tests+ Test.DataSync.RLSIntegrationSpec.tests+ Test.DataSync.DataSyncIntegrationSpec.tests
+ changelog.md view
@@ -0,0 +1,18 @@+# Changelog for `ihp-datasync`++## v1.5.0++- Migrate from postgresql-simple to hasql with prepared statements+- Use hasql for RLS queries instead of falling back to postgresql-simple+- Replace `DynamicValue` with Aeson `Value` for simpler JSON handling+- Send append-only deltas for streaming text updates+- Fix `CountSubscription` comparing against initial count instead of last-sent count+- Fix DataSync trigger installation deadlock+- Return HTTP 400 JSON responses instead of crashing on invalid `requestBodyJSON`+- Use `postgresql-types` (`Point`, `Polygon`, `Interval`, `Inet`) instead of custom types+- Schema-aware typed parameter encoding+- Add end-to-end integration tests against real PostgreSQL++## v1.4.0++- Initial release as a standalone package, extracted from ihp
+ ihp-datasync.cabal view
@@ -0,0 +1,154 @@+cabal-version: 2.2+name: ihp-datasync+version: 1.5.0+synopsis: IHP DataSync Framework+description: DataSync module for the Integrated Haskell Platform, enabling real-time data synchronization and RESTful API capabilities for IHP applications.+license: MIT+license-file: LICENSE+author: digitally induced GmbH+maintainer: hello@digitallyinduced.com+homepage: https://ihp.digitallyinduced.com/+bug-reports: https://github.com/digitallyinduced/ihp/issues+copyright: (c) digitally induced GmbH+category: Web, IHP, DataSync+stability: Stable+tested-with: GHC == 9.8.4+build-type: Simple+extra-source-files: changelog.md++Flag FastBuild+ Default: False+ Description: Disables all optimisations, leads to faster build time++common shared-properties+ default-language: GHC2021+ build-depends:+ base >= 4.17.0 && < 4.22+ , classy-prelude+ , mono-traversable+ , transformers+ , mtl+ , text+ , hasql+ , hasql-dynamic-statements+ , hasql-pool+ , hasql-transaction+ , bytestring+ , aeson+ , uuid+ , time+ , attoparsec+ , case-insensitive+ , haskell-src-meta+ , template-haskell+ , haskell-src-exts+ , interpolate+ , containers+ , http-media+ , wai+ , warp+ , websockets+ , wai-websockets+ , unliftio+ , async+ , ihp+ , ihp-log+ , ihp-hsx+ , deepseq+ , safe-exceptions+ , http-types+ , scientific+ , vector+ , stm+ , unordered-containers+ , vault+ , typerep-map+ , postgresql-types+ , hasql-mapping+ , hasql-postgresql-types+ default-extensions:+ OverloadedStrings+ , NoImplicitPrelude+ , ImplicitParams+ , DisambiguateRecordFields+ , DuplicateRecordFields+ , OverloadedLabels+ , DataKinds+ , QuasiQuotes+ , TypeFamilies+ , PackageImports+ , RecordWildCards+ , DefaultSignatures+ , FunctionalDependencies+ , PartialTypeSignatures+ , BlockArguments+ , LambdaCase+ , TemplateHaskell+ , OverloadedRecordDot+ , DeepSubsumption+ if flag(FastBuild)+ ghc-options:+ -Wunused-imports+ -Wunused-foralls+ -Werror=missing-fields+ -Winaccessible-code+ -Wmissed-specialisations+ -Wall-missed-specialisations+ -Wno-ambiguous-fields+ -Werror=incomplete-patterns+ -Werror=unused-imports+ else+ ghc-options:+ -fstatic-argument-transformation+ -funbox-strict-fields+ -haddock+ -Wunused-imports+ -Wunused-foralls+ -Werror=missing-fields+ -Winaccessible-code+ -Wmissed-specialisations+ -Wall-missed-specialisations+ -Wno-ambiguous-fields+ -Werror=incomplete-patterns+ -Werror=unused-imports++library+ import: shared-properties+ hs-source-dirs: .+ exposed-modules:+ IHP.DataSync.Types+ , IHP.DataSync.RowLevelSecurity+ , IHP.DataSync.DynamicQuery+ , IHP.DataSync.DynamicQueryCompiler+ , IHP.DataSync.TypedEncoder+ , IHP.DataSync.ChangeNotifications+ , IHP.DataSync.Controller+ , IHP.DataSync.ControllerImpl+ , IHP.DataSync.REST.Types+ , IHP.DataSync.REST.Routes+ , IHP.DataSync.REST.Controller+ , IHP.DataSync.Pool+ , IHP.DataSync.Role+ , IHP.DataSync.Hasql+ other-modules:+ Paths_ihp_datasync+ autogen-modules:+ Paths_ihp_datasync+++test-suite ihp-datasync-hspec+ import: shared-properties+ type: exitcode-stdio-1.0+ hs-source-dirs: .+ main-is: Test/Main.hs+ ghc-options: -threaded+ other-modules: Test.DataSync.DynamicQueryCompiler+ , Test.DataSync.TypedEncoder+ , Test.DataSync.ChangeNotifications+ , Test.DataSync.RLSIntegrationSpec+ , Test.DataSync.DataSyncIntegrationSpec+ build-depends: hspec++source-repository head+ type: git+ location: https://github.com/digitallyinduced/ihp