diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,3 @@
+Changelog for hasql-queue
+- 1.0.0.0
+  - First release!
diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,30 @@
+Copyright Author Jonathan Fischoff (c) 2020
+
+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,33 @@
+[![Travis CI Status](https://travis-ci.org/jfischoff/hasql-queue.svg?branch=master)](http://travis-ci.org/jfischoff/hasql-queue)
+
+# hasql-queue
+
+This module utilizes PostgreSQL to implement a durable queue for efficently processing payloads.
+
+Typically a producer would enqueue a new payload as part of larger database transaction
+
+```haskell
+createAccount userRecord = transaction Serializable Write $ do
+  createUser userRecord
+  enqueueNotification "queue_channel" emailEncoder [makeVerificationEmail userRecord]
+```
+
+In another thread or process the consumer would drain the queue.
+
+```haskell
+  -- Wait for a single new record and try to send the email 5 times for giving
+  -- up and marking the payload as failed.
+  forever $ withDequeue "queue_channel" conn emailDecoder 5 1 $
+    mapM_ sendEmail
+```
+
+In the example above we used the `Session` API for enqueuing and the `IO` for
+dequeuing.
+
+The `Session` API is useful for composing larger transactions and the `IO` utilizes PostgreSQL notifications to avoid polling.
+
+## Installation
+
+```bash
+stack install hasql-queue
+```
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/benchmarks/Main.hs b/benchmarks/Main.hs
new file mode 100644
--- /dev/null
+++ b/benchmarks/Main.hs
@@ -0,0 +1,102 @@
+module Main where
+import           System.Environment
+import           Hasql.Queue.Migrate
+import           Data.IORef
+import           Control.Exception
+import           Crypto.Hash.SHA1 (hash)
+import qualified Data.ByteString.Base64.URL as Base64
+import qualified Data.ByteString.Char8 as BSC
+import           Data.Pool
+import           Database.Postgres.Temp
+import           Control.Concurrent
+import           Control.Monad (replicateM, forever, void)
+import           Hasql.Session
+import           Hasql.Connection
+import           Data.Function
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Hasql.Statement
+import qualified Hasql.Queue.Internal as I
+import qualified Hasql.Queue.Session as S
+import           Data.Int
+
+-- TODO need to make sure the number of producers and consumers does not go over the number of connections
+
+withConn :: DB -> (Connection -> IO a) -> IO a
+withConn db f = do
+  let connStr = toConnectionString db
+  bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f
+
+withSetup :: (Pool Connection -> IO ()) -> IO ()
+withSetup f = do
+  -- Helper to throw exceptions
+  let throwE x = either throwIO pure =<< x
+
+  throwE $ withDbCache $ \dbCache -> do
+    --let combinedConfig = autoExplainConfig 15 <> cacheConfig dbCache
+    let combinedConfig = defaultConfig <> cacheConfig dbCache
+    migratedConfig <- throwE $ cacheAction (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash
+        $ BSC.pack $ migrationQueryString "int4")
+        (flip withConn $ flip migrate "int4")
+        combinedConfig
+    withConfig migratedConfig $ \db -> do
+      print $ toConnectionString db
+
+      f =<< createPool
+              (either (throwIO . userError . show) pure =<< acquire (toConnectionString db)
+              ) release 2 60 49
+
+payload :: Int32
+payload = 1
+
+main :: IO ()
+main = do
+  [producerCount, consumerCount, time, initialDequeueCount, initialEnqueueCount, batchCount] <- map read <$> getArgs
+  -- create a temporary database
+  enqueueCounter <- newIORef (0 :: Int)
+  dequeueCounter <- newIORef (0 :: Int)
+
+  let printCounters = do
+        finalEnqueueCount <- readIORef enqueueCounter
+        finalDequeueCount <- readIORef dequeueCounter
+        putStrLn $ "Time " <> show time <> " secs"
+        putStrLn $ "Enqueue Count: " <> show finalEnqueueCount
+        putStrLn $ "Dequeue Count: " <> show finalDequeueCount
+
+  flip finally printCounters $ withSetup $ \pool -> do
+    -- enqueue the enqueueCount + dequeueCount
+    let enqueueAction = void $ withResource pool $ \conn -> I.runThrow (S.enqueue E.int4 [payload]) conn
+        dequeueAction = void $ withResource pool $ \conn -> fix $ \next ->
+          I.runThrow (S.dequeue D.int4 batchCount) conn >>= \case
+              [] -> next
+              _ -> pure ()
+
+    let enqueueInsertSql = "INSERT INTO payloads (attempts, value) SELECT 0, g.value FROM generate_series(1, $1) AS g (value)"
+        enqueueInsertStatement =
+          statement (fromIntegral initialEnqueueCount) $ Statement enqueueInsertSql (E.param $ E.nonNullable E.int4) D.noResult False
+
+    withResource pool $ run enqueueInsertStatement
+
+    let dequeueInsertSql = "INSERT INTO payloads (attempts, state, value) SELECT 0, 'dequeued', g.value FROM generate_series(1, $1) AS g (value)"
+        dequeueInsertStatement =
+          statement (fromIntegral initialDequeueCount) $ Statement dequeueInsertSql (E.param $ E.nonNullable E.int4) D.noResult False
+
+    withResource pool $ run dequeueInsertStatement
+
+    withResource pool $ \conn -> void $ run (sql "VACUUM FULL ANALYZE") conn
+    putStrLn "Finished VACUUM FULL ANALYZE"
+
+    let enqueueLoop = forever $ do
+          enqueueAction
+          atomicModifyIORef' enqueueCounter $ \x -> (x+1, ())
+
+        dequeueLoop = forever $ do
+           dequeueAction
+           atomicModifyIORef' dequeueCounter $ \x -> (x+1, ())
+
+    -- Need better exception behavior ... idk ... I'll deal with this later
+    _enqueueThreads <- replicateM producerCount $ forkIO enqueueLoop
+    _dequeueThreads <- replicateM consumerCount $ forkIO dequeueLoop
+
+    threadDelay $ time * 1000000
+    throwIO $ userError "Finished"
diff --git a/hasql-queue.cabal b/hasql-queue.cabal
new file mode 100644
--- /dev/null
+++ b/hasql-queue.cabal
@@ -0,0 +1,149 @@
+cabal-version: 1.12
+
+-- This file has been generated from package.yaml by hpack version 0.31.2.
+--
+-- see: https://github.com/sol/hpack
+--
+-- hash: 2c8109092a14882df219b99da03aea13b68292c07f409a58fa68028d6f263eb1
+
+name:           hasql-queue
+version:        1.0.1
+synopsis:       A PostgreSQL backed queue
+description:    This module utilize PostgreSQL to implement a durable queue for efficently processing arbitrary payloads which can be represented as JSON.
+                .
+                Typically a producer would enqueue a new payload as part of larger database
+                transaction
+                .
+                >  createAccount userRecord = do
+                >     'runDBTSerializable' $ do
+                >        createUserDB userRecord
+                >        'enqueueDB' "queue_schema" $ makeVerificationEmail userRecord
+                .
+                In another thread or process, the consumer would drain the queue.
+                .
+                >   forever $ do
+                >     -- Attempt get a payload or block until one is available
+                >     payload <- lock "queue" conn
+                >
+                >     -- Perform application specifc parsing of the payload value
+                >     case fromJSON $ pValue payload of
+                >       Success x -> sendEmail x -- Perform application specific processing
+                >       Error err -> logErr err
+                >
+                >     -- Remove the payload from future processing
+                >     dequeue "queue" conn $ pId payload
+                >
+                > To support multiple queues in the same database, the API expects a table name string
+                > to determine which queue tables to use.
+category:       Web
+homepage:       https://github.com/jfischoff/postgresql-queue#readme
+bug-reports:    https://github.com/jfischoff/postgresql-queue/issues
+author:         Jonathan Fischoff
+maintainer:     jonathangfischoff@gmail.com
+copyright:      2020 Jonathan Fischoff
+license:        BSD3
+license-file:   LICENSE
+tested-with:    GHC ==8.8.1
+build-type:     Simple
+extra-source-files:
+    README.md
+    CHANGELOG.md
+
+source-repository head
+  type: git
+  location: https://github.com/jfischoff/postgresql-queue
+
+library
+  exposed-modules:
+      Hasql.Queue.Internal
+      Hasql.Queue.IO
+      Hasql.Queue.Migrate
+      Hasql.Queue.Session
+  other-modules:
+      Paths_hasql_queue
+  hs-source-dirs:
+      src
+  default-extensions: OverloadedStrings LambdaCase RecordWildCards TupleSections GeneralizedNewtypeDeriving QuasiQuotes ScopedTypeVariables TypeApplications AllowAmbiguousTypes
+  ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls
+  build-depends:
+      aeson
+    , base >=4.7 && <5
+    , bytestring
+    , exceptions
+    , hasql
+    , hasql-notifications
+    , here
+    , monad-control
+    , random
+    , stm
+    , text
+    , time
+    , transformers
+  default-language: Haskell2010
+
+executable benchmark
+  main-is: Main.hs
+  other-modules:
+      Paths_hasql_queue
+  hs-source-dirs:
+      benchmarks
+  default-extensions: OverloadedStrings LambdaCase RecordWildCards TupleSections GeneralizedNewtypeDeriving QuasiQuotes ScopedTypeVariables TypeApplications AllowAmbiguousTypes
+  ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls -O2 -threaded -rtsopts -with-rtsopts=-N -g2
+  build-depends:
+      aeson
+    , async
+    , base >=4.7 && <5
+    , base64-bytestring
+    , bytestring
+    , cryptohash-sha1
+    , exceptions
+    , hasql
+    , hasql-notifications
+    , hasql-queue
+    , here
+    , monad-control
+    , random
+    , resource-pool
+    , stm
+    , text
+    , time
+    , tmp-postgres
+    , transformers
+  default-language: Haskell2010
+
+test-suite unit-tests
+  type: exitcode-stdio-1.0
+  main-is: Main.hs
+  other-modules:
+      Database.Hasql.Queue.IOSpec
+      Database.Hasql.Queue.SessionSpec
+      Paths_hasql_queue
+  hs-source-dirs:
+      test
+  default-extensions: OverloadedStrings LambdaCase RecordWildCards TupleSections GeneralizedNewtypeDeriving QuasiQuotes ScopedTypeVariables TypeApplications AllowAmbiguousTypes
+  ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls -O2 -threaded -rtsopts -with-rtsopts=-N
+  build-depends:
+      aeson
+    , async
+    , base >=4.7 && <5
+    , base64-bytestring
+    , bytestring
+    , cryptohash-sha1
+    , exceptions
+    , hasql
+    , hasql-notifications
+    , hasql-queue
+    , here
+    , hspec
+    , hspec-core
+    , hspec-expectations-lifted
+    , monad-control
+    , random
+    , resource-pool
+    , split
+    , stm
+    , text
+    , time
+    , tmp-postgres
+    , transformers
+  default-language: Haskell2010
diff --git a/src/Hasql/Queue/IO.hs b/src/Hasql/Queue/IO.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/IO.hs
@@ -0,0 +1,128 @@
+{-|
+IO based API for a PostgreSQL backed queue. The API utilizes PostgreSQL
+notifications.
+-}
+module Hasql.Queue.IO
+  ( enqueue
+  , withDequeue
+  -- ** Listing API
+  , I.PayloadId
+  , failed
+  , dequeued
+  -- ** Advanced API
+  , withDequeueWith
+  , I.WithNotifyHandlers (..)
+  ) where
+
+import qualified Hasql.Queue.Session as S
+import qualified Hasql.Queue.Internal as I
+import           Hasql.Connection
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Control.Exception
+import           Data.Function
+import           Data.ByteString (ByteString)
+
+{-|Enqueue a payload.
+-}
+enqueue :: ByteString
+        -- ^ Notification channel name. Any valid PostgreSQL identifier
+        -> Connection
+        -- ^ Connection
+        -> E.Value a
+        -- ^ Payload encoder
+        -> [a]
+        -- ^ List of payloads to enqueue
+        -> IO ()
+enqueue channel conn encoder xs = I.runThrow (S.enqueueNotify channel encoder xs) conn
+
+{-|
+Wait for the next payload and process it. If the continuation throws an
+exception the payloads are put back in the queue. An 'IOError' is caught
+and 'withDequeue' will retry up to the retry count. If 'withDequeue' fails
+after too many retries the final exception is rethrown. If individual payloads are
+are attempted more than the retry count they are set as "failed". See 'failed'
+to receive the list of failed payloads.
+
+If the queue is empty 'withDequeue' will block until it recieves a notification
+from the PostgreSQL server.
+-}
+withDequeue :: ByteString
+            -- ^ Notification channel name. Any valid PostgreSQL identifier
+            -> Connection
+            -- ^ Connection
+            -> D.Value a
+            -- ^ Payload decoder
+            -> Int
+            -- ^ Retry count
+            -> Int
+            -- ^ Element count
+            -> ([a] -> IO b)
+            -- ^ Continuation
+            -> IO b
+withDequeue = withDequeueWith @IOError mempty
+
+{-|
+Retrieve the payloads that have entered a failed state. See 'withDequeue' for how that
+occurs. The function returns a list of values and an id. The id is used the starting
+place for the next batch of values. If 'Nothing' is passed the list starts at the
+beginning.
+-}
+failed :: Connection
+       -> D.Value a
+       -- ^ Payload decoder
+       -> Maybe I.PayloadId
+       -- ^ Starting position of payloads. Pass 'Nothing' to
+       --   start at the beginning
+       -> Int
+       -- ^ Count
+       -> IO (I.PayloadId, [a])
+failed conn decoder mPayload count = I.runThrow (S.failed decoder mPayload count) conn
+
+{-|
+Retrieve the payloads that have been successfully dequeued.
+The function returns a list of values and an id. The id is used the starting
+place for the next batch of values. If 'Nothing' is passed the list starts at the
+beginning.
+-}
+dequeued :: Connection
+         -> D.Value a
+         -- ^ Payload decoder
+         -> Maybe I.PayloadId
+         -- ^ Starting position of payloads. Pass 'Nothing' to
+         --   start at the beginning
+         -> Int
+         -- ^ Count
+         -> IO (I.PayloadId, [a])
+dequeued conn decoder mPayload count = I.runThrow (S.dequeued decoder mPayload count) conn
+
+{-|
+A more general configurable version of 'withDequeue'. Unlike 'withDequeue' one
+can specify the
+-}
+withDequeueWith :: forall e a b
+                 . Exception e
+                => I.WithNotifyHandlers
+                -- Event handlers for events that occur as 'withDequeWith' loops
+                -> ByteString
+                -- ^ Notification channel name. Any valid PostgreSQL identifier
+                -> Connection
+                -- ^ Connection
+                -> D.Value a
+                -- ^ Payload decoder
+                -> Int
+                -- ^ Retry count
+                -> Int
+                -- ^ Element count
+                -> ([a] -> IO b)
+                -- ^ Continuation
+                -> IO b
+withDequeueWith withNotifyHandlers channel conn decoder retryCount count f = (fix $ \restart i -> do
+    try (I.withNotifyWith withNotifyHandlers channel conn (I.withDequeue decoder retryCount count f) id) >>= \case
+      Right x -> pure x
+      Left (e :: e) ->
+        if i < retryCount then
+          restart $ i + 1
+        else
+          throwIO e
+  ) 0
diff --git a/src/Hasql/Queue/Internal.hs b/src/Hasql/Queue/Internal.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/Internal.hs
@@ -0,0 +1,248 @@
+module Hasql.Queue.Internal where
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Hasql.Session
+import           Hasql.Notification
+import           Control.Monad (unless)
+import           Data.Function(fix)
+import           Hasql.Connection
+import           Data.Int
+import           Data.Functor.Contravariant
+import           Data.String.Here.Uninterpolated
+import           Hasql.Statement
+import           Data.ByteString (ByteString)
+import           Control.Exception
+import           Control.Monad.IO.Class
+import           Data.Typeable
+
+-- | A 'Payload' can exist in three states in the queue, 'Enqueued',
+--   and 'Dequeued'. A 'Payload' starts in the 'Enqueued' state and is locked
+--   so some sort of process can occur with it, usually something in 'IO'.
+--   Once the processing is complete, the `Payload' is moved the 'Dequeued'
+--   state, which is the terminal state.
+data State = Enqueued | Dequeued | Failed
+  deriving (Show, Eq, Ord, Enum, Bounded)
+
+state :: E.Params a -> D.Result b -> ByteString -> Statement a b
+state enc dec theSql = Statement theSql enc dec True
+
+stateDecoder :: D.Value State
+stateDecoder = D.enum $ \txt ->
+  if txt == "enqueued" then
+    pure Enqueued
+  else if txt == "dequeued" then
+    pure Dequeued
+  else if txt == "failed" then
+    pure Failed
+  else Nothing
+
+stateEncoder :: E.Value State
+stateEncoder = E.enum $ \case
+  Enqueued -> "enqueued"
+  Dequeued -> "dequeued"
+  Failed   -> "failed"
+
+initialPayloadId :: PayloadId
+initialPayloadId = PayloadId (-1)
+
+{-|
+Internal payload id. Used by the public api as continuation token
+for pagination.
+-}
+newtype PayloadId = PayloadId { unPayloadId :: Int64 }
+  deriving (Eq, Show)
+
+-- | The fundemental record stored in the queue. The queue is a single table
+-- and each row consists of a 'Payload'
+data Payload a = Payload
+  { pId         :: PayloadId
+  , pState      :: State
+  -- TODO do I need this?
+  , pAttempts   :: Int
+  , pModifiedAt :: Int
+  -- TODO rename. I don't need this either.
+  , pValue      :: a
+  } deriving (Show, Eq)
+
+-- | 'Payload' decoder
+payloadDecoder :: D.Value a -> D.Row (Payload a)
+payloadDecoder thePayloadDecoder
+   =  Payload
+  <$> payloadIdRow
+  <*> D.column (D.nonNullable stateDecoder)
+  <*> D.column (D.nonNullable $ fromIntegral <$> D.int4)
+  <*> D.column (D.nonNullable $ fromIntegral <$> D.int4)
+  <*> D.column (D.nonNullable thePayloadDecoder)
+
+payloadIdEncoder :: E.Value PayloadId
+payloadIdEncoder = unPayloadId >$< E.int8
+
+payloadIdDecoder :: D.Value PayloadId
+payloadIdDecoder = PayloadId <$> D.int8
+
+payloadIdRow :: D.Row PayloadId
+payloadIdRow = D.column (D.nonNullable payloadIdDecoder)
+
+-- TODO include special cases for single element insertion
+enqueuePayload :: E.Value a -> [a] -> Session [PayloadId]
+enqueuePayload theEncoder values = do
+  let theQuery = [here|
+        INSERT INTO payloads (attempts, value)
+        SELECT 0, * FROM unnest($1)
+        RETURNING id
+        |]
+      encoder = E.param $ E.nonNullable $ E.foldableArray $ E.nonNullable theEncoder
+      decoder = D.rowList (D.column (D.nonNullable payloadIdDecoder))
+      theStatement = Statement theQuery encoder decoder True
+
+  statement values theStatement
+
+dequeuePayload :: D.Value a -> Int -> Session [Payload a]
+dequeuePayload valueDecoder count = do
+  let multipleQuery = [here|
+        UPDATE payloads
+        SET state='dequeued'
+        WHERE id in
+          ( SELECT p1.id
+            FROM payloads AS p1
+            WHERE p1.state='enqueued'
+            ORDER BY p1.modified_at ASC
+            FOR UPDATE SKIP LOCKED
+            LIMIT $1
+          )
+        RETURNING id, state, attempts, modified_at, value
+      |]
+      multipleEncoder = E.param $ E.nonNullable $ fromIntegral >$< E.int4
+
+      singleQuery = [here|
+        UPDATE payloads
+        SET state='dequeued'
+        WHERE id in
+          ( SELECT p1.id
+            FROM payloads AS p1
+            WHERE p1.state='enqueued'
+            ORDER BY p1.modified_at ASC
+            FOR UPDATE SKIP LOCKED
+            LIMIT 1
+          )
+        RETURNING id, state, attempts, modified_at, value
+      |]
+
+      singleEncoder = mempty
+
+      decoder = D.rowList $ payloadDecoder valueDecoder
+
+      theStatement = case count of
+        1 -> Statement singleQuery singleEncoder decoder True
+        _ -> Statement multipleQuery multipleEncoder decoder True
+  statement count theStatement
+
+-- | Get the 'Payload' given a 'PayloadId'
+getPayload :: D.Value a -> PayloadId -> Session (Maybe (Payload a))
+getPayload decoder payloadId = do
+  let theQuery = [here|
+    SELECT id, state, attempts, modified_at, value
+    FROM payloads
+    WHERE id = $1
+  |]
+
+      encoder = E.param (E.nonNullable payloadIdEncoder)
+  statement payloadId $ Statement theQuery encoder (D.rowMaybe $ payloadDecoder decoder) True
+
+
+-- | Get the number of rows in the 'Enqueued' state.
+getCount :: Session Int64
+getCount = do
+  let decoder = D.singleRow (D.column (D.nonNullable D.int8))
+      theSql = [here|
+            SELECT count(*)
+            FROM payloads
+            WHERE state='enqueued';
+        |]
+      theStatement = Statement theSql mempty decoder True
+  statement () theStatement
+
+incrementAttempts :: Int -> [PayloadId] -> Session ()
+incrementAttempts retryCount pids = do
+  let theQuery = [here|
+        UPDATE payloads
+        SET state=CASE WHEN attempts >= $1 THEN 'failed' :: state_t ELSE 'enqueued' END
+          , attempts=attempts+1
+        WHERE id = ANY($2)
+        |]
+      encoder = (fst >$< E.param (E.nonNullable E.int4)) <>
+                (snd >$< E.param (E.nonNullable $ E.foldableArray $ E.nonNullable payloadIdEncoder))
+
+      theStatement = Statement theQuery encoder D.noResult True
+
+  statement (fromIntegral retryCount, pids) theStatement
+
+-- | Dequeue and
+--   Move to Internal
+-- This should use bracketOnError
+withDequeue :: D.Value a -> Int -> Int -> ([a] -> IO b) -> Session (Maybe b)
+withDequeue decoder retryCount count f = do
+  sql "BEGIN;SAVEPOINT temp"
+  dequeuePayload decoder count >>= \case
+    [] ->  Nothing <$ sql "COMMIT"
+    xs -> fmap Just $ do
+      liftIO (try $ f $ fmap pValue xs) >>= \case
+        Left  (e :: SomeException) -> do
+           sql "ROLLBACK TO SAVEPOINT temp; RELEASE SAVEPOINT temp"
+           let pids = fmap pId xs
+           incrementAttempts retryCount pids
+           sql "COMMIT"
+           liftIO (throwIO e)
+        Right x ->  x <$ sql "COMMIT"
+
+-- TODO remove
+newtype QueryException = QueryException QueryError
+  deriving (Eq, Show, Typeable)
+
+instance Exception QueryException
+
+runThrow :: Session a -> Connection -> IO a
+runThrow sess conn = either (throwIO . QueryException) pure =<< run sess conn
+
+execute :: Connection -> ByteString -> IO ()
+execute conn theSql = runThrow (sql theSql) conn
+
+-- Block until a payload notification is fired. Fired during insertion.
+notifyPayload :: ByteString -> Connection -> IO ()
+notifyPayload channel conn = fix $ \restart -> do
+  Notification {..} <- either throwIO pure =<< getNotification conn
+  unless (notificationChannel == channel) restart
+
+-- | To aid in observability and white box testing
+data WithNotifyHandlers = WithNotifyHandlers
+  { withNotifyHandlersAfterAction        :: IO ()
+  -- ^ An event that is trigger after the initial action, e.g.
+  --   before dequeue is called.
+  , withNotifyHandlersBeforeNotification :: IO ()
+  -- ^ An event that is triggered before the blocking on a
+  --   notification.
+  }
+
+instance Semigroup WithNotifyHandlers where
+  x <> y = WithNotifyHandlers
+    { withNotifyHandlersAfterAction = withNotifyHandlersAfterAction x <> withNotifyHandlersAfterAction y
+    , withNotifyHandlersBeforeNotification = withNotifyHandlersBeforeNotification      x <> withNotifyHandlersBeforeNotification      y
+    }
+
+instance Monoid WithNotifyHandlers where
+  mempty = WithNotifyHandlers mempty mempty
+
+withNotifyWith :: WithNotifyHandlers -> ByteString -> Connection -> Session a -> (a -> Maybe b) -> IO b
+withNotifyWith WithNotifyHandlers {..} channel conn action theCast = bracket_
+  (execute conn $ "LISTEN " <> channel)
+  (execute conn $ "UNLISTEN " <> channel)
+  $ fix $ \restart -> do
+    x <- runThrow action conn
+    withNotifyHandlersAfterAction
+    case theCast x of
+      Nothing -> do
+        -- TODO record the time here
+        withNotifyHandlersBeforeNotification
+        notifyPayload channel conn
+        restart
+      Just xs -> pure xs
diff --git a/src/Hasql/Queue/Migrate.hs b/src/Hasql/Queue/Migrate.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/Migrate.hs
@@ -0,0 +1,81 @@
+{-|
+Functions for migrating the database to create the necessary
+functions for the package.
+
+Users can use these functions or copy and paste the tables
+to create these tables through a standalone migration
+system.
+-}
+{-# OPTIONS_HADDOCK prune #-}
+{-# LANGUAGE QuasiQuotes, OverloadedStrings #-}
+module Hasql.Queue.Migrate where
+import           Control.Monad
+import           Data.String
+import           Data.String.Here.Interpolated
+import           Hasql.Connection
+import           Hasql.Session
+
+
+{-|
+The DDL statements to create the schema given a value type.
+-}
+migrationQueryString :: String
+                     -- ^ @value@ column type, e.g. @int4@ or
+                     -- @jsonb@.
+                     -> String
+migrationQueryString valueType = [i|
+    DO $$
+  BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN
+      CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');
+    END IF;
+  END$$;
+
+  CREATE SEQUENCE IF NOT EXISTS modified_index START 1;
+
+  CREATE TABLE IF NOT EXISTS payloads
+  ( id BIGSERIAL PRIMARY KEY
+  , attempts int NOT NULL DEFAULT 0
+  , state state_t NOT NULL DEFAULT 'enqueued'
+  , modified_at int8 NOT NULL DEFAULT nextval('modified_index')
+  , value ${valueType} NOT NULL
+  ) WITH (fillfactor = 50);
+
+  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at)
+    WHERE (state = 'enqueued');
+
+|]
+
+{-| This function creates a table and enumeration type that is
+    appriopiate for the queue. The following sql is used.
+
+ @
+ DO $$
+  BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN
+      CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');
+    END IF;
+  END$$;
+
+  CREATE SEQUENCE IF NOT EXISTS modified_index START 1;
+
+  CREATE TABLE IF NOT EXISTS payloads
+  ( id BIGSERIAL PRIMARY KEY
+  , attempts int NOT NULL DEFAULT 0
+  , state state_t NOT NULL DEFAULT 'enqueued'
+  , modified_at int8 NOT NULL DEFAULT nextval('modified_index')
+  , value VALUE_TYPE NOT NULL
+  ) WITH (fillfactor = 50);
+
+  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at)
+    WHERE (state = 'enqueued');
+ @
+
+The @VALUE_TYPE@ needs to passed in through the second argument.
+-}
+migrate :: Connection
+        -> String
+        -- ^ The type of the @value@ column
+        -> IO ()
+migrate conn valueType = void $
+  run (sql $ fromString $ migrationQueryString valueType) conn
diff --git a/src/Hasql/Queue/Session.hs b/src/Hasql/Queue/Session.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/Session.hs
@@ -0,0 +1,180 @@
+{-|
+'Session' based API for a PostgreSQL backed queue.
+-}
+module Hasql.Queue.Session
+  ( enqueue
+  , enqueueNotify
+  , dequeue
+  -- ** Listing API
+  , PayloadId
+  , failed
+  , dequeued
+  ) where
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Hasql.Session
+import           Data.Functor.Contravariant
+import           Data.String.Here.Uninterpolated
+import           Hasql.Statement
+import           Hasql.Queue.Internal
+import           Data.Maybe
+import           Data.ByteString (ByteString)
+
+{-|Enqueue a payload.
+-}
+enqueue :: E.Value a
+        -- ^ Payload encoder
+        -> [a]
+        -- ^ List of payloads to enqueue
+        -> Session ()
+enqueue theEncoder = \case
+  [x] -> do
+    let theQuery =
+          [here|
+            INSERT INTO payloads (attempts, value)
+            VALUES (0, $1)
+          |]
+
+        encoder = E.param $ E.nonNullable theEncoder
+
+    statement x $ Statement theQuery encoder D.noResult True
+
+  xs -> do
+    let theQuery =
+          [here|
+            INSERT INTO payloads (attempts, value)
+            SELECT 0, * FROM unnest($1)
+          |]
+
+        encoder = E.param $ E.nonNullable $ E.foldableArray $ E.nonNullable theEncoder
+
+    statement xs $ Statement theQuery encoder D.noResult True
+
+{-|Enqueue a payload send a notification on the
+specified channel.
+-}
+enqueueNotify :: ByteString
+              -- ^ Notification channel name. Any valid PostgreSQL identifier
+              -> E.Value a
+              -- ^ Payload encoder
+              -> [a]
+              -- ^ List of payloads to enqueue
+              -> Session ()
+enqueueNotify channel theEncoder values = do
+  enqueue theEncoder values
+  sql $ "NOTIFY " <> channel
+
+{-|
+Dequeue a list of payloads
+-}
+dequeue :: D.Value a
+        -- ^ Payload decoder
+        -> Int
+        -- ^ Element count
+        -> Session [a]
+dequeue valueDecoder count = do
+  let multipleQuery = [here|
+        UPDATE payloads
+        SET state='dequeued'
+        WHERE id in
+          ( SELECT p1.id
+            FROM payloads AS p1
+            WHERE p1.state='enqueued'
+            ORDER BY p1.modified_at ASC
+            FOR UPDATE SKIP LOCKED
+            LIMIT $1
+          )
+        RETURNING value
+      |]
+      multipleEncoder = E.param $ E.nonNullable $ fromIntegral >$< E.int4
+
+      singleQuery = [here|
+        UPDATE payloads
+        SET state='dequeued'
+        WHERE id in
+          ( SELECT p1.id
+            FROM payloads AS p1
+            WHERE p1.state='enqueued'
+            ORDER BY p1.modified_at ASC
+            FOR UPDATE SKIP LOCKED
+            LIMIT 1
+          )
+        RETURNING value
+      |]
+
+      singleEncoder = mempty
+
+      decoder = D.rowList $ D.column $ D.nonNullable $ valueDecoder
+
+      theStatement = case count of
+        1 -> Statement singleQuery singleEncoder decoder True
+        _ -> Statement multipleQuery multipleEncoder decoder True
+  statement count theStatement
+
+fst3 :: (a, b, c) -> a
+fst3 (x, _, _) = x
+
+snd3 :: (a, b, c) -> b
+snd3 (_, x, _) = x
+
+trd3 :: (a, b, c) -> c
+trd3 (_, _, x) = x
+
+listState :: State -> D.Value a -> Maybe PayloadId -> Int -> Session (PayloadId, [a])
+listState theState valueDecoder mPayloadId count = do
+  let theQuery = [here|
+        SELECT id, value
+        FROM payloads
+        WHERE state = ($1 :: state_t)
+          AND id > $2
+        ORDER BY id ASC
+        LIMIT $3
+        |]
+      encoder = (fst3 >$< E.param (E.nonNullable stateEncoder))
+             <> (snd3 >$< E.param (E.nonNullable payloadIdEncoder))
+             <> (trd3 >$< E.param (E.nonNullable E.int4))
+
+      decoder =  D.rowList
+              $  (,)
+             <$> D.column (D.nonNullable payloadIdDecoder)
+             <*> D.column (D.nonNullable valueDecoder)
+      theStatement = Statement theQuery encoder decoder True
+
+      defaultPayloadId = fromMaybe initialPayloadId mPayloadId
+
+  idsAndValues <- statement (theState, defaultPayloadId, fromIntegral count) theStatement
+  pure $ case idsAndValues of
+    [] -> (defaultPayloadId, [])
+    xs -> (fst $ last xs, map snd xs)
+
+{-|
+Retrieve the payloads that have entered a failed state. See 'withDequeue' for how that
+occurs. The function returns a list of values and an id. The id is used the starting
+place for the next batch of values. If 'Nothing' is passed the list starts at the
+beginning.
+-}
+failed :: D.Value a
+       -- ^ Payload decoder
+       -> Maybe PayloadId
+       -- ^ Starting position of payloads. Pass 'Nothing' to
+       --   start at the beginning
+       -> Int
+       -- ^ Count
+       -> Session (PayloadId, [a])
+failed = listState Failed
+
+{-|
+Retrieve the payloads that have been successfully dequeued.
+The function returns a list of values and an id. The id is used the starting
+place for the next batch of values. If 'Nothing' is passed the list starts at the
+beginning.
+-}
+dequeued :: D.Value a
+         -- ^ Payload decoder
+         -> Maybe PayloadId
+         -- ^ Starting position of payloads. Pass 'Nothing' to
+         --   start at the beginning
+         -> Int
+         -- ^ Count
+         -> Session (PayloadId, [a])
+dequeued = listState Dequeued
diff --git a/test/Database/Hasql/Queue/IOSpec.hs b/test/Database/Hasql/Queue/IOSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Database/Hasql/Queue/IOSpec.hs
@@ -0,0 +1,206 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards   #-}
+{-# LANGUAGE ScopedTypeVariables   #-}
+module Database.Hasql.Queue.IOSpec where
+import           Control.Concurrent
+import           Control.Concurrent.STM
+import           Control.Concurrent.Async
+import           Control.Exception as E
+import           Control.Monad
+import           Data.Aeson
+import           Data.Function
+import           Data.IORef
+import           Data.List
+import           Hasql.Queue.IO
+import           Hasql.Queue.Migrate
+import           Test.Hspec                     (SpecWith, Spec, describe, it, afterAll, beforeAll, runIO)
+import           Test.Hspec.Expectations.Lifted
+import           Data.List.Split
+import           Database.Postgres.Temp as Temp
+import           Data.Pool
+import           Data.Foldable
+import           Crypto.Hash.SHA1 (hash)
+import qualified Data.ByteString.Base64.URL as Base64
+import qualified Data.ByteString.Char8 as BSC
+import           Data.ByteString (ByteString)
+import           Hasql.Connection
+import           Hasql.Session
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Data.Int
+import           Data.Typeable
+import qualified Hasql.Queue.Internal as I
+import           Hasql.Queue.Internal (Payload (..))
+
+aroundAll :: forall a. ((a -> IO ()) -> IO ()) -> SpecWith a -> Spec
+aroundAll withFunc specWith = do
+  (var, stopper, asyncer) <- runIO $
+    (,,) <$> newEmptyMVar <*> newEmptyMVar <*> newIORef Nothing
+  let theStart :: IO a
+      theStart = do
+
+        thread <- async $ do
+          withFunc $ \x -> do
+            putMVar var x
+            takeMVar stopper
+          pure $ error "Don't evaluate this"
+
+        writeIORef asyncer $ Just thread
+
+        either pure pure =<< (wait thread `race` takeMVar var)
+
+      theStop :: a -> IO ()
+      theStop _ = do
+        putMVar stopper ()
+        traverse_ cancel =<< readIORef asyncer
+
+  beforeAll theStart $ afterAll theStop $ specWith
+
+withConn :: Temp.DB -> (Connection -> IO a) -> IO a
+withConn db f = do
+  let connStr = toConnectionString db
+  E.bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f
+
+runThrow :: Session a -> Connection -> IO a
+runThrow sess conn = either (throwIO . I.QueryException) pure =<< run sess conn
+
+getCount :: Connection -> IO Int64
+getCount = runThrow I.getCount
+
+getPayload :: Connection -> D.Value a -> I.PayloadId -> IO (Maybe (I.Payload a))
+getPayload conn decoder payloadId = runThrow (I.getPayload decoder payloadId) conn
+
+withSetup :: (Pool Connection -> IO ()) -> IO ()
+withSetup f = either throwIO pure <=< withDbCache $ \dbCache -> do
+  migratedConfig <- either throwIO pure =<<
+      cacheAction
+        (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash
+          $ BSC.pack $ migrationQueryString "int4")
+        (flip withConn $ flip migrate "int4")
+        (verboseConfig <> cacheConfig dbCache)
+  withConfig migratedConfig $ \db -> do
+    f =<< createPool
+      (either (throwIO . userError . show) pure =<< acquire (toConnectionString db))
+      release
+      2
+      60
+      50
+
+channel :: ByteString
+channel = "hey"
+
+withConnection :: (Connection -> IO ()) -> Pool Connection -> IO ()
+withConnection = flip withResource
+
+runReadCommitted :: Pool Connection -> Session a -> IO a
+runReadCommitted = flip withReadCommitted
+
+withReadCommitted :: Session a -> Pool Connection -> IO a
+withReadCommitted action pool = do
+  let wrappedAction = do
+        sql "BEGIN"
+        r <- action
+        sql "ROLLBACK"
+        pure r
+  withResource pool $ \conn ->
+    either (throwIO . userError . show) pure =<< run wrappedAction conn
+
+{-
+runNoTransaction :: Pool Connection -> Session a -> IO a
+runNoTransaction pool session = withResource pool $ \conn ->
+  either (throwIO . userError . show) pure =<< run action conn
+-}
+
+data FailedwithDequeue = FailedwithDequeue
+  deriving (Show, Eq, Typeable)
+
+instance Exception FailedwithDequeue
+
+spec :: Spec
+spec = describe "Hasql.Queue.IO" $ do
+  aroundAll withSetup $ describe "basic" $ do
+    it "withDequeue blocks until something is enqueued: before" $ withConnection $ \conn -> do
+      void $ enqueue channel conn E.int4 [1]
+      res <- withDequeueWith @IOException mempty channel conn D.int4 1 1 pure
+      res `shouldBe` [1]
+      getCount conn `shouldReturn` 0
+
+    it "withDequeue blocks until something is enqueued: during" $ withConnection $ \conn -> do
+      afterActionMVar  <- newEmptyMVar
+      beforeNotifyMVar <- newEmptyMVar
+
+      let handlers = WithNotifyHandlers
+            { withNotifyHandlersAfterAction = putMVar afterActionMVar ()
+            , withNotifyHandlersBeforeNotification      = takeMVar beforeNotifyMVar
+            }
+
+      -- This is the definition of IO.dequeue
+      resultThread <- async $ withDequeueWith @IOException handlers channel conn D.int4 1 1 pure
+      takeMVar afterActionMVar
+
+      void $ enqueue "hey" conn E.int4 [1]
+
+      putMVar beforeNotifyMVar ()
+
+      wait resultThread `shouldReturn` [1]
+
+    it "withDequeue blocks until something is enqueued: after" $ withConnection $ \conn -> do
+      resultThread <- async $ withDequeueWith @IOException mempty channel conn D.int4 1 1 pure
+      void $ enqueue channel conn E.int4 [1]
+
+      wait resultThread `shouldReturn` [1]
+
+    it "withDequeue fails and sets the retries to +1" $ withConnection $ \conn -> do
+      [payloadId] <- runThrow (I.enqueuePayload E.int4 [1]) conn
+      handle (\FailedwithDequeue -> pure ()) $ withDequeue channel conn D.int4 0 1 $ \_ -> throwIO FailedwithDequeue
+      Just Payload {..} <- getPayload conn D.int4 payloadId
+
+      pState `shouldBe` I.Failed
+      pAttempts  `shouldBe` 1
+
+    it "withDequeue succeeds even if the first attempt fails" $ withConnection $ \conn -> do
+      [payloadId] <- runThrow (I.enqueuePayload E.int4 [1]) conn
+
+      ref <- newIORef (0 :: Int)
+
+      withDequeueWith @FailedwithDequeue mempty channel conn D.int4 1 1 (\_ -> do
+        count <- readIORef ref
+        writeIORef ref $ count + 1
+        when (count < 1) $ throwIO FailedwithDequeue
+        pure '!') `shouldReturn` '!'
+
+      Just Payload {..} <- getPayload conn D.int4 payloadId
+
+      pState `shouldBe` I.Dequeued
+      -- Failed attempts I guess
+      pAttempts  `shouldBe` 1
+
+    it "enqueues and dequeues concurrently withDequeue" $ \testDB -> do
+      let withPool' = flip withConnection testDB
+          elementCount = 1000 :: Int
+          expected = [0 .. elementCount - 1]
+
+      ref <- newTVarIO []
+
+      loopThreads <- replicateM 35 $ async $ withPool' $ \c -> fix $ \next -> do
+        lastCount <- withDequeue channel c D.int4 1 1 $ \[x] -> do
+          atomically $ do
+            xs <- readTVar ref
+            writeTVar ref $ x : xs
+            return $ length xs + 1
+
+        when (lastCount < elementCount) next
+
+      forM_ (chunksOf (elementCount `div` 11) expected) $ \xs -> forkIO $ void $ withPool' $ \c ->
+         forM_ xs $ \i -> enqueue channel c E.int4 [fromIntegral i]
+
+      _ <- waitAnyCancel loopThreads
+      xs <- atomically $ readTVar ref
+      let Just decoded = mapM (decode . encode) xs
+      sort decoded `shouldBe` sort expected
+
+    it "enqueue returns a PayloadId that cooresponds to the entry it added" $ withConnection $ \conn -> do
+      [payloadId] <- runThrow (I.enqueuePayload E.int4 [1]) conn
+      Just actual <- getPayload conn D.int4 payloadId
+
+      pValue actual `shouldBe` 1
diff --git a/test/Database/Hasql/Queue/SessionSpec.hs b/test/Database/Hasql/Queue/SessionSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Database/Hasql/Queue/SessionSpec.hs
@@ -0,0 +1,226 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards   #-}
+{-# LANGUAGE ScopedTypeVariables   #-}
+module Database.Hasql.Queue.SessionSpec where
+import           Hasql.Queue.Internal
+import           Control.Concurrent
+import           Control.Concurrent.Async
+import           Control.Exception as E
+import           Control.Monad
+import           Data.IORef
+import           Hasql.Queue.Session
+import           Hasql.Queue.Migrate
+import           Test.Hspec                     (SpecWith, Spec, describe, parallel, it, afterAll, beforeAll, runIO)
+import           Test.Hspec.Expectations.Lifted
+import           Control.Monad.Catch
+import           Control.Monad.IO.Class
+import           Database.Postgres.Temp as Temp
+import           Data.Pool
+import           Data.Foldable
+import           Test.Hspec.Core.Spec (sequential)
+import           Crypto.Hash.SHA1 (hash)
+import qualified Data.ByteString.Base64.URL as Base64
+import qualified Data.ByteString.Char8 as BSC
+import           Hasql.Connection
+import           Hasql.Session
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Data.Typeable
+import           Data.Int
+
+aroundAll :: forall a. ((a -> IO ()) -> IO ()) -> SpecWith a -> Spec
+aroundAll withFunc specWith = do
+  (var, stopper, asyncer) <- runIO $
+    (,,) <$> newEmptyMVar <*> newEmptyMVar <*> newIORef Nothing
+  let theStart :: IO a
+      theStart = do
+
+        thread <- async $ do
+          withFunc $ \x -> do
+            putMVar var x
+            takeMVar stopper
+          pure $ error "Don't evaluate this"
+
+        writeIORef asyncer $ Just thread
+
+        either pure pure =<< (wait thread `race` takeMVar var)
+
+      theStop :: a -> IO ()
+      theStop _ = do
+        putMVar stopper ()
+        traverse_ cancel =<< readIORef asyncer
+
+  beforeAll theStart $ afterAll theStop $ specWith
+
+withConn :: Temp.DB -> (Connection -> IO a) -> IO a
+withConn db f = do
+  let connStr = toConnectionString db
+  E.bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f
+
+withSetup :: (Pool Connection -> IO ()) -> IO ()
+withSetup f = either throwIO pure <=< withDbCache $ \dbCache -> do
+  migratedConfig <- either throwIO pure =<<
+      cacheAction
+        (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash
+          $ BSC.pack $ migrationQueryString "int4")
+        (flip withConn $ flip migrate "int4")
+        (verboseConfig <> cacheConfig dbCache)
+  withConfig migratedConfig $ \db -> do
+    f =<< createPool
+      (either (throwIO . userError . show) pure =<< acquire (toConnectionString db))
+      release
+      2
+      60
+      50
+
+withConnection :: (Connection -> IO ()) -> Pool Connection -> IO ()
+withConnection = flip withResource
+
+runImplicitTransaction :: Pool Connection -> Session a -> IO a
+runImplicitTransaction pool action = do
+  let wrappedAction = do
+        r <- action
+        pure r
+  withResource pool $ \conn ->
+    either (throwIO . userError . show) pure =<< run wrappedAction conn
+
+runReadCommitted :: Pool Connection -> Session a -> IO a
+runReadCommitted = flip withReadCommitted
+
+withReadCommitted :: Session a -> Pool Connection -> IO a
+withReadCommitted action pool = do
+  let wrappedAction = do
+        sql "BEGIN"
+        r <- action
+        sql "ROLLBACK"
+        pure r
+  withResource pool $ \conn ->
+    either (throwIO . userError . show) pure =<< run wrappedAction conn
+
+newtype TooManyRetries = TooManyRetries Int64
+  deriving (Show, Eq, Typeable)
+
+instance Exception TooManyRetries
+
+spec :: Spec
+spec = describe "Hasql.Queue.Session" $ parallel $ do
+  sequential $ aroundAll withSetup $ describe "basic" $ do
+    it "is okay to migrate multiple times" $ withConnection $ \conn ->
+      liftIO $ migrate conn "int4"
+
+    it "empty locks nothing" $ \pool -> do
+      runReadCommitted pool (withDequeue D.int4 8 1 return) >>= \x ->
+        x `shouldBe` Nothing
+    it "empty gives count 0" $ \pool ->
+      runReadCommitted pool getCount `shouldReturn` 0
+
+    it "dequeued paging works" $ \pool -> do
+      (a, b) <- runReadCommitted pool $ do
+        enqueue E.int4 [1,2,3,4]
+
+        void $ dequeue D.int4 4
+
+        (next, xs) <- dequeued D.int4 Nothing 2
+
+        (_, ys) <- dequeued D.int4 (Just next) 2
+
+        pure (xs, ys)
+
+      a `shouldBe` [1,2]
+      b `shouldBe` [3,4]
+
+    it "failed paging works" $ \pool -> do
+      runImplicitTransaction pool $ enqueue E.int4 [1,2,3,4]
+
+      replicateM_ 8 $ E.handle (\(_ :: TooManyRetries) -> pure ()) $
+        runImplicitTransaction pool $ do
+          void $ withDequeue D.int4 1 1 $ const $
+            throwM $ TooManyRetries 1
+
+      (a, b) <- runImplicitTransaction pool $ do
+        (next, xs) <- failed D.int4 Nothing 2
+        (_, ys) <- failed D.int4 (Just next) 2
+
+        pure (xs, ys)
+
+      a `shouldBe` [1,2]
+      b `shouldBe` [3,4]
+
+    it "enqueue/withDequeue" $ \pool -> do
+      (withDequeueResult, firstCount, secondCount) <- runReadCommitted pool $ do
+        enqueueNotify "hey" E.int4 [1]
+        firstCount <- getCount
+        withDequeueResult <- withDequeue D.int4 8 1 (`shouldBe` [1])
+
+        secondCount <- getCount
+        pure (withDequeueResult, firstCount, secondCount)
+
+      firstCount `shouldBe` 1
+      secondCount `shouldBe` 0
+      withDequeueResult `shouldBe` Just ()
+
+    it "enqueue/withDequeue/retries" $ \pool -> do
+      runImplicitTransaction pool $ enqueue E.int4 [1]
+
+      e <- E.try $ runImplicitTransaction pool $ do
+        theCount <- getCount
+
+        void $ withDequeue D.int4 8 1 $ const $
+            throwM $ TooManyRetries theCount
+
+      (e :: Either TooManyRetries ()) `shouldBe` Left (TooManyRetries 1)
+
+      runImplicitTransaction pool (dequeuePayload D.int4 1) >>= \[(Payload {..})] -> do
+        pAttempts `shouldBe` 1
+        pValue `shouldBe` 1
+
+      runImplicitTransaction pool $ enqueue E.int4 [1]
+
+      e1 <- E.try $ runImplicitTransaction pool $ do
+        theCount <- getCount
+
+        void $ withDequeue D.int4 8 1 $ const $
+            throwM $ TooManyRetries theCount
+
+      (e1 :: Either TooManyRetries ()) `shouldBe` Left (TooManyRetries 1)
+
+      replicateM_ 6 $ E.handle (\(_ :: TooManyRetries) -> pure ()) $ runImplicitTransaction pool $ do
+          void $ withDequeue D.int4 8 1 $ const $
+            throwM $ TooManyRetries 1
+
+      runImplicitTransaction pool (dequeuePayload D.int4 1) >>= \[(Payload {..})] -> do
+        pAttempts `shouldBe` 7
+        pValue `shouldBe` 1
+
+    it "enqueue/withDequeue/timesout" $ \pool -> do
+      e <- E.try $ runReadCommitted pool $ do
+        enqueue E.int4 [1]
+        firstCount <- getCount
+
+        void $ withDequeue D.int4 0 1 $ const $
+            throwM $ TooManyRetries firstCount
+
+      (e :: Either TooManyRetries ())`shouldBe` Left (TooManyRetries 1)
+
+      runReadCommitted pool getCount `shouldReturn` 0
+
+    it "selects the oldest first" $ \pool -> do
+      (firstCount, firstwithDequeueResult, secondwithDequeueResult, secondCount) <- runReadCommitted pool $ do
+        enqueue E.int4 [1]
+        liftIO $ threadDelay 100
+
+        enqueue E.int4 [2]
+
+        firstCount <- getCount
+
+        firstwithDequeueResult   <- withDequeue D.int4 8 1 (`shouldBe` [1])
+        secondwithDequeueResult <- withDequeue D.int4 8 1 (`shouldBe` [2])
+
+        secondCount <- getCount
+        pure (firstCount, firstwithDequeueResult, secondwithDequeueResult, secondCount)
+
+      firstCount `shouldBe` 2
+      firstwithDequeueResult `shouldBe` Just ()
+
+      secondCount `shouldBe` 0
+      secondwithDequeueResult `shouldBe` Just ()
diff --git a/test/Main.hs b/test/Main.hs
new file mode 100644
--- /dev/null
+++ b/test/Main.hs
@@ -0,0 +1,9 @@
+import           Test.Hspec
+import           Database.Hasql.Queue.SessionSpec as S
+import           Database.Hasql.Queue.IOSpec as I
+
+
+main :: IO ()
+main = hspec $ do
+  S.spec
+  I.spec