postgresql-replicant (empty) → 0.1.0.1

raw patch · 18 files changed

+1600/−0 lines, 18 filesdep +aesondep +asyncdep +attoparsecsetup-changed

Dependencies added: aeson, async, attoparsec, base, binary, bits, bytestring, cereal, containers, hspec, keep-alive, postgresql-libpq, postgresql-replicant, scientific, stm, text, time

Files

• ChangeLog.md +29/−0
• LICENSE +30/−0
• README.md +115/−0
• Setup.hs +2/−0
• app/Main.hs +24/−0
• postgresql-replicant.cabal +96/−0
• src/Database/PostgreSQL/Replicant.hs +119/−0
• src/Database/PostgreSQL/Replicant/Exception.hs +9/−0
• src/Database/PostgreSQL/Replicant/Message.hs +406/−0
• src/Database/PostgreSQL/Replicant/PostgresUtils.hs +19/−0
• src/Database/PostgreSQL/Replicant/Protocol.hs +234/−0
• src/Database/PostgreSQL/Replicant/Queue.hs +77/−0
• src/Database/PostgreSQL/Replicant/ReplicationSlot.hs +150/−0
• src/Database/PostgreSQL/Replicant/Serialize.hs +10/−0
• src/Database/PostgreSQL/Replicant/State.hs +44/−0
• src/Database/PostgreSQL/Replicant/Types/Lsn.hs +104/−0
• src/Database/PostgreSQL/Replicant/Util.hs +12/−0
• test/Spec.hs +120/−0

@@ -0,0 +1,29 @@+All notable changes to this project will be documented in this file.++The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).++## [Unreleased]++## [0.1.0.1] - 2021-06-22++This is a minor update to our experimental release, fixing some pretty+major issues:++### Fixed+- WalSender on the server hanging up because replicant wasn't sending+  updates+- Minor cleanup, throw better errors instead of printing debug+  information++## [0.1.0.0] - 2021-07-27++This is an experimental release to test out this library and get it+ready and polished for 1.0!++### Added+- Initial release!+- `withLogicalStream`+- Protocol types, serialzation instances, etc+- Logical Replication Messages+- Stream state handling

@@ -0,0 +1,30 @@+Copyright James King (c) 2020, 2021++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,115 @@+# postgres-replicant++A PostgreSQL streaming logical replication client for Haskell.++React to database changes in real-time.++This library is currently _EXPERIMENTAL_ and is not ready for+production yet.  Developers are encouraged to test out the library,+add issues, contribute and make it awesome.++## Setup++In order to use a logical replication client you need to set the+`wal_level` in Postgres to `logical`.  You can set that, for example,+in your `postgres.conf` file:++    wal_level = logical++Then restart your server, log in as `postgres` or another super user+and check:++    SHOW wal_level++It should show _logical_.++You will also need a user who is+[allowed](https://www.postgresql.org/docs/9.1/sql-alterrole.html+"PostgreSQL user role documentation") to use replication features.++Then add a database and your user with the `REPLICATION` trait to the+database, grant all on it if you need to.++We haven't added authorization support yet to the library so make sure+in `pg_hba.conf` you `trust` local connections:++    host    all             all             127.0.0.1/32            trust++Caveat emptor.++You will also need to install the `wal2json` plugin for your+PostgreSQL server.  For example in Ubuntu-like distributions you can:++    $ sudo apt install postgresql-12-wal2json++Assuming you have installed PostgreSQL from the package repositories.++You can then setup a basic "hello world" program using this library:++``` haskell+module Main where++import Control.Exception++import Database.PostgreSQL.Replicant++main :: IO ()+main = do+  let settings = PgSettings "my-user" "my-database" "localhost" "5432" "testing"+  withLogicalStream settings $ \change -> do+    print change+      `catch` \err ->+      print err+```++Which should connect, create a `testing` replication slot, and start+sending your callback the changes to your database as they arrive.++### WAL Output Plugins++Presently we only support the `wal2json` WAL output plugin and only+the v1 format.  Support for the v2 format is planned as are more+output plugins.++## Example Program++Included is a simple program, `replicant-example` which should help+you test your connection settings and make sure that you can connect+and receive WAL changes.++You can change the connection settings through environment variables:++    PG_USER=myuser PG_DATABASE=mydb replicant-example++The configuration settings are:++- `PG_USER`: The replication user to connect as+- `PG_DATABASE`: The database to connect to+- `PG_HOST`: The host name of the database to connect to+- `PG_PORT`: The port to connect on+- `PG_SLOTNAME`: The replication slot to create or connect to+- `PG_UPDATEDELAY`: The frequency (in _ms_) to update PostgreSQL on+  the stream status++## FAQ++### Why not use LISTEN/NOTIFY++You don't have to set up triggers for each table with logical+replication.  You also don't have the message size limitation.  That+limitation often forces clients to perform a query for each message to+fetch the data and can use more than one connection.  We only use one+connection.++### What are the trade offs?++This library uses logical replication slots.  If your program loses+the connection to the database and cannot reconnect the server will+hold onto WAL files until your program can reconnect and resume the+replication stream.  On a busy database, depending on your+configuration, this can eat up disk space quickly.++It's a good idea to monitor your replication slots.  If+`postgresql-replicant` cannot reconnect to the slot and resume the+stream you may need to have a strategy for dropping the slot and+recreating it once conditions return to normal.  Have a strategy.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,24 @@+module Main where++import Data.Maybe+import Control.Exception+import System.Environment++import Database.PostgreSQL.Replicant++main :: IO ()+main = do+  dbUser <- maybe "postgresql" id <$> lookupEnv "PG_USER"+  dbName <- maybe "postgresql" id <$> lookupEnv "PG_DATABASE"+  dbHost <- maybe "localhost" id <$> lookupEnv "PG_HOST"+  dbPort <- maybe "5432" id <$> lookupEnv "PG_PORT"+  dbSlot <- maybe "replicant_test" id <$> lookupEnv "PG_SLOTNAME"+  dbUpdateDelay <- maybe "3000" id <$> lookupEnv "PG_UPDATEDELAY"+  let settings = PgSettings dbUser dbName dbHost dbPort dbSlot dbUpdateDelay+  withLogicalStream settings $ \change -> do+    putStrLn "Change received!"+    print change+  `catch`+  \exc -> do+    putStrLn "Something bad happened: "+    print @SomeException exc

@@ -0,0 +1,96 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: b9e10bcacd11fef9dd3cf878a5223b111e9355f146fc121702189290e50c23ed++name:           postgresql-replicant+version:        0.1.0.1+synopsis:       PostgreSQL logical streaming replication library+description:    Please see the README on GitHub at <https://github.com/agentultra/postgres-replicant#readme>+category:       Experimental, Database+homepage:       https://github.com/agentultra/postgresql-replicant#readme+bug-reports:    https://github.com/agentultra/postgresql-replicant/issues+author:         James King+maintainer:     james@agentultra.com+copyright:      2020, 2021, James King+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    ChangeLog.md+    LICENSE++source-repository head+  type: git+  location: https://github.com/agentultra/postgresql-replicant++library+  exposed-modules:+      Database.PostgreSQL.Replicant+      Database.PostgreSQL.Replicant.Exception+      Database.PostgreSQL.Replicant.Message+      Database.PostgreSQL.Replicant.PostgresUtils+      Database.PostgreSQL.Replicant.Protocol+      Database.PostgreSQL.Replicant.Queue+      Database.PostgreSQL.Replicant.ReplicationSlot+      Database.PostgreSQL.Replicant.Serialize+      Database.PostgreSQL.Replicant.State+      Database.PostgreSQL.Replicant.Types.Lsn+      Database.PostgreSQL.Replicant.Util+  other-modules:+      Paths_postgresql_replicant+  hs-source-dirs:+      src+  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings TypeApplications+  build-depends:+      aeson+    , async+    , attoparsec+    , base >=4.7 && <5+    , bits+    , bytestring+    , cereal+    , containers+    , keep-alive+    , postgresql-libpq+    , scientific+    , stm+    , text+    , time+  default-language: Haskell2010++executable replicant-example+  main-is: Main.hs+  other-modules:+      Paths_postgresql_replicant+  hs-source-dirs:+      app+  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings TypeApplications+  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      base >=4.7 && <5+    , postgresql-libpq+    , postgresql-replicant+  default-language: Haskell2010++test-suite postgres-replicant-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Paths_postgresql_replicant+  hs-source-dirs:+      test+  default-extensions: DeriveGeneric GADTs LambdaCase OverloadedStrings TypeApplications+  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      base >=4.7 && <5+    , binary+    , bytestring+    , cereal+    , hspec+    , postgresql-replicant+  default-language: Haskell2010

@@ -0,0 +1,119 @@+{-# LANGUAGE RecordWildCards #-}++{-|+Module      : Database.PostgreSQL.Replicant+Description : A PostgreSQL streaming replication library+Copyright   : (c) James King, 2020, 2021+License     : BSD3+Maintainer  : james@agentultra.com+Stability   : experimental+Portability : POSIX++Connect to a PostgreSQL server as a logical replication client and+receive changes.++The basic API is this:++@+  withLogicalStream defaultSettings $ \change -> do+    print change+    `catch` \err -> do+      show err+@++This is a low-level library meant to give the primitives necessary to+library authors to add streaming replication support.  The API here to+rather simplistic but should be hooked up to something like conduit to+provide better ergonomics.+-}++module Database.PostgreSQL.Replicant+    ( withLogicalStream+    , PgSettings (..)+    ) where++import Control.Concurrent+import Control.Exception+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as B+import qualified Data.Text as T+import qualified Data.Text.Read as T+import Database.PostgreSQL.LibPQ+import Network.Socket.KeepAlive+import System.Posix.Types++import Database.PostgreSQL.Replicant.Exception+import Database.PostgreSQL.Replicant.Protocol+import Database.PostgreSQL.Replicant.Message+import Database.PostgreSQL.Replicant.ReplicationSlot+import Database.PostgreSQL.Replicant.Util++data PgSettings+  = PgSettings+  { pgUser        :: String+  , pgDbName      :: String+  , pgHost        :: String+  , pgPort        :: String+  , pgSlotName    :: String+  , pgUpdateDelay :: String -- ^ Controls how frequently the+                            -- primaryKeepAlive thread updates+                            -- PostgresSQL in @ms@+  }+  deriving (Eq, Show)++pgConnectionString :: PgSettings -> ByteString+pgConnectionString PgSettings {..} = B.intercalate " "+  [ "user=" <> (B.pack pgUser)+  , "dbname=" <> (B.pack pgDbName)+  , "host=" <> (B.pack pgHost)+  , "port=" <> (B.pack pgPort)+  , "replication=database"+  ]++-- | Connect to a PostgreSQL database as a user with the replication+-- attribute and start receiving changes using the logical replication+-- protocol.  Logical replication happens at the query level so the+-- changes you get represent the set of queries in a transaction:+-- /insert/, /update/, and /delete/.+--+-- This function will create the replication slot, if it doesn't+-- exist, or reconnect to it otherwise and restart the stream from+-- where the replication slot left off.+--+-- This function can throw exceptions in @IO@ and shut-down the+-- socket in case of any error.+withLogicalStream :: PgSettings -> (Change -> IO a) -> IO ()+withLogicalStream settings cb = do+  conn <- connectStart $ pgConnectionString settings+  mFd <- socket conn+  sockFd <- maybeThrow (ReplicantException "withLogicalStream: could not get socket fd") mFd+  pollResult <- pollConnectStart conn sockFd+  let updateFreq = getUpdateDelay settings+  case pollResult of+    PollingFailed -> throwIO $ ReplicantException "withLogicalStream: Unable to connect to the database"+    PollingOk -> do+      maybeInfo <- identifySystemSync conn+      _ <- maybeThrow (ReplicantException "withLogicalStream: could not get system information") maybeInfo+      repSlot <- setupReplicationSlot conn $ B.pack . pgSlotName $ settings+      startReplicationStream conn (slotName repSlot) (slotRestart repSlot) updateFreq cb+      pure ()+  where+    pollConnectStart :: Connection -> Fd -> IO PollingStatus+    pollConnectStart conn fd@(Fd cint) = do+      pollStatus <- connectPoll conn+      case pollStatus of+        PollingReading -> do+          threadWaitRead fd+          pollConnectStart conn fd+        PollingWriting -> do+          threadWaitWrite fd+          pollConnectStart conn fd+        PollingOk -> do+          _ <- setKeepAlive cint $ KeepAlive True 60 2+          pure PollingOk+        PollingFailed -> pure PollingFailed+    getUpdateDelay :: PgSettings -> Int+    getUpdateDelay PgSettings {..} =+      case T.decimal . T.pack $ pgUpdateDelay of+        Left _ -> 3000+        Right (i, _) -> i

@@ -0,0 +1,9 @@+module Database.PostgreSQL.Replicant.Exception where++import Control.Exception+import Data.Typeable++newtype ReplicantException = ReplicantException String+  deriving (Show, Typeable)++instance Exception ReplicantException

@@ -0,0 +1,406 @@+{-|+Module      : Database.PostgreSQL.Replicant.Message+Description : Streaming replication message types+Copyright   : (c) James King, 2020, 2021+License     : BSD3+Maintainer  : james@agentultra.com+Stability   : experimental+Portability : POSIX++This module contains the binary protocols messages used in the+streaming replication protocol as well as the messages used in the+body of the logical stream messages.+-}+module Database.PostgreSQL.Replicant.Message where++import Control.Monad+import Data.Aeson+import Data.ByteString (ByteString)+import Data.Scientific (Scientific)+import Data.Serialize+import Data.Text (Text)+import GHC.Generics+import GHC.Int++import Database.PostgreSQL.Replicant.Serialize+import Database.PostgreSQL.Replicant.Types.Lsn++-- WAL Replication Stream messages++-- | Indicates whether the server or the client should respond to the+-- message.+data ResponseExpectation+  = ShouldRespond+  | DoNotRespond+  deriving (Eq, Generic, Show)++instance Serialize ResponseExpectation where+  put ShouldRespond = putWord8 1+  put DoNotRespond  = putWord8 0++  get = do+    responseFlag <- getWord8+    case responseFlag of+      0 -> pure DoNotRespond+      1 -> pure ShouldRespond+      _ -> fail "Unrecognized response expectation flag"++-- | The Postgres WAL sender thread may periodically send these+-- messages.  When the server expects the client to respond it updates+-- its internal state of the client based on the response.  Failure to+-- respond results in the dreaded "WAL timeout" error.+--+-- See StandbyStatusUpdate for the message the client should respond+-- with.+data PrimaryKeepAlive+  = PrimaryKeepAlive+  { primaryKeepAliveWalEnd              :: !Int64+  , primaryKeepAliveSendTime            :: !Int64+  , primaryKeepAliveResponseExpectation :: !ResponseExpectation+  }+  deriving (Eq, Generic, Show)++instance Serialize PrimaryKeepAlive where+  put (PrimaryKeepAlive walEnd sendTime responseExpectation) = do+    putWord8 0x6B -- 'k'+    putInt64be walEnd+    putInt64be sendTime+    put responseExpectation++  get = do+    _ <- getBytes 1+    walEnd <- getInt64be+    sendTime <- getInt64be+    responseExpectation <- get+    pure $ PrimaryKeepAlive walEnd sendTime responseExpectation++-- | Sent by the client.  Can be sent periodically or in response to a+-- PrimaryKeepAlive message from the server.  Gives the server+-- information about the client stream state.+data StandbyStatusUpdate+  = StandbyStatusUpdate+  { standbyStatuUpdateLastWalByteReceived  :: !LSN+  , standbyStatusUpdateLastWalByteFlushed  :: !LSN+  , standbyStatusUpdateLastWalByteApplied  :: !LSN+  , standbyStatusUpdateSendTime            :: !Int64+  , standbyStatusUpdateResponseExpectation :: !ResponseExpectation+  }+  deriving (Eq, Generic, Show)++instance Serialize StandbyStatusUpdate where+  put (StandbyStatusUpdate+       walReceived+       walFlushed+       walApplied+       sendTime+       responseExpectation) = do+    putWord8 0x72 -- 'r'+    put walReceived+    put walFlushed+    put walApplied+    putInt64be sendTime+    put responseExpectation++  get = do+    _ <- getBytes 1 -- should expect 0x72, 'r'+    walReceived         <- get+    walFlushed          <- get+    walApplied          <- get+    sendTime            <- getInt64be+    responseExpectation <- get+    pure $+      StandbyStatusUpdate+      walReceived+      walFlushed+      walApplied+      sendTime+      responseExpectation++-- | Carries WAL segments in the streaming protocol.+data XLogData+  = XLogData+  { xLogDataWalStart :: !LSN+  , xLogDataWalEnd   :: !LSN+  , xLogDataSendTime :: !Int64+  , xLogDataWalData  :: !ByteString+  }+  deriving (Eq, Generic, Show)++instance Serialize XLogData where+  put (XLogData walStart walEnd sendTime walData) = do+    putWord8 0x77 -- 'w'+    put walStart+    put walEnd+    putInt64be sendTime+    putByteString walData++  get = do+    _ <- getBytes 1 -- should expec '0x77', 'w'+    walStart <- get+    walEnd   <- get+    sendTime <- getInt64be+    walData  <- consumeByteStringToEnd+    pure $ XLogData walStart walEnd sendTime walData++-- | Not used yet but enables streaming in hot-standby mode.+data HotStandbyFeedback+  = HotStandbyFeedback+  { hotStandbyFeedbackClientSendTime :: !Int64+  , hotStandbyFeedbackCurrentXmin    :: !Int32+  , hotStandbyFeedbackCurrentEpoch   :: !Int32+  }+  deriving (Eq, Generic, Show)++instance Serialize HotStandbyFeedback where+  put (HotStandbyFeedback clientSendTime currentXMin currentEpoch) = do+    putWord8 0x68+    putInt64be clientSendTime+    putInt32be currentXMin+    putInt32be currentEpoch++  get = do+    _ <- getBytes 1 -- should expect '0x68' 'h'+    clientSendTime <- getInt64be+    currentXmin <- getInt32be+    currentEpoch <- getInt32be+    pure $ HotStandbyFeedback clientSendTime currentXmin currentEpoch++-- | This structure wraps the two messages sent by the server so that+-- we get a Serialize instance for both.+data WalCopyData+  = XLogDataM !XLogData+  | KeepAliveM !PrimaryKeepAlive+  deriving (Eq, Generic, Show)++instance Serialize WalCopyData where+  put (XLogDataM xLogData)   = put xLogData+  put (KeepAliveM keepAlive) = put keepAlive+  get = do+    messageTypeFlag <- lookAhead $ getWord8+    case messageTypeFlag of+      -- 'w' XLogData+      0x77 -> do+        xLogData <- get+        pure $ XLogDataM xLogData+      -- 'k' PrimaryKeepAlive+      0x6B -> do+        keepAlive <- get+        pure $ KeepAliveM keepAlive+      _    -> fail "Unrecognized WalCopyData"++-- WAL Log Data++-- | Wraps Postgres values.  Since this library currently supports+-- only the @wal2json@ logical decoder plugin we have the JSON values+-- much like Aeson does.+data WalValue+  = WalString !Text+  | WalNumber !Scientific+  | WalBool !Bool+  | WalNull+  deriving (Eq, Generic, Show)++instance FromJSON WalValue where+  parseJSON (String txt) = pure $ WalString txt+  parseJSON (Number num) = pure $ WalNumber num+  parseJSON (Bool bool)  = pure $ WalBool bool+  parseJSON Null         = pure WalNull+  parseJSON _            = fail "Unrecognized WalValue"++instance ToJSON WalValue where+  toJSON (WalString txt) = String txt+  toJSON (WalNumber n)   = Number n+  toJSON (WalBool b)     = Bool b+  toJSON (WalNull)       = Null++-- | Represents a single table column.  We only support the `wal2json`+-- logical decoder plugin and make no attempt to parse anything but+-- JSON-like primitives.+data Column+  = Column+  { columnName  :: !Text+  , columnType  :: !Text+  , columnValue :: !WalValue+  }+  deriving (Eq, Show)++data ColumnParseError+  = ColumnLengthMatchError+  deriving (Eq, Show)++-- | Some WAL output plugins encode column values in three, equal+-- length, heterogeneous lists.+columns :: [Text] -> [Text] -> [WalValue] -> Either ColumnParseError [Column]+columns colNames colTypes colValues+  | length colNames == length colTypes && length colTypes == length colValues+  = Right (map toColumn $ zip3 colNames colTypes colValues)+  | otherwise = Left ColumnLengthMatchError+  where+    toColumn (n, t, v) = Column n t v++fromColumn :: Column -> (Text, Text, WalValue)+fromColumn (Column cName cType cValue) = (cName, cType, cValue)++fromColumns :: [Column] -> ([Text], [Text], [WalValue])+fromColumns = unzip3 . map fromColumn++-- | Represents a single insert query in the logical replication+-- format.+data Insert+  = Insert+  { insertSchema  :: !String+  , insertTable   :: !String+  , insertColumns :: ![Column]+  }+  deriving (Eq, Show)++instance FromJSON Insert where+  parseJSON = withObject "Insert" $ \o -> do+    kind         <- o .: "kind"+    unless (kind == String "insert") $ fail "Invalid insert"+    schema       <- o .: "schema"+    table        <- o .: "table"+    columnNames  <- o .: "columnnames"+    columnTypes  <- o .: "columntypes"+    columnValues <- o .: "columnvalues"+    case columns columnNames columnTypes columnValues of+      Left err      -> fail $ show err+      Right columns -> pure $ Insert schema table columns++instance ToJSON Insert where+  toJSON (Insert schema table columns) =+    let (cNames, cTypes, cValues) = fromColumns columns+    in object [ "kind"         .= String "insert"+              , "schema"       .= schema+              , "table"        .= table+              , "columnnames"  .= cNames+              , "columntypes"  .= cTypes+              , "columnvalues" .= cValues+              ]++-- | Represents a single update query in the logical replication+-- format.+data Update+  = Update+  { updateSchema  :: !Text+  , updateTable   :: !Text+  , updateColumns :: ![Column]+  }+  deriving (Eq, Show)++instance FromJSON Update where+  parseJSON = withObject "Insert" $ \o -> do+    kind         <- o .: "kind"+    unless (kind == String "update") $ fail "Invalid update"+    schema       <- o .: "schema"+    table        <- o .: "table"+    columnNames  <- o .: "columnnames"+    columnTypes  <- o .: "columntypes"+    columnValues <- o .: "columnvalues"+    case columns columnNames columnTypes columnValues of+      Left err      -> fail $ show err+      Right columns -> pure $ Update schema table columns++instance ToJSON Update where+  toJSON (Update schema table columns) =+    let (cNames, cTypes, cValues) = fromColumns columns+    in object [ "kind"         .= String "update"+              , "schema"       .= schema+              , "table"        .= table+              , "columnnames"  .= cNames+              , "columntypes"  .= cTypes+              , "columnvalues" .= cValues+              ]++-- | Represents a single delete query in the logical replication format+data Delete+  = Delete+  { deleteSchema  :: !Text+  , deleteTable   :: !Text+  , deleteColumns :: ![Column]+  }+  deriving (Eq, Show)++instance FromJSON Delete where+  parseJSON = withObject "Delete" $ \o -> do+    kind      <- o .: "kind"+    unless (kind == String "delete") $ fail "Invalid delete"+    schema    <- o .: "schema"+    table     <- o .: "table"+    oldKeys   <- o .: "oldkeys"+    keyNames  <- oldKeys .: "keynames"+    keyTypes  <- oldKeys .: "keytypes"+    keyValues <- oldKeys .: "keyvalues"+    case columns keyNames keyTypes keyValues of+      Left err      -> fail $ show err+      Right columns -> pure $ Delete schema table columns++instance ToJSON Delete where+  toJSON (Delete schema table columns) =+    let (cNames, cTypes, cValues) = fromColumns columns+    in object [ "kind"    .= String "delete"+              , "schema"  .= schema+              , "table"   .= table+              , "oldkeys" .= object+                [ "keynames"  .= cNames+                , "keytypes"  .= cTypes+                , "keyvalues" .= cValues+                ]+              ]++-- | Occasionally the server may also send these for informational+-- purposes and can be ignored.  May be used internally.+data Message+  = Message+  { messageTransactional :: !Bool+  , messagePrefix        :: !Text+  , messageContent       :: !Text+  }+  deriving (Eq, Show)++instance FromJSON Message where+  parseJSON = withObject "Message" $ \o -> do+    kind          <- o .: "kind"+    unless (kind == String "message") $ fail "Invalid message"+    transactional <- o .: "transactional"+    prefix        <- o .: "prefix"+    content       <- o .: "content"+    pure $ Message transactional prefix content++instance ToJSON Message where+  toJSON (Message transactional prefix content)+    = object+    [ "kind"          .= String "message"+    , "transactional" .= transactional+    , "prefix"        .= prefix+    , "content"       .= content+    ]++data WalLogData+  = WInsert !Insert+  | WUpdate !Update+  | WDelete !Delete+  | WMessage !Message+  deriving (Eq, Generic, Show)++instance ToJSON WalLogData where+  toJSON = genericToJSON defaultOptions { sumEncoding = UntaggedValue }++instance FromJSON WalLogData where+  parseJSON = genericParseJSON defaultOptions { sumEncoding = UntaggedValue }++data Change+  = Change+  { changeNextLSN :: LSN+  , changeDeltas  :: [WalLogData]+  }+  deriving (Eq, Generic, Show)++instance ToJSON Change where+  toJSON = genericToJSON defaultOptions++instance FromJSON Change where+  parseJSON = withObject "Change" $ \o -> do+    nextLSN <- o .: "nextlsn"+    deltas  <- o .: "change"+    pure $ Change nextLSN deltas

@@ -0,0 +1,19 @@+module Database.PostgreSQL.Replicant.PostgresUtils where++import Data.Fixed+import Data.Time+import GHC.Int++postgresEpoch :: IO Int64+postgresEpoch = do+  let epoch = mkUTCTime (2000, 1, 1) (0, 0, 0)+  now <- getCurrentTime+  pure $ round . (* 1000000) . nominalDiffTimeToSeconds $ diffUTCTime now epoch++-- From https://www.williamyaoh.com/posts/2019-09-16-time-cheatsheet.html+mkUTCTime :: (Integer, Int, Int)+          -> (Int, Int, Pico)+          -> UTCTime+mkUTCTime (year, mon, day) (hours, mins, secs) =+  UTCTime (fromGregorian year mon day)+          (timeOfDayToTime (TimeOfDay hours mins secs))

@@ -0,0 +1,234 @@+{-|+Module      : Database.PostgreSQL.Replicant.Protocol+Description : Streaming replication protocol+Copyright   : (c) James King, 2020, 2021+License     : BSD3+Maintainer  : james@agentultra.com+Stability   : experimental+Portability : POSIX++This module implements the Postgres streaming replication protocol.++See: https://www.postgresql.org/docs/9.5/protocol-replication.html+-}+module Database.PostgreSQL.Replicant.Protocol where++import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception.Base+import Control.Monad (forever)+import Data.Aeson (eitherDecode')+import Data.ByteString (ByteString)+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Char8 as B+import Data.Maybe+import Data.Serialize hiding (flush)+import Database.PostgreSQL.LibPQ++import Database.PostgreSQL.Replicant.Exception+import Database.PostgreSQL.Replicant.Message+import Database.PostgreSQL.Replicant.PostgresUtils+import Database.PostgreSQL.Replicant.State+import Database.PostgreSQL.Replicant.Types.Lsn++-- | The information returned by the @IDENTIFY_SYSTEM@ command+-- establishes the stream's log start, position, and information about+-- the database.+data IdentifySystem+  = IdentifySystem+  { identifySystemSytemId  :: ByteString+  , identifySystemTimeline :: ByteString+  , identifySystemLogPos   :: LSN+  , identifySystemDbName   :: Maybe ByteString+  }+  deriving (Eq, Show)++identifySystemCommand :: ByteString+identifySystemCommand = "IDENTIFY_SYSTEM"++-- | Synchronously execute the @IDENTIFY SYSTEM@ command which returns+-- some basic system information about the server.+identifySystemSync :: Connection -> IO (Maybe IdentifySystem)+identifySystemSync conn = do+  result <- exec conn identifySystemCommand+  case result of+    Just r -> do+      resultStatus <- resultStatus r+      case resultStatus of+        TuplesOk -> do+          systemId <- getvalue' r (toRow 0) (toColumn 0)+          timeline <- getvalue' r (toRow 0) (toColumn 1)+          logpos   <- getvalue' r (toRow 0) (toColumn 2)+          dbname   <- getvalue' r (toRow 0) (toColumn 3)+          case (systemId, timeline, logpos, dbname) of+            (Just s, Just t, Just l, d) -> do+              case fromByteString l of+                Left _ -> pure Nothing+                Right logPosLsn -> do+                  pure $ Just (IdentifySystem s t logPosLsn d)+            _ -> pure Nothing+        _ -> do+          err <- fromMaybe "identifySystemSync: unknown error" <$> errorMessage conn+          throwIO $ ReplicantException (B.unpack err)+    _ -> do+      err <- fromMaybe "identifySystemSync: unknown error" <$> errorMessage conn+      throwIO $ ReplicantException (B.unpack err)++-- | Create a @START_REPLICATION_SLOT@ query, escaping the slot name+-- passed in by the user.+startReplicationCommand :: Connection -> ByteString -> LSN -> IO ByteString+startReplicationCommand conn slotName systemLogPos = do+  escapedName <- escapeIdentifier conn slotName+  case escapedName of+    Nothing -> throwIO $ ReplicantException $ "Invalid slot name: " ++ show slotName+    Just escaped ->+      pure $+      B.intercalate+      ""+      [ "START_REPLICATION SLOT "+      , escaped+      , " LOGICAL "+      , (toByteString systemLogPos)+      , " (\"include-lsn\" 'on')"+      ]++-- | This handles the COPY OUT mode messages.  PostgreSQL uses this+-- mode to copy the data from a WAL log file to the socket in the+-- streaming replication protocol.+handleCopyOutData+  :: TChan PrimaryKeepAlive+  -> WalProgressState+  -> Connection+  -> (Change -> IO a)+  -> IO ()+handleCopyOutData chan walState conn cb = forever $ do+  d <- getCopyData conn False+  case d of+    CopyOutRow row -> handleReplicationRow chan walState conn row cb+    CopyOutError   -> handleReplicationError conn+    _              -> handleReplicationNoop++handleReplicationRow+  :: TChan PrimaryKeepAlive+  -> WalProgressState+  -> Connection+  -> ByteString+  -> (Change -> IO a)+  -> IO ()+handleReplicationRow keepAliveChan walState _ row cb =+  case decode @WalCopyData row of+    Left err ->+      throwIO+      $ ReplicantException+      $ "handleReplicationRow (decode error): " ++ err+    Right m  -> case m of+      XLogDataM xlog -> do+        case eitherDecode' @Change $ BL.fromStrict $ xLogDataWalData xlog of+          Left err ->+            throwIO+            $ ReplicantException+            $ "handleReplicationRow (parse error): " ++ err+          Right walLogData -> do+            _ <- updateWalProgress walState (changeNextLSN walLogData)+            _ <- cb walLogData+            pure ()+      KeepAliveM keepAlive -> atomically $ writeTChan keepAliveChan keepAlive++-- | Used to re-throw an exception received from the server.+handleReplicationError :: Connection -> IO ()+handleReplicationError conn = do+  err <- errorMessage conn+  throwIO (ReplicantException $ B.unpack . fromMaybe "Unknown error" $ err)+  pure ()++handleReplicationNoop :: IO ()+handleReplicationNoop = pure ()++-- | Initiate the streaming replication protocol handler.  This will+-- race the /keep-alive/ and /copy data/ handler threads.  It will+-- catch and rethrow exceptions from either thread if any fails or+-- returns.+startReplicationStream :: Connection -> ByteString -> LSN -> Int -> (Change -> IO a) -> IO ()+startReplicationStream conn slotName systemLogPos _ cb = do+  let initialWalProgress = WalProgress systemLogPos systemLogPos systemLogPos+  walProgressState <- WalProgressState <$> newMVar initialWalProgress+  replicationCommandQuery <- startReplicationCommand conn slotName systemLogPos+  result <- exec conn replicationCommandQuery+  case result of+    Nothing -> do+      err <- fromMaybe "startReplicationStream: unknown error starting stream"+        <$> errorMessage conn+      throwIO $ ReplicantException $ "startReplicationStream: " ++ B.unpack err+    Just r  -> do+      status <- resultStatus r+      case status of+        CopyBoth -> do+          keepAliveChan <- atomically newTChan+          race+            (keepAliveHandler conn keepAliveChan walProgressState)+            (handleCopyOutData keepAliveChan walProgressState conn cb)+            `catch`+            \exc -> do+              finish conn+              throwIO @SomeException exc+          return ()+        _ -> do+          err <- fromMaybe "startReplicationStream: unknown error entering COPY mode" <$> errorMessage conn+          throwIO $ ReplicantException $ B.unpack err++-- | This listens on the channel for /primary keep-alive messages/+-- from the server and responds to them with the /update status/+-- message using the current WAL stream state.  It will attempt to+-- buffer prior update messages when the socket is blocked.+keepAliveHandler :: Connection -> TChan PrimaryKeepAlive -> WalProgressState -> IO ()+keepAliveHandler conn msgs walProgressState = forever $ do+  mKeepAlive <- atomically $ tryReadTChan msgs+  case mKeepAlive of+    Nothing -> do+      sendStatusUpdate conn walProgressState+      threadDelay 3000000+    Just keepAlive' -> do+      case primaryKeepAliveResponseExpectation keepAlive' of+        DoNotRespond -> do+          threadDelay 1000+        ShouldRespond -> do+          sendStatusUpdate conn walProgressState++sendStatusUpdate+  :: Connection+  -> WalProgressState+  -> IO ()+sendStatusUpdate conn w@(WalProgressState walState) = do+  (WalProgress received flushed applied) <- readMVar walState+  timestamp <- postgresEpoch+  let statusUpdate =+        StandbyStatusUpdate+        received+        flushed+        applied+        timestamp+        DoNotRespond+  copyResult <- putCopyData conn $ encode statusUpdate+  case copyResult of+    CopyInOk -> do+      flushResult <- flush conn+      case flushResult of+        FlushOk -> pure ()+        FlushFailed -> do+          err <- fromMaybe "sendStatusUpdate: error flushing message to server" <$> errorMessage conn+          throwIO $ ReplicantException $ B.unpack err+        FlushWriting -> tryAgain conn w+    CopyInError -> do+      err <- fromMaybe "sendStatusUpdate: unknown error sending COPY IN" <$> errorMessage conn+      throwIO $ ReplicantException $ B.unpack err+    CopyInWouldBlock -> tryAgain conn w+  where+    tryAgain c ws = do+      mSockFd <- socket c+      case mSockFd of+        Nothing ->+          throwIO $ ReplicantException "sendStatusUpdate: failed to get socket fd"+        Just sockFd -> do+          threadWaitWrite sockFd+          sendStatusUpdate conn ws

@@ -0,0 +1,77 @@+{-|+Module      : Database.PostgreSQL.Replicant.Queue+Description : Bounded and unbounded FIFO queues+Copyright   : (c) James King, 2020, 2021+License     : BSD3+Maintainer  : james@agentultra.com+Stability   : experimental+Portability : POSIX++Shared FIFO queues+-}+module Database.PostgreSQL.Replicant.Queue where++import Control.Concurrent.MVar+import Data.Sequence (Seq, ViewR (..), (<|), (|>))+import qualified Data.Sequence as S++data BoundedFifoQueueMeta a+  = BoundedFifoQueueMeta+  { boundedFifoQueueSize :: Int+  , boundedFifoQueue     :: Seq a+  }+  deriving (Eq, Show)++newtype BoundedFifoQueue a = BoundedFifoQueue (MVar (BoundedFifoQueueMeta a))++newtype BoundedQueueException a+  = BoundedQueueOverflow a+  deriving (Eq, Show)++emptyBounded :: Int -> IO (BoundedFifoQueue a)+emptyBounded size =+  BoundedFifoQueue <$> newMVar (BoundedFifoQueueMeta size S.empty)++enqueueBounded :: BoundedFifoQueue a -> a -> IO (Either (BoundedQueueException a) ())+enqueueBounded (BoundedFifoQueue mQueue) x = do+  b@(BoundedFifoQueueMeta size queue) <- takeMVar mQueue+  if size == S.length queue+    then pure $ Left $ BoundedQueueOverflow x+    else do+    putMVar mQueue $ b { boundedFifoQueue = x <| queue }+    pure $ Right ()++newtype FifoQueue a = FifoQueue (MVar (Seq a))++empty :: IO (FifoQueue a)+empty = FifoQueue <$> newMVar S.empty++-- | Return @True@ if the queue is empty+null :: FifoQueue a -> IO Bool+null (FifoQueue mQueue) = do+  queue <- readMVar mQueue+  pure $ S.null queue++-- | Remove an item from the end of the non-empty queue.+dequeue :: FifoQueue a -> IO (Maybe a)+dequeue (FifoQueue mQueue) = do+  queue <- takeMVar mQueue+  case S.viewr queue of+    S.EmptyR -> do+      putMVar mQueue queue+      pure Nothing+    rest :> x -> do+      putMVar mQueue rest+      pure $ Just x++-- | Put an item on the front of the queue.+enqueue :: FifoQueue a -> a -> IO ()+enqueue (FifoQueue mQueue) x = do+  queue <- takeMVar mQueue+  putMVar mQueue $ x <| queue++-- | Put an item on the end of the queue so that it will be dequeued first.+enqueueRight :: FifoQueue a -> a -> IO ()+enqueueRight (FifoQueue mQueue) x = do+  queue <- takeMVar mQueue+  putMVar mQueue $ queue |> x

@@ -0,0 +1,150 @@+{-|+Module      : Database.PostgreSQL.Replicant.ReplicationSlot+Description : Replication slot query commands+Copyright   : (c) James King, 2020, 2021+License     : BSD3+Maintainer  : james@agentultra.com+Stability   : experimental+Portability : POSIX++This module contains the PostgreSQL queries, types, and functions for+working with querying, creating, and working with replication slots.+-}+module Database.PostgreSQL.Replicant.ReplicationSlot where++import Control.Exception+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import Database.PostgreSQL.LibPQ++import Database.PostgreSQL.Replicant.Exception+import Database.PostgreSQL.Replicant.Types.Lsn++data ReplicationSlotInfo+  = ReplicationSlotInfo+  { slotName    :: ByteString+  , slotPlugin  :: ByteString+  , slotType    :: ReplicationSlotType+  , slotActive  :: ReplicationSlotActive+  , slotRestart :: LSN+  }+  deriving (Eq, Show)++data ReplicationSlotType = Logical | Physical | UnknownSlotType+  deriving (Eq, Show)++parseSlotType :: ByteString -> ReplicationSlotType+parseSlotType "logical"  = Logical+parseSlotType "physical" = Physical+parseSlotType _          = UnknownSlotType++data ReplicationSlotActive = Active | Inactive+  deriving (Eq, Show)++parseSlotActive :: ByteString -> ReplicationSlotActive+parseSlotActive "t" = Active+parseSlotActive "f" = Inactive+parseSlotActive _   = Inactive++createReplicationSlotCommand :: Connection -> ByteString -> IO ByteString+createReplicationSlotCommand conn slotName = do+  escapedName <- escapeIdentifier conn slotName+  case escapedName of+    Nothing -> throwIO $ ReplicantException $ "Invalid slot name: " ++ show slotName+    Just escaped ->+      pure $+      B.intercalate+      ""+      [ "CREATE_REPLICATION_SLOT"+      , escaped+      , "LOGICAL wal2json"+      ]++-- | Create a replication slot using synchronous query execution.+--+-- May throw an exception if the command fails.+createReplicationSlotSync :: Connection -> ByteString -> IO ReplicationSlotInfo+createReplicationSlotSync conn slotName = do+  createReplicationSlotQuery <- createReplicationSlotCommand conn slotName+  result <- exec conn createReplicationSlotQuery+  case result of+    Just r -> do+      resultStatus <- resultStatus r+      case resultStatus of+        TuplesOk -> do+          sName           <- getvalue' r (toRow 0) (toColumn 0)+          consistentPoint <- getvalue' r (toRow 0) (toColumn 1)+          outputPlugin    <- getvalue' r (toRow 0) (toColumn 3)+          case (sName, consistentPoint, outputPlugin) of+            (Just s, Just c, Just op) ->+              case fromByteString c of+                Left _ -> throwIO $ ReplicantException "createReplicationSlotSync: invalid LSN detected"+                Right lsn -> pure $ ReplicationSlotInfo s op Logical Active lsn+            _ -> do+              err <- maybe "createReplicationSlotSync: unknown error" id <$> errorMessage conn+              throwIO $ ReplicantException (B8.unpack err)+        _ -> do+          err <- maybe "createReplicationSlotSync: unknown error" id <$> errorMessage conn+          throwIO $ ReplicantException (B8.unpack err)+    _ -> do+      err <- maybe "createReplicationSlotSync: unknown error" id <$> errorMessage conn+      throwIO $ ReplicantException (B8.unpack err)++getReplicationSlotInfoCommand :: Connection -> ByteString -> IO ByteString+getReplicationSlotInfoCommand conn slotName = do+  escapedName <- escapeStringConn conn slotName+  case escapedName of+    Nothing -> throwIO $ ReplicantException $ "Invalid slot name: " ++ show slotName+    Just escaped ->+      pure $+      B.intercalate+      ""+      [ "select slot_name, plugin, slot_type, active, restart_lsn from pg_replication_slots where slot_name = '"+      , escaped+      , "';"+      ]++-- | Get information about an existing replication slot.  Returns+-- @Nothing@ when the requested slot cannot be found.+--+-- May throw an exception if the command query fails.+getReplicationSlotSync :: Connection -> ByteString -> IO (Maybe ReplicationSlotInfo)+getReplicationSlotSync conn slotName = do+  replicationSlotInfoQuery <- getReplicationSlotInfoCommand conn slotName+  result <- exec conn replicationSlotInfoQuery+  case result of+    Just r -> do+      resultStatus <- resultStatus r+      case resultStatus of+        TuplesOk -> do+          nRows <- ntuples r+          if nRows == 0+            then pure Nothing+            else do+            slotName    <- getvalue' r (toRow 0) (toColumn 0)+            slotPlugin  <- getvalue' r (toRow 0) (toColumn 1)+            slotType    <- getvalue' r (toRow 0) (toColumn 2)+            slotActive  <- getvalue' r (toRow 0) (toColumn 3)+            slotRestart <- getvalue' r (toRow 0) (toColumn 4)+            case (slotName, slotPlugin, slotType, slotActive, slotRestart) of+              (Just n, Just p, Just t, Just a, Just restart) -> do+                case fromByteString restart of+                  Left _ -> pure Nothing -- TODO: this shouldn't happen...+                  Right lsn -> pure $ Just $ ReplicationSlotInfo n p (parseSlotType t) (parseSlotActive a) lsn+              _ ->  pure Nothing+        _ -> pure Nothing+    _ -> do+      err <- maybe "getReplicationSlotSync: unknown error" id <$> errorMessage conn+      throwIO $ ReplicantException (B8.unpack err)++-- | Create replication slot or retrieve an existing slot.+--+-- Can throw exceptions from @getReplicationSlotSync@ or+-- @createReplicationSlotSync@.+setupReplicationSlot :: Connection -> ByteString -> IO ReplicationSlotInfo+setupReplicationSlot conn slotName = do+  maybeSlot <- getReplicationSlotSync conn slotName+  case maybeSlot of+    Just slot -> pure $ slot+    Nothing   -> createReplicationSlotSync conn slotName

@@ -0,0 +1,10 @@+module Database.PostgreSQL.Replicant.Serialize where++import Data.ByteString (ByteString)+import Data.Serialize++-- | Consume the rest of the @Get a@ input as a ByteString+consumeByteStringToEnd :: Get ByteString+consumeByteStringToEnd = do+  numRemaining <- remaining+  getByteString numRemaining

@@ -0,0 +1,44 @@+{-|+Module      : Database.PostgreSQL.Replicant.State+Description : Internal replication stream state+Copyright   : (c) James King, 2020, 2021+License     : BSD3+Maintainer  : james@agentultra.com+Stability   : experimental+Portability : POSIX++This module has the types and functions for maintaining the client+stream state.++After initiating a replication stream the wal_sender process on the+server may require clients to periodically send progress updates.  The+wal_sender process uses those updates to maintain its internal view of+the clients' state.++This enables the server to report on things like replication lag and+enables the client to disconnect and restart the stream where it left+off.+-}+module Database.PostgreSQL.Replicant.State where++import Control.Concurrent+import Database.PostgreSQL.Replicant.Types.Lsn++data WalProgress+  = WalProgress+  { walProgressReceived :: LSN+  , walProgressFlushed  :: LSN+  , walProgressApplied  :: LSN+  }+  deriving (Eq, Show)++newtype WalProgressState = WalProgressState (MVar WalProgress)++updateWalProgress :: WalProgressState -> LSN -> IO ()+updateWalProgress (WalProgressState state) lsn = do+  walProgress <- takeMVar state+  putMVar state+    $ walProgress { walProgressReceived = lsn+                  , walProgressFlushed  = lsn+                  , walProgressApplied  = lsn+                  }

@@ -0,0 +1,104 @@+{-|+Module      : Database.PostgreSQL.Replicant.Types.Lsn+Description : Types and parsers for LSNs+Copyright   : (c) James King, 2020, 2021+License     : BSD3+Maintainer  : james@agentultra.com+Stability   : experimental+Portability : POSIX++/Log Sequence Number/ or LSN is a pointer to a place inside of a WAL+log file.  It contains the file name and an offset in bytes encoded in+two parts.++LSNs can be serialized into 64-bit big-endian numbers in the binary+protocol but are also represented textually in query results and other+places.++This module follows a similar convention to many containers libraries+and should probably be imported qualified to avoid name clashes if+needed.++See: https://www.postgresql.org/docs/10/datatype-pg-lsn.html+-}+module Database.PostgreSQL.Replicant.Types.Lsn where++import Data.Aeson+import Data.Attoparsec.ByteString.Char8+import Data.Bits+import Data.Bits.Extras+import Data.ByteString (ByteString ())+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Builder as Builder+import Data.ByteString.Lazy.Builder.ASCII (word32Hex)+import Data.Serialize+import Data.Word+import qualified Data.Text.Encoding as T+import GHC.Int++data LSN = LSN+  { filepart :: !Int32 -- ^ Filepart+  , offset :: !Int32 -- ^ Offset+  }+  deriving (Show, Eq)++instance Ord LSN where+  compare (LSN l0 r0) (LSN l1 r1) =+    compare l0 l1 <> compare r0 r1++instance Serialize LSN where+  put = putInt64be . toInt64+  get = fromInt64 <$> getInt64be++instance ToJSON LSN where+  toJSON = String . T.decodeUtf8 . toByteString++instance FromJSON LSN where+  parseJSON = withText "LSN" $ \txt ->+    case fromByteString . T.encodeUtf8 $ txt of+      Left err  -> fail err+      Right lsn -> pure lsn++-- | Convert an LSN to a 64-bit integer+toInt64 :: LSN -> Int64+toInt64 (LSN filePart offSet) =+  let r = w64 filePart `shiftL` 32+  in fromIntegral $ r .|. fromIntegral offSet++-- | Convert a 64-bit integer to an LSN+fromInt64 :: Int64 -> LSN+fromInt64 x =+  let mask = w64 $ maxBound @Word32+      offSet = fromIntegral . w32 $ mask .&. fromIntegral x+      filePart = fromIntegral $ x `shiftR` 32+  in LSN filePart offSet++lsnParser :: Parser LSN+lsnParser = LSN <$> (hexadecimal <* char '/') <*> hexadecimal++fromByteString :: ByteString -> Either String LSN+fromByteString = parseOnly lsnParser++-- | Note that as of bytestring ~0.10.12.0 we don't have upper-case+-- hex encoders but the patch to add them has been merged and when+-- available we should switch to them+toByteString :: LSN -> ByteString+toByteString (LSN filepart off) = BL.toStrict+  $ Builder.toLazyByteString+  ( word32Hex (fromIntegral filepart)+    <> Builder.char7 '/'+    <> word32Hex (fromIntegral off)+  )++-- | Add a number of bytes to an LSN+add :: LSN -> Int64 -> LSN+add lsn bytes = fromInt64 . (+ bytes) . toInt64 $ lsn++-- | Subtract a number of bytes from an LSN+sub :: LSN -> Int64 -> LSN+sub lsn bytes = fromInt64 . flip (-) bytes . toInt64 $ lsn++-- | Subtract two LSN's to calculate the difference of bytes between+-- them.+subLsn :: LSN -> LSN -> Int64+subLsn lsn1 lsn2 = toInt64 lsn1 - toInt64 lsn2

@@ -0,0 +1,12 @@+module Database.PostgreSQL.Replicant.Util where++import Control.Exception+import GHC.Int++mkInt64 :: Int -> Int64+mkInt64 k = fromIntegral k * 1000000++maybeThrow :: Exception e => e -> Maybe a -> IO a+maybeThrow exc = \case+  Nothing -> throwIO exc+  Just x  -> pure x

@@ -0,0 +1,120 @@+import Test.Hspec++import Data.Serialize+import Data.ByteString (ByteString)+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import Data.Word+import GHC.Int++import Database.PostgreSQL.Replicant.Message+import Database.PostgreSQL.Replicant.Types.Lsn+import qualified Database.PostgreSQL.Replicant.Queue as Q++examplePrimaryKeepAliveMessage :: ByteString+examplePrimaryKeepAliveMessage+  = B.concat+    [ (B.pack [0x6B])+    , (encode @LSN (fromInt64 123))+    , (encode @LSN (fromInt64 346))+    , (encode @Word8 1)+    ]++main :: IO ()+main = hspec $ do+  context "Message" $ do+    describe "PrimaryKeepAlive" $ do+      it "should decode a valid primary keep alive message" $ do+        (decode $ examplePrimaryKeepAliveMessage)+          `shouldBe`+          (Right $ PrimaryKeepAlive 123 346 ShouldRespond)++    describe "StandbyStatusUpdate" $ do+      it "should encode a valid standby status update message" $ do+        (encode $ StandbyStatusUpdate+         (fromInt64 213)+         (fromInt64 232)+         (fromInt64 234)+         454+         DoNotRespond)+          `shouldBe`+          B.concat+          [ B.pack [0x72]+          , encode @LSN (fromInt64 213)+          , encode @LSN (fromInt64 232)+          , encode @LSN (fromInt64 234)+          , encode @Int64 454+          , encode @Word8 0+          ]++    describe "XLogData" $ do+      it "should encode/decode a valid xLogData message" $ do+        let msg = XLogData (fromInt64 123) (fromInt64 234) 345 (B8.pack "hello")+        (decode . encode $ msg)+          `shouldBe`+          Right msg++    describe "HotStandbyFeedback" $ do+      it "should encode/decode a valid HotStandbyFeedback message" $ do+        let msg = HotStandbyFeedback 123 234 456+        (decode . encode $ msg)+          `shouldBe`+          Right msg++    describe "WalCopyData" $ do+      it "should encode/decode an XLogData message" $ do+        let msg = XLogDataM (XLogData (fromInt64 123) (fromInt64 234) 345 (B8.pack "hello"))+        (decode . encode $ msg)+          `shouldBe`+          Right msg++      it "should encode/decode a PrimaryKeepAlive message" $ do+        let msg = KeepAliveM (PrimaryKeepAlive 123 346 ShouldRespond)+        (decode . encode $ msg)+          `shouldBe`+          Right msg++  context "Types" $ do+    describe "LSN" $ do+      context "Serializable" $ do+        it "should be serializable" $ do+          let lsn = LSN 2 23+          (decode . encode $ lsn) `shouldBe` Right lsn++        it "should be equivalent to fromByteString/toByteString" $ do+          let (Right lsn) = fromByteString "16/3002D50"+          (toByteString <$> (decode . encode @LSN $ lsn)) `shouldBe` Right "16/3002d50"++  describe "FifoQueue" $ do+    it "should dequeue Nothing if the queue is empty" $ do+      queue <- Q.empty @Int+      res <- Q.dequeue queue+      res `shouldBe` Nothing++    it "should respect first-in, first-out" $ do+      queue <- Q.empty+      Q.enqueue queue 1+      Q.enqueue queue 2+      Q.enqueue queue 3+      res <- Q.dequeue queue+      res `shouldBe` Just 1++    context "null" $ do+      it "should return True if the queue is empty" $ do+        queue <- Q.empty+        res <- Q.null queue+        res `shouldBe` True++      it "should return False if the queue has at least one element" $ do+        queue <- Q.empty+        Q.enqueue queue 1+        res <- Q.null queue+        res `shouldBe` False++    context "enqueueRight" $ do+      it "should enqueue an item at the head of the queue" $ do+        queue <- Q.empty+        Q.enqueue queue 1+        Q.enqueueRight queue 2+        res <- Q.dequeue queue+        res `shouldBe` Just 2