diff --git a/ChangeLog.md b/ChangeLog.md
new file mode 100644
--- /dev/null
+++ b/ChangeLog.md
@@ -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
diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -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.
diff --git a/README.md b/README.md
new file mode 100644
--- /dev/null
+++ b/README.md
@@ -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.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/app/Main.hs b/app/Main.hs
new file mode 100644
--- /dev/null
+++ b/app/Main.hs
@@ -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
diff --git a/postgresql-replicant.cabal b/postgresql-replicant.cabal
new file mode 100644
--- /dev/null
+++ b/postgresql-replicant.cabal
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant.hs b/src/Database/PostgreSQL/Replicant.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Exception.hs b/src/Database/PostgreSQL/Replicant/Exception.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Exception.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Message.hs b/src/Database/PostgreSQL/Replicant/Message.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Message.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/PostgresUtils.hs b/src/Database/PostgreSQL/Replicant/PostgresUtils.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/PostgresUtils.hs
@@ -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))
diff --git a/src/Database/PostgreSQL/Replicant/Protocol.hs b/src/Database/PostgreSQL/Replicant/Protocol.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Protocol.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Queue.hs b/src/Database/PostgreSQL/Replicant/Queue.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Queue.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/ReplicationSlot.hs b/src/Database/PostgreSQL/Replicant/ReplicationSlot.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/ReplicationSlot.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Serialize.hs b/src/Database/PostgreSQL/Replicant/Serialize.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Serialize.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/State.hs b/src/Database/PostgreSQL/Replicant/State.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/State.hs
@@ -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
+                  }
diff --git a/src/Database/PostgreSQL/Replicant/Types/Lsn.hs b/src/Database/PostgreSQL/Replicant/Types/Lsn.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Types/Lsn.hs
@@ -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
diff --git a/src/Database/PostgreSQL/Replicant/Util.hs b/src/Database/PostgreSQL/Replicant/Util.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Replicant/Util.hs
@@ -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
diff --git a/test/Spec.hs b/test/Spec.hs
new file mode 100644
--- /dev/null
+++ b/test/Spec.hs
@@ -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