diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,19 @@
 Changelog for hasql-queue
+-1.2.0.0
+  - Escape notification channel #100
+  - Remove dequeued count from benchmarks #99
+  - Document new api #98
+  - Create teardown for migration #96
+  - Extend API to use schemas #95
+  - Add `delete` to the `AtLeastOnce` APIs #92
+  - Change `failed` function to `failures` and return all `PayloadIds` #91
+  - Reorg API #89
+  - Use a compound index to make the partial index work #86
+  - Specialize single element dequeue #84
+  - Delete instead of changing the state to dequeued #82
+  -
+
+
 - 1.0.1.1
   - Fixed cabal meta data and copyright
 
diff --git a/benchmarks/Main.hs b/benchmarks/Main.hs
--- a/benchmarks/Main.hs
+++ b/benchmarks/Main.hs
@@ -17,7 +17,8 @@
 import qualified Hasql.Decoders as D
 import           Hasql.Statement
 import qualified Hasql.Queue.Internal as I
-import qualified Hasql.Queue.Session as S
+import qualified Hasql.Queue.Low.AtLeastOnce as IO
+import qualified Hasql.Queue.High.ExactlyOnce as S
 import           Data.Int
 
 -- TODO need to make sure the number of producers and consumers does not go over the number of connections
@@ -27,14 +28,26 @@
   let connStr = toConnectionString db
   bracket (either (throwIO . userError . show) pure =<< acquire connStr) release f
 
-withSetup :: (Pool Connection -> IO ()) -> IO ()
-withSetup f = do
+durableConfig :: Int -> Config
+durableConfig microseconds = defaultConfig <> mempty
+  { postgresConfigFile =
+      [ ("wal_level", "replica")
+      , ("archive_mode", "on")
+      , ("max_wal_senders", "2")
+      , ("fsync", "on")
+      , ("synchronous_commit", "on")
+      , ("commit_delay", show microseconds)
+      ]
+  }
+
+withSetup :: Int -> Bool -> (Pool Connection -> IO ()) -> IO ()
+withSetup microseconds durable 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
+    let combinedConfig = (if durable then durableConfig microseconds else defaultConfig) <> cacheConfig dbCache
     migratedConfig <- throwE $ cacheAction (("~/.tmp-postgres/" <>) . BSC.unpack . Base64.encode . hash
         $ BSC.pack $ migrationQueryString "int4")
         (flip withConn $ flip migrate "int4")
@@ -51,7 +64,8 @@
 
 main :: IO ()
 main = do
-  [producerCount, consumerCount, time, initialDequeueCount, initialEnqueueCount, batchCount] <- map read <$> getArgs
+  [producerCount, consumerCount, time, initialEnqueueCount, enqueueBatchCount, dequeueBatchCount, notify, durable, microseconds]
+    <- map read <$> getArgs
   -- create a temporary database
   enqueueCounter <- newIORef (0 :: Int)
   dequeueCounter <- newIORef (0 :: Int)
@@ -63,11 +77,16 @@
         putStrLn $ "Enqueue Count: " <> show finalEnqueueCount
         putStrLn $ "Dequeue Count: " <> show finalDequeueCount
 
-  flip finally printCounters $ withSetup $ \pool -> do
+  flip finally printCounters $ withSetup microseconds (1 == durable) $ \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
+    let enqueueAction = if notify > 0
+          then void $ withResource pool $ \conn -> IO.enqueue "channel" conn E.int4 (replicate enqueueBatchCount payload)
+          else void $ withResource pool $ \conn -> I.runThrow (S.enqueue E.int4 (replicate enqueueBatchCount payload)) conn
+        dequeueAction = if notify > 0
+          then void $ withResource pool $ \conn ->
+            IO.withDequeue "channel" conn D.int4 1 dequeueBatchCount (const $ pure ())
+          else void $ withResource pool $ \conn -> fix $ \next ->
+            I.runThrow (S.dequeue D.int4 dequeueBatchCount) conn >>= \case
               [] -> next
               _ -> pure ()
 
@@ -76,12 +95,6 @@
           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"
diff --git a/hasql-queue-tmp-db/Main.hs b/hasql-queue-tmp-db/Main.hs
new file mode 100644
--- /dev/null
+++ b/hasql-queue-tmp-db/Main.hs
@@ -0,0 +1,27 @@
+import Database.Postgres.Temp
+import Control.Concurrent
+import Hasql.Queue.Migrate
+import qualified Data.ByteString.Char8 as BSC
+import qualified Data.ByteString.Base64.URL as Base64
+import           Control.Exception
+import           Control.Monad
+import           Crypto.Hash.SHA1 (hash)
+import           Hasql.Connection
+
+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
+
+
+main :: IO ()
+main = 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")
+        (autoExplainConfig 1  <> cacheConfig dbCache)
+  withConfig migratedConfig $ \db -> do
+    print $ toConnectionString db
+    forever $ threadDelay 100000000
diff --git a/hasql-queue.cabal b/hasql-queue.cabal
--- a/hasql-queue.cabal
+++ b/hasql-queue.cabal
@@ -4,10 +4,10 @@
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: 1e85f1ecfea2da607af0d7d470a2487f040dbe1020c661774ca3ebbaec5fcf34
+-- hash: 0c6f854b16370d7971eb6b012e349140f35a0b01327d5144a2a21335b7def387
 
 name:           hasql-queue
-version:        1.0.1.1
+version:        1.2.0.0
 synopsis:       A PostgreSQL backed queue
 description:    A PostgreSQL backed queue. Please see README.md
 category:       Web
@@ -30,10 +30,14 @@
 
 library
   exposed-modules:
+      Hasql.Queue.High.AtLeastOnce
+      Hasql.Queue.High.AtMostOnce
+      Hasql.Queue.High.ExactlyOnce
       Hasql.Queue.Internal
-      Hasql.Queue.IO
+      Hasql.Queue.Low.AtLeastOnce
+      Hasql.Queue.Low.AtMostOnce
+      Hasql.Queue.Low.ExactlyOnce
       Hasql.Queue.Migrate
-      Hasql.Queue.Session
   other-modules:
       Paths_hasql_queue
   hs-source-dirs:
@@ -64,7 +68,7 @@
   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
+  ghc-options: -Wall -Wno-unused-do-bind -Wno-unused-foralls -O2 -threaded -rtsopts -with-rtsopts=-N
   build-depends:
       aeson
     , async
@@ -88,12 +92,48 @@
     , transformers
   default-language: Haskell2010
 
+executable hasql-queue-tmp-db
+  main-is: Main.hs
+  other-modules:
+      Paths_hasql_queue
+  hs-source-dirs:
+      hasql-queue-tmp-db
+  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-queue
+    , here
+    , monad-control
+    , postgresql-libpq
+    , postgresql-libpq-notify >=0.2.0.0
+    , random
+    , 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
+      Hasql.Queue.High.AtLeastOnceSpec
+      Hasql.Queue.High.AtMostOnceSpec
+      Hasql.Queue.High.ExactlyOnceSpec
+      Hasql.Queue.Low.AtLeastOnceSpec
+      Hasql.Queue.Low.AtMostOnceSpec
+      Hasql.Queue.Low.ExactlyOnceSpec
+      Hasql.Queue.MigrateSpec
+      Hasql.Queue.TestUtils
       Paths_hasql_queue
   hs-source-dirs:
       test
diff --git a/src/Hasql/Queue/High/AtLeastOnce.hs b/src/Hasql/Queue/High/AtLeastOnce.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/High/AtLeastOnce.hs
@@ -0,0 +1,99 @@
+module Hasql.Queue.High.AtLeastOnce where
+import qualified Hasql.Queue.High.ExactlyOnce as H
+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
+
+{-|Enqueue a list of payloads.
+-}
+enqueue :: Connection
+        -- ^ Connection
+        -> E.Value a
+        -- ^ Payload encoder
+        -> [a]
+        -- ^ List of payloads to enqueue
+        -> IO ()
+enqueue conn encoder xs = I.runThrow (H.enqueue 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. '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 'failures'
+to receive the list of failed payloads.
+
+If the queue is empty 'withDequeue' return 'Nothing'. If there are
+any entries 'withDequeue' will wrap the list in 'Just'.
+-}
+withDequeue :: Connection
+            -- ^ Connection
+            -> D.Value a
+            -- ^ Payload decoder
+            -> Int
+            -- ^ Retry count
+            -> Int
+            -- ^ Element count
+            -> ([a] -> IO b)
+            -- ^ Continuation
+            -> IO (Maybe b)
+withDequeue = withDequeueWith @IOError
+
+
+{-|
+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.
+-}
+failures :: 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)]
+failures conn decoder mPayload count = I.runThrow (I.failures decoder mPayload count) conn
+
+
+{-|
+Permantently remove a failed payload.
+-}
+delete :: Connection
+       -> [I.PayloadId]
+       -> IO ()
+delete conn xs = I.runThrow (I.delete xs) conn
+
+{-|
+A more general configurable version of 'withDequeue'. Unlike 'withDequeue' one
+can specify the exception that causes a retry. Additionally event
+handlers can be specified to observe the internal behavior of the
+retry loop.
+-}
+withDequeueWith :: forall e a b
+                 . Exception e
+                => Connection
+                -- ^ Connection
+                -> D.Value a
+                -- ^ Payload decoder
+                -> Int
+                -- ^ Retry count
+                -> Int
+                -- ^ Element count
+                -> ([a] -> IO b)
+                -- ^ Continuation
+                -> IO (Maybe b)
+withDequeueWith conn decoder retryCount count f = (fix $ \restart i -> do
+    try (flip I.runThrow conn $ I.withDequeue decoder retryCount count f) >>= \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/High/AtMostOnce.hs b/src/Hasql/Queue/High/AtMostOnce.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/High/AtMostOnce.hs
@@ -0,0 +1,29 @@
+module Hasql.Queue.High.AtMostOnce where
+import qualified Hasql.Queue.High.ExactlyOnce as H
+import qualified Hasql.Queue.Internal as I
+import           Hasql.Connection
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+
+{-|Enqueue a payload.
+-}
+enqueue :: Connection
+        -- ^ Connection
+        -> E.Value a
+        -- ^ Payload encoder
+        -> [a]
+        -- ^ List of payloads to enqueue
+        -> IO ()
+enqueue conn encoder xs = I.runThrow (H.enqueue encoder xs) conn
+
+{-|
+Dequeue a list of payloads.
+-}
+dequeue :: Connection
+        -- ^ Connection
+        -> D.Value a
+        -- ^ Payload decoder
+        -> Int
+        -- ^ Element count
+        -> IO [a]
+dequeue conn theDecoder batchCount = I.runThrow (H.dequeue theDecoder batchCount) conn
diff --git a/src/Hasql/Queue/High/ExactlyOnce.hs b/src/Hasql/Queue/High/ExactlyOnce.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/High/ExactlyOnce.hs
@@ -0,0 +1,94 @@
+{-|
+A high throughput 'Session' based API for a PostgreSQL backed queue.
+-}
+module Hasql.Queue.High.ExactlyOnce
+  ( enqueue
+  , dequeue
+  ) 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
+
+
+
+{-|Enqueue a payload.
+-}
+enqueue :: E.Value a
+        -- ^ Payload encoder
+        -> [a]
+        -- ^ List of payloads to enqueue
+        -> Session ()
+enqueue theEncoder = \case
+  []  -> pure ()
+  [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
+
+
+{-|
+Dequeue a list of payloads
+-}
+dequeue :: D.Value a
+        -- ^ Payload decoder
+        -> Int
+        -- ^ Element count
+        -> Session [a]
+dequeue valueDecoder count
+  | count <= 0 = pure []
+  | otherwise = do
+  let multipleQuery = [here|
+        DELETE FROM payloads
+        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|
+        DELETE FROM payloads
+        WHERE id =
+          ( 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
diff --git a/src/Hasql/Queue/IO.hs b/src/Hasql/Queue/IO.hs
deleted file mode 100644
--- a/src/Hasql/Queue/IO.hs
+++ /dev/null
@@ -1,128 +0,0 @@
-{-|
-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
--- a/src/Hasql/Queue/Internal.hs
+++ b/src/Hasql/Queue/Internal.hs
@@ -1,3 +1,7 @@
+{-|
+Internal module. Changes to this modules are not reflected in the
+package version.
+-}
 module Hasql.Queue.Internal where
 import qualified Hasql.Encoders as E
 import qualified Hasql.Decoders as D
@@ -12,16 +16,19 @@
 import           Hasql.Statement
 import           Data.ByteString (ByteString)
 import           Control.Exception
-import           Control.Monad.IO.Class
 import           Data.Typeable
 import qualified Database.PostgreSQL.LibPQ as PQ
+import           Data.Maybe
+import           Control.Monad.IO.Class
+import           Data.Text (Text)
+import qualified Data.Text.Encoding as TE
 
 -- | 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
+data State = Enqueued | Failed
   deriving (Show, Eq, Ord, Enum, Bounded)
 
 state :: E.Params a -> D.Result b -> ByteString -> Statement a b
@@ -31,8 +38,6 @@
 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
@@ -40,7 +45,6 @@
 stateEncoder :: E.Value State
 stateEncoder = E.enum $ \case
   Enqueued -> "enqueued"
-  Dequeued -> "dequeued"
   Failed   -> "failed"
 
 initialPayloadId :: PayloadId
@@ -101,8 +105,7 @@
 dequeuePayload :: D.Value a -> Int -> Session [Payload a]
 dequeuePayload valueDecoder count = do
   let multipleQuery = [here|
-        UPDATE payloads
-        SET state='dequeued'
+        DELETE FROM payloads
         WHERE id in
           ( SELECT p1.id
             FROM payloads AS p1
@@ -116,9 +119,8 @@
       multipleEncoder = E.param $ E.nonNullable $ fromIntegral >$< E.int4
 
       singleQuery = [here|
-        UPDATE payloads
-        SET state='dequeued'
-        WHERE id in
+        DELETE FROM payloads
+        WHERE id =
           ( SELECT p1.id
             FROM payloads AS p1
             WHERE p1.state='enqueued'
@@ -178,24 +180,8 @@
 
   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)
@@ -233,17 +219,106 @@
 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)
+data NoRows = NoRows
+  deriving (Show, Eq, Typeable)
+
+instance Exception NoRows
+
+withNotifyWith :: WithNotifyHandlers
+               -> Text
+               -> Connection
+               -> Session a
+               -> IO a
+withNotifyWith WithNotifyHandlers {..} channel conn action = bracket_
+  (flip runThrow conn $ statement channel $ Statement "SELECT listen_on($1)" (E.param $ E.nonNullable E.text) D.noResult True)
+  (flip runThrow conn $ statement channel $ Statement "SELECT unlisten_on($1)" (E.param $ E.nonNullable E.text) D.noResult True)
   $ fix $ \restart -> do
-    x <- runThrow action conn
+    x <- try $ runThrow action conn
     withNotifyHandlersAfterAction
-    case theCast x of
-      Nothing -> do
+    case x of
+      Left NoRows  -> do
         -- TODO record the time here
         withNotifyHandlersBeforeNotification
-        notifyPayload channel conn
+        notifyPayload (TE.encodeUtf8 channel) conn
         restart
-      Just xs -> pure xs
+      Right xs -> pure xs
+
+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
+
+  statement (theState, defaultPayloadId, fromIntegral count) theStatement
+{-|
+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.
+-}
+failures :: D.Value a
+         -- ^ Payload decoder
+         -> Maybe PayloadId
+         -- ^ Starting position of payloads. Pass 'Nothing' to
+         --   start at the beginning
+         -> Int
+         -- ^ Count
+         -> Session [(PayloadId, a)]
+failures = listState Failed
+
+-- 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
+  -- TODO turn to a save point
+  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"
+
+delete :: [PayloadId] -> Session ()
+delete xs = do
+  let theQuery = [here|
+        DELETE FROM payloads
+        WHERE id = ANY($1)
+        |]
+
+      encoder = E.param
+              $ E.nonNullable
+              $ E.foldableArray
+              $ E.nonNullable payloadIdEncoder
+
+  statement xs $ Statement theQuery encoder D.noResult True
diff --git a/src/Hasql/Queue/Low/AtLeastOnce.hs b/src/Hasql/Queue/Low/AtLeastOnce.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/Low/AtLeastOnce.hs
@@ -0,0 +1,125 @@
+{-|
+A high throughput 'Session' based API for a PostgreSQL backed queue.
+-}
+module Hasql.Queue.Low.AtLeastOnce
+  ( enqueue
+  , withDequeue
+  -- ** Listing API
+  , I.PayloadId
+  , failures
+  , delete
+  -- ** Advanced API
+  , withDequeueWith
+  , I.WithNotifyHandlers (..)
+  ) where
+
+import qualified Hasql.Queue.Low.ExactlyOnce as E
+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.Text (Text)
+import           Control.Monad.IO.Class
+
+{-|Enqueue a list of payloads.
+-}
+enqueue :: Text
+        -- ^ 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 (E.enqueue 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. '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 'failures'
+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 :: Text
+            -- ^ 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.
+-}
+failures :: 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)]
+failures conn decoder mPayload count = I.runThrow (I.failures decoder mPayload count) conn
+
+{-|
+Permantently remove a failed payload.
+-}
+delete :: Connection
+       -> [I.PayloadId]
+       -> IO ()
+delete conn xs = I.runThrow (I.delete xs) conn
+
+{-|
+A more general configurable version of 'withDequeue'. Unlike 'withDequeue' one
+can specify the exception that causes a retry. Additionally event
+handlers can be specified to observe the internal behavior of the
+retry loop.
+-}
+withDequeueWith :: forall e a b
+                 . Exception e
+                => I.WithNotifyHandlers
+                -- ^ Event handlers for events that occur as 'withDequeWith' loops
+                -> Text
+                -- ^ 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
+    let action = I.withDequeue decoder retryCount count f >>= \case
+          Nothing -> liftIO $ throwIO I.NoRows
+          Just x  -> pure x
+
+    try (I.withNotifyWith withNotifyHandlers channel conn action) >>= \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/Low/AtMostOnce.hs b/src/Hasql/Queue/Low/AtMostOnce.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/Low/AtMostOnce.hs
@@ -0,0 +1,36 @@
+module Hasql.Queue.Low.AtMostOnce where
+import qualified Hasql.Queue.Low.ExactlyOnce as E
+import           Hasql.Connection
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Data.Text(Text)
+import qualified Hasql.Queue.Internal as I
+
+
+{-|Enqueue a payload.
+-}
+enqueue :: Text
+        -- ^ 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 (E.enqueue channel encoder xs) conn
+
+{-|
+Dequeue a list of payloads.
+-}
+dequeue :: Text
+        -- ^ Notification channel name. Any valid PostgreSQL identifier
+        -> Connection
+        -- ^ Connection
+        -> D.Value a
+        -- ^ Payload decoder
+        -> Int
+        -- ^ Element count
+        -> IO [a]
+dequeue channel conn theDecoder batchCount =
+  E.withDequeue channel conn theDecoder batchCount id
diff --git a/src/Hasql/Queue/Low/ExactlyOnce.hs b/src/Hasql/Queue/Low/ExactlyOnce.hs
new file mode 100644
--- /dev/null
+++ b/src/Hasql/Queue/Low/ExactlyOnce.hs
@@ -0,0 +1,64 @@
+module Hasql.Queue.Low.ExactlyOnce
+  ( enqueue
+  , withDequeue
+  , withDequeueWith
+  ) where
+import qualified Hasql.Queue.High.ExactlyOnce as H
+import Control.Exception
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Hasql.Session
+import           Hasql.Statement
+import           Hasql.Connection
+import qualified Hasql.Queue.Internal as I
+import Control.Monad.IO.Class
+import           Data.Text(Text)
+
+{-|Enqueue a payload send a notification on the
+specified channel.
+-}
+enqueue :: Text
+        -- ^ Notification channel name. Any valid PostgreSQL identifier
+        -> E.Value a
+        -- ^ Payload encoder
+        -> [a]
+        -- ^ List of payloads to enqueue
+        -> Session ()
+enqueue channel theEncoder values = do
+  H.enqueue theEncoder values
+  statement channel $ Statement "SELECT notify_on($1)" (E.param $ E.nonNullable E.text) D.noResult True
+
+dequeueOrRollbackAndThrow :: D.Value a -> Int -> Session [a]
+dequeueOrRollbackAndThrow theDecoder dequeueCount = H.dequeue theDecoder dequeueCount >>= \case
+  [] -> liftIO $ throwIO I.NoRows
+  xs -> pure xs
+
+withDequeue :: Text
+            -- ^ Notification channel name. Any valid PostgreSQL identifier
+            -> Connection
+            -- ^ Connection
+            -> D.Value a
+            -- ^ Payload decoder
+            -> Int
+            -- ^ Batch count
+            -> (Session [a] -> Session b)
+            -- ^ Transaction runner
+            -> IO b
+withDequeue = withDequeueWith mempty
+
+withDequeueWith :: I.WithNotifyHandlers
+                -- ^ Event handlers for events that occur as 'withDequeWith' loops
+                -> Text
+                -- ^ Notification channel name. Any valid PostgreSQL identifier
+                -> Connection
+                -- ^ Connection
+                -> D.Value a
+                -- ^ Payload decoder
+                -> Int
+                -- ^ Batch count
+                -> (Session [a] -> Session b)
+                -- ^ Transaction runner
+                -> IO b
+withDequeueWith withNotifyHandlers channel conn theDecoder dequeueCount runner
+  = I.withNotifyWith withNotifyHandlers channel conn
+  $ runner (dequeueOrRollbackAndThrow theDecoder dequeueCount)
diff --git a/src/Hasql/Queue/Migrate.hs b/src/Hasql/Queue/Migrate.hs
--- a/src/Hasql/Queue/Migrate.hs
+++ b/src/Hasql/Queue/Migrate.hs
@@ -9,7 +9,7 @@
 {-# OPTIONS_HADDOCK prune #-}
 {-# LANGUAGE QuasiQuotes, OverloadedStrings #-}
 module Hasql.Queue.Migrate where
-import           Control.Monad
+import qualified Hasql.Queue.Internal as I
 import           Data.String
 import           Data.String.Here.Interpolated
 import           Hasql.Connection
@@ -24,10 +24,28 @@
                      -- @jsonb@.
                      -> String
 migrationQueryString valueType = [i|
+  CREATE OR REPLACE FUNCTION notify_on(channel text) RETURNs VOID AS $$
+    BEGIN
+      EXECUTE (format(E'NOTIFY %I', channel));
+    END;
+  $$ LANGUAGE plpgsql;
+
+  CREATE OR REPLACE FUNCTION listen_on(channel text) RETURNS VOID AS $$
+    BEGIN
+      EXECUTE (format(E'LISTEN %I', channel));
+    END;
+  $$ LANGUAGE plpgsql;
+
+  CREATE OR REPLACE FUNCTION unlisten_on(channel text) RETURNS VOID AS $$
+    BEGIN
+      EXECUTE (format(E'UNLISTEN %I', channel));
+    END;
+  $$ LANGUAGE plpgsql;
+
     DO $$
   BEGIN
     IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN
-      CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');
+      CREATE TYPE state_t AS ENUM ('enqueued', 'failed');
     END IF;
   END$$;
 
@@ -39,9 +57,9 @@
   , 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)
+  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at, state)
     WHERE (state = 'enqueued');
 
 |]
@@ -51,9 +69,28 @@
 
  @
  DO $$
+  CREATE OR REPLACE FUNCTION notify_on(channel text) RETURNs VOID AS $$
+    BEGIN
+      EXECUTE (format(E'NOTIFY %I', channel));
+    END;
+  $$ LANGUAGE plpgsql;
+
+  CREATE OR REPLACE FUNCTION listen_on(channel text) RETURNS VOID AS $$
+    BEGIN
+      EXECUTE (format(E'LISTEN %I', channel));
+    END;
+  $$ LANGUAGE plpgsql;
+
+  CREATE OR REPLACE FUNCTION unlisten_on(channel text) RETURNS VOID AS $$
+    BEGIN
+      EXECUTE (format(E'UNLISTEN %I', channel));
+    END;
+  $$ LANGUAGE plpgsql;
+
+    DO $$
   BEGIN
     IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'state_t') THEN
-      CREATE TYPE state_t AS ENUM ('enqueued', 'dequeued', 'failed');
+      CREATE TYPE state_t AS ENUM ('enqueued', 'failed');
     END IF;
   END$$;
 
@@ -64,10 +101,10 @@
   , 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);
+  , value ${VALUE_TYPE} NOT NULL
+  );
 
-  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at)
+  CREATE INDEX IF NOT EXISTS active_modified_at_idx ON payloads USING btree (modified_at, state)
     WHERE (state = 'enqueued');
  @
 
@@ -77,5 +114,21 @@
         -> String
         -- ^ The type of the @value@ column
         -> IO ()
-migrate conn valueType = void $
-  run (sql $ fromString $ migrationQueryString valueType) conn
+migrate conn valueType =
+  I.runThrow (sql $ fromString $ migrationQueryString valueType) conn
+
+{-|
+Drop everything created by 'migrate'
+-}
+teardown :: Connection -> IO ()
+teardown conn = do
+  let theQuery = [i|
+        DROP TABLE IF EXISTS payloads;
+        DROP TYPE IF EXISTS state_t;
+        DROP SEQUENCE IF EXISTS modified_index;
+        DROP FUNCTION IF EXISTS notify_on;
+        DROP FUNCTION IF EXISTS listen_on;
+        DROP FUNCTION IF EXISTS unlisten_on;
+      |]
+
+  I.runThrow (sql theQuery) conn
diff --git a/src/Hasql/Queue/Session.hs b/src/Hasql/Queue/Session.hs
deleted file mode 100644
--- a/src/Hasql/Queue/Session.hs
+++ /dev/null
@@ -1,180 +0,0 @@
-{-|
-'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
deleted file mode 100644
--- a/test/Database/Hasql/Queue/IOSpec.hs
+++ /dev/null
@@ -1,206 +0,0 @@
-{-# 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
deleted file mode 100644
--- a/test/Database/Hasql/Queue/SessionSpec.hs
+++ /dev/null
@@ -1,226 +0,0 @@
-{-# 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/Hasql/Queue/High/AtLeastOnceSpec.hs b/test/Hasql/Queue/High/AtLeastOnceSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/High/AtLeastOnceSpec.hs
@@ -0,0 +1,92 @@
+module Hasql.Queue.High.AtLeastOnceSpec where
+import           Hasql.Queue.High.AtLeastOnce
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Test.Hspec                     (Spec, describe, parallel, it)
+import           Test.Hspec.Expectations.Lifted
+import           Test.Hspec.Core.Spec (sequential)
+import           Hasql.Queue.TestUtils
+import qualified Hasql.Queue.Internal as I
+import           Control.Exception as E
+import           Hasql.Connection
+import           Data.Typeable
+import           Data.IORef
+import           Control.Monad
+
+data FailedwithDequeue = FailedwithDequeue
+  deriving (Show, Eq, Typeable)
+
+instance Exception FailedwithDequeue
+
+getPayload :: Connection -> D.Value a -> I.PayloadId -> IO (Maybe (I.Payload a))
+getPayload conn decoder payloadId = I.runThrow (I.getPayload decoder payloadId) conn
+
+spec :: Spec
+spec = describe "Hasql.Queue.High.AtLeastOnce" $ parallel $ do
+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do
+    it "enqueue nothing gives nothing" $ withConnection $ \conn -> do
+      enqueue conn E.int4 []
+      withDequeue conn D.int4 1 1 pure `shouldReturn` Nothing
+
+    it "enqueue 1 gives 1" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1]
+      withDequeue conn D.int4 1 1 pure `shouldReturn` Just [1]
+
+    it "dequeue give nothing after enqueueing everything" $ withConnection $ \conn -> do
+      withDequeue conn D.int4 1 1 pure `shouldReturn` Nothing
+
+    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1]
+      enqueue conn E.int4 [2]
+      withDequeue conn D.int4 1 1 pure `shouldReturn` Just [1]
+      withDequeue conn D.int4 1 1 pure `shouldReturn` Just [2]
+
+    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1, 2, 3]
+      withDequeue conn D.int4 1 2 pure `shouldReturn` Just [1, 2]
+
+      withDequeue conn D.int4 1 2 pure `shouldReturn` Just [3]
+
+    it "withDequeue fails if a non IOError is thrown" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1]
+      handle (\FailedwithDequeue -> pure Nothing) $
+        withDequeue conn D.int4 2 1 $ \_ -> throwIO FailedwithDequeue
+
+      failures conn D.int4 Nothing 1 `shouldReturn` []
+      withDequeue conn D.int4 0 1 pure `shouldReturn` Just [1]
+
+    it "withDequeue fails if throws occur and retry is zero" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1]
+      handle (\(_ :: IOError) -> pure Nothing) $
+        withDequeue conn D.int4 0 1 $ \_ -> throwIO $ userError "hey"
+
+      [(pId, x)] <- failures conn D.int4 Nothing 1
+      x `shouldBe` 1
+      delete conn [pId]
+
+    it "withDequeue succeeds even if the first attempt fails" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1]
+
+      ref <- newIORef (0 :: Int)
+
+      withDequeue conn D.int4 1 1 (\_ -> do
+        count <- readIORef ref
+        writeIORef ref $ count + 1
+        when (count < 1) $ throwIO $ userError "hey"
+        pure '!') `shouldReturn` Just '!'
+
+      withDequeue conn D.int4 1 1 pure `shouldReturn` Nothing
+      readIORef ref `shouldReturn` 2
+
+    it "failures paging works" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [2]
+      enqueue conn E.int4 [3]
+
+      handle (\(_ :: IOError) -> pure Nothing) $
+        withDequeue conn D.int4 0 1 $ \_ -> throwIO $ userError "fds"
+      handle (\(_ :: IOError) -> pure Nothing) $
+        withDequeue conn D.int4 0 1 $ \_ -> throwIO $ userError "fds"
+
+      [(next, x)] <- failures conn D.int4 Nothing 1
+      x `shouldBe` 2
+      fmap (fmap snd) (failures conn D.int4 (Just next) 2) `shouldReturn` [3]
diff --git a/test/Hasql/Queue/High/AtMostOnceSpec.hs b/test/Hasql/Queue/High/AtMostOnceSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/High/AtMostOnceSpec.hs
@@ -0,0 +1,34 @@
+module Hasql.Queue.High.AtMostOnceSpec where
+import           Hasql.Queue.High.AtMostOnce
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Test.Hspec                     (Spec, describe, parallel, it)
+import           Test.Hspec.Expectations.Lifted
+import           Test.Hspec.Core.Spec (sequential)
+import           Hasql.Queue.TestUtils
+
+spec :: Spec
+spec = describe "Hasql.Queue.High.AtMostOnce" $ parallel $ do
+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do
+    it "enqueue nothing gives nothing" $ withConnection $ \conn -> do
+      enqueue conn E.int4 []
+      dequeue conn D.int4 1 `shouldReturn` []
+
+    it "enqueue 1 gives 1" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1]
+      dequeue conn D.int4 1 `shouldReturn` [1]
+
+    it "dequeue give nothing after enqueueing everything" $ withConnection $ \conn -> do
+      dequeue conn D.int4 1 `shouldReturn` []
+
+    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1]
+      enqueue conn E.int4 [2]
+      dequeue conn D.int4 1 `shouldReturn` [1]
+      dequeue conn D.int4 1 `shouldReturn` [2]
+
+    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do
+      enqueue conn E.int4 [1, 2, 3]
+      dequeue conn D.int4 2 `shouldReturn` [1, 2]
+
+      dequeue conn D.int4 2 `shouldReturn` [3]
diff --git a/test/Hasql/Queue/High/ExactlyOnceSpec.hs b/test/Hasql/Queue/High/ExactlyOnceSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/High/ExactlyOnceSpec.hs
@@ -0,0 +1,47 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards   #-}
+{-# LANGUAGE ScopedTypeVariables   #-}
+module Hasql.Queue.High.ExactlyOnceSpec where
+import           Control.Exception as E
+import           Hasql.Queue.High.ExactlyOnce
+import           Test.Hspec                     (Spec, describe, parallel, it)
+import           Test.Hspec.Expectations.Lifted
+import           Test.Hspec.Core.Spec (sequential)
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Data.Typeable
+import           Data.Int
+import           Hasql.Queue.TestUtils
+
+-- Fix this to be more of what I would expec
+
+newtype TooManyRetries = TooManyRetries Int64
+  deriving (Show, Eq, Typeable)
+
+instance Exception TooManyRetries
+
+spec :: Spec
+spec = describe "Hasql.Queue.High.ExactlyOnce" $ parallel $ do
+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do
+    it "enqueue nothing gives nothing" $ withReadCommitted $ do
+      enqueue E.int4 []
+      dequeue D.int4 1 `shouldReturn` []
+
+    it "enqueue 1 gives 1" $ withReadCommitted $ do
+      enqueue E.int4 [1]
+      dequeue D.int4 1 `shouldReturn` [1]
+
+    it "dequeue give nothing after enqueueing everything" $ withReadCommitted $ do
+      dequeue D.int4 1 `shouldReturn` []
+
+    it "dequeueing is in FIFO order" $ withReadCommitted $ do
+      enqueue E.int4 [1]
+      enqueue E.int4 [2]
+      dequeue D.int4 1 `shouldReturn` [1]
+      dequeue D.int4 1 `shouldReturn` [2]
+
+    it "dequeueing a batch of elements works" $ withReadCommitted $ do
+      enqueue E.int4 [1, 2, 3]
+      dequeue D.int4 2 `shouldReturn` [1, 2]
+
+      dequeue D.int4 2 `shouldReturn` [3]
diff --git a/test/Hasql/Queue/Low/AtLeastOnceSpec.hs b/test/Hasql/Queue/Low/AtLeastOnceSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/Low/AtLeastOnceSpec.hs
@@ -0,0 +1,149 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards   #-}
+{-# LANGUAGE ScopedTypeVariables   #-}
+module Hasql.Queue.Low.AtLeastOnceSpec 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.Low.AtLeastOnce
+import           Test.Hspec                     (Spec, describe, it)
+import           Test.Hspec.Expectations.Lifted
+import           Data.List.Split
+import           Data.Text(Text)
+import           Hasql.Connection
+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 (..))
+import           Hasql.Queue.TestUtils
+import           System.Timeout
+
+getCount :: Connection -> IO Int64
+getCount = I.runThrow I.getCount
+
+getPayload :: Connection -> D.Value a -> I.PayloadId -> IO (Maybe (I.Payload a))
+getPayload conn decoder payloadId = I.runThrow (I.getPayload decoder payloadId) conn
+
+channel :: Text
+channel = "hey"
+
+data FailedwithDequeue = FailedwithDequeue
+  deriving (Show, Eq, Typeable)
+
+instance Exception FailedwithDequeue
+
+spec :: Spec
+spec = describe "Hasql.Queue.Low.AtLeastOnce" $ aroundAll withSetup $ describe "enqueue/withDequeue" $ do
+  it "enqueue nothing timesout" $ withConnection $ \conn -> do
+    enqueue channel conn E.int4 []
+    timeout 100000 (withDequeue channel conn D.int4 1 1 pure) `shouldReturn` Nothing
+
+  it "enqueue 1 gives 1" $ withConnection $ \conn -> do
+    enqueue channel conn E.int4 [1]
+    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [1]
+
+  it "dequeue timesout after enqueueing everything" $ withConnection $ \conn -> do
+    timeout 100000 (withDequeue channel conn D.int4 1 1 pure) `shouldReturn` Nothing
+
+  it "dequeueing is in FIFO order" $ withConnection $ \conn -> do
+    enqueue channel conn E.int4 [1]
+    enqueue channel conn E.int4 [2]
+    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [1]
+    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [2]
+
+  it "dequeueing a batch of elements works" $ withConnection $ \conn -> do
+    enqueue channel conn E.int4 [1, 2, 3]
+    withDequeue channel conn D.int4 1 2 pure `shouldReturn` [1, 2]
+
+    withDequeue channel conn D.int4 1 1 pure `shouldReturn` [3]
+
+  it "withDequeue blocks until something is enqueued: before" $ withConnection $ \conn -> do
+    void $ enqueue channel conn E.int4 [1]
+    res <- withDequeue channel conn D.int4 1 1 pure
+    res `shouldBe` [1]
+
+  it "withDequeue blocks until something is enqueued: during" $ withConnection $ \conn -> do
+    afterActionMVar  <- newEmptyMVar
+    beforeNotifyMVar <- newEmptyMVar
+
+    let handlers = I.WithNotifyHandlers
+          { withNotifyHandlersAfterAction        = putMVar afterActionMVar ()
+          , withNotifyHandlersBeforeNotification = takeMVar beforeNotifyMVar
+          }
+
+    -- This is the definition of IO.dequeue
+    resultThread <- async $ withDequeueWith @IOError 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" $ withConnection2 $ \(conn1, conn2) -> do
+    thread <- async $ withDequeue channel conn1 D.int4 1 1 pure
+    timeout 100000 (wait thread) `shouldReturn` Nothing
+
+    enqueue channel conn2 E.int4 [1]
+
+    wait thread `shouldReturn` [1]
+
+  -- TODO redo just using failures
+  it "withDequeue fails and sets the retries to +1" $ withConnection $ \conn -> do
+    enqueue channel conn E.int4 [1]
+    handle (\(_ :: IOError) -> pure ()) $ withDequeue channel conn D.int4 0 1 $ \_ -> throwIO $ userError "hey"
+    xs <- failures conn D.int4 Nothing 1
+
+    map snd xs `shouldBe` [1]
+
+  it "withDequeue succeeds even if the first attempt fails" $ withConnection $ \conn -> do
+    [payloadId] <- I.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` '!'
+
+    getPayload conn D.int4 payloadId `shouldReturn` Nothing
+
+  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] <- I.runThrow (I.enqueuePayload E.int4 [1]) conn
+    Just actual <- getPayload conn D.int4 payloadId
+
+    pValue actual `shouldBe` 1
diff --git a/test/Hasql/Queue/Low/AtMostOnceSpec.hs b/test/Hasql/Queue/Low/AtMostOnceSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/Low/AtMostOnceSpec.hs
@@ -0,0 +1,48 @@
+module Hasql.Queue.Low.AtMostOnceSpec where
+import           Hasql.Queue.Low.AtMostOnce
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Test.Hspec                     (Spec, describe, parallel, it)
+import           Test.Hspec.Expectations.Lifted
+import           Test.Hspec.Core.Spec (sequential)
+import           Hasql.Queue.TestUtils
+import           Data.Text(Text)
+import           System.Timeout
+import           Control.Concurrent.Async
+
+channel :: Text
+channel = "channel"
+
+spec :: Spec
+spec = describe "Hasql.Queue.Low.AtMostOnce" $ parallel $ do
+  sequential $ aroundAll withSetup $ describe "enqueue/dequeue" $ do
+    it "enqueue nothing timesout" $ withConnection $ \conn -> do
+      enqueue channel conn E.int4 []
+      timeout 100000 (dequeue channel conn D.int4 1) `shouldReturn` Nothing
+
+    it "enqueue 1 gives 1" $ withConnection $ \conn -> do
+      enqueue channel conn E.int4 [1]
+      dequeue channel conn D.int4 1 `shouldReturn` [1]
+
+    it "dequeue timesout after enqueueing everything" $ withConnection $ \conn -> do
+      timeout 100000 (dequeue channel conn D.int4 1) `shouldReturn` Nothing
+
+    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do
+      enqueue channel conn E.int4 [1]
+      enqueue channel conn E.int4 [2]
+      dequeue channel conn D.int4 1 `shouldReturn` [1]
+      dequeue channel conn D.int4 1 `shouldReturn` [2]
+
+    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do
+      enqueue channel conn E.int4 [1, 2, 3]
+      dequeue channel conn D.int4 2 `shouldReturn` [1, 2]
+
+      dequeue channel conn D.int4 2 `shouldReturn` [3]
+
+    it "dequeueing blocks until something is enqueued" $ withConnection2 $ \(conn1, conn2) -> do
+      thread <- async $ dequeue channel conn1 D.int4 1
+      timeout 100000 (wait thread) `shouldReturn` Nothing
+
+      enqueue channel conn2 E.int4 [1]
+
+      wait thread `shouldReturn` [1]
diff --git a/test/Hasql/Queue/Low/ExactlyOnceSpec.hs b/test/Hasql/Queue/Low/ExactlyOnceSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/Low/ExactlyOnceSpec.hs
@@ -0,0 +1,85 @@
+module Hasql.Queue.Low.ExactlyOnceSpec where
+import           Control.Exception as E
+import           Hasql.Queue.Low.ExactlyOnce
+import qualified Hasql.Queue.Internal as I
+import           Test.Hspec                     (Spec, describe, it)
+import           Test.Hspec.Expectations.Lifted
+import qualified Hasql.Encoders as E
+import qualified Hasql.Decoders as D
+import           Data.Typeable
+import           Data.Int
+import           Hasql.Queue.TestUtils
+import           System.Timeout
+import           Control.Concurrent.Async
+import           Hasql.Queue.Internal (runThrow)
+import           Control.Concurrent
+import           Control.Monad
+import           Data.Text (Text)
+
+-- Fix this to be more of what I would expec
+
+newtype TooManyRetries = TooManyRetries Int64
+  deriving (Show, Eq, Typeable)
+
+instance Exception TooManyRetries
+
+channel :: Text
+channel = "channel"
+
+spec :: Spec
+spec = describe "Hasql.Queue.High.ExactlyOnce" $ do
+  aroundAll withSetup $ describe "enqueue/withDequeue" $ do
+    it "enqueue nothing timesout" $ withConnection $ \conn -> do
+      runThrow (enqueue channel E.int4 []) conn
+      timeout 100000 (withDequeue channel conn D.int4 1 id) `shouldReturn` Nothing
+
+    it "enqueue 1 gives 1" $ withConnection $ \conn -> do
+      runThrow (enqueue channel E.int4 [1]) conn
+      withDequeue channel conn D.int4 1 id `shouldReturn` [1]
+
+    it "dequeue timesout after enqueueing everything" $ withConnection $ \conn -> do
+      timeout 100000 (withDequeue channel conn D.int4 1 id) `shouldReturn` Nothing
+
+    it "dequeueing is in FIFO order" $ withConnection $ \conn -> do
+      runThrow (enqueue channel E.int4 [1]) conn
+      runThrow (enqueue channel E.int4 [2]) conn
+      withDequeue channel conn D.int4 1 id `shouldReturn` [1]
+      withDequeue channel conn D.int4 1 id `shouldReturn` [2]
+
+    it "dequeueing a batch of elements works" $ withConnection $ \conn -> do
+      runThrow (enqueue channel E.int4 [1, 2, 3]) conn
+      withDequeue channel conn D.int4 1 id `shouldReturn` [1, 2]
+
+      withDequeue channel conn D.int4 1 id `shouldReturn` [3]
+
+    it "withDequeue blocks until something is enqueued: before" $ withConnection $ \conn -> do
+      void $ runThrow (enqueue channel E.int4 [1]) conn
+      res <- withDequeue channel conn D.int4 1 id
+      res `shouldBe` [1]
+
+    it "withDequeue blocks until something is enqueued: during" $ withConnection $ \conn -> do
+      afterActionMVar  <- newEmptyMVar
+      beforeNotifyMVar <- newEmptyMVar
+
+      let handlers = I.WithNotifyHandlers
+            { withNotifyHandlersAfterAction        = putMVar afterActionMVar ()
+            , withNotifyHandlersBeforeNotification = takeMVar beforeNotifyMVar
+            }
+
+      -- This is the definition of IO.dequeue
+      resultThread <- async $ withDequeueWith handlers channel conn D.int4 1 id
+      takeMVar afterActionMVar
+
+      void $ runThrow (enqueue "hey" E.int4 [1]) conn
+
+      putMVar beforeNotifyMVar ()
+
+      wait resultThread `shouldReturn` [1]
+
+    it "withDequeue blocks until something is enqueued: after" $ withConnection2 $ \(conn1, conn2) -> do
+      thread <- async $ withDequeue channel conn1 D.int4 1 id
+      timeout 100000 (wait thread) `shouldReturn` Nothing
+
+      runThrow (enqueue channel E.int4 [1]) conn2
+
+      wait thread `shouldReturn` [1]
diff --git a/test/Hasql/Queue/MigrateSpec.hs b/test/Hasql/Queue/MigrateSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/MigrateSpec.hs
@@ -0,0 +1,57 @@
+module Hasql.Queue.MigrateSpec where
+import Hasql.Queue.Migrate
+import           Hasql.Queue.TestUtils
+import           Test.Hspec                     (Spec, describe, parallel, it, shouldReturn)
+import           Test.Hspec.Core.Spec (sequential)
+import           Control.Monad.IO.Class
+import qualified Hasql.Decoders as D
+import           Hasql.Session
+import           Hasql.Statement
+import           Data.String.Here.Uninterpolated
+import qualified Hasql.Queue.Internal as I
+
+
+spec :: Spec
+spec = describe "Hasql.Queue.High.ExactlyOnce" $ parallel $ do
+  sequential $ aroundAll withSetup $ describe "basic" $ do
+    it "is okay to migrate multiple times" $ withConnection $ \conn ->
+      liftIO $ migrate conn "int4"
+
+    it "drops all items" $ withConnection $ \conn -> do
+      teardown conn
+
+      let theQuery = [here|
+            SELECT EXISTS (
+            SELECT FROM information_schema.tables
+            WHERE  table_schema = 'public'
+            AND    table_name   = 'payloads'
+            );
+          |]
+
+          decoder = D.singleRow $ D.column $ D.nonNullable D.bool
+      I.runThrow (statement () $ Statement theQuery mempty decoder True) conn `shouldReturn` False
+
+      let theQuery' = [here|
+             SELECT EXISTS (
+             SELECT FROM pg_catalog.pg_class c
+             JOIN   pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+             WHERE  n.nspname = 'public'
+             AND    c.relname = 'modified_index'
+          )
+          |]
+
+          decoder' = D.singleRow $ D.column $ D.nonNullable D.bool
+      I.runThrow (statement () $ Statement theQuery' mempty decoder' True) conn `shouldReturn` False
+
+      let theQuery'' = [here|
+             SELECT EXISTS (
+             SELECT 1
+             FROM pg_type t
+             JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace
+             WHERE t.typname = 'state_t'
+             AND n.nspname = 'public'
+          )
+          |]
+
+          decoder'' = D.singleRow $ D.column $ D.nonNullable D.bool
+      I.runThrow (statement () $ Statement theQuery'' mempty decoder'' True) conn `shouldReturn` False
diff --git a/test/Hasql/Queue/TestUtils.hs b/test/Hasql/Queue/TestUtils.hs
new file mode 100644
--- /dev/null
+++ b/test/Hasql/Queue/TestUtils.hs
@@ -0,0 +1,89 @@
+module Hasql.Queue.TestUtils where
+import qualified Data.ByteString.Char8 as BSC
+import Control.Concurrent.Async
+import           Database.Postgres.Temp as Temp
+import           Test.Hspec
+import           Control.Exception as E
+import           Data.Pool
+import           Hasql.Connection
+import           Hasql.Session
+import qualified Data.ByteString.Base64.URL as Base64
+import           Control.Concurrent
+import           Data.IORef
+import           Data.Foldable
+import           Control.Monad ((<=<))
+import           Crypto.Hash.SHA1 (hash)
+import           Hasql.Queue.Migrate
+
+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
+
+withConnection2 :: ((Connection, Connection) -> IO ()) -> Pool Connection -> IO ()
+withConnection2 f pool = withResource pool $ \conn1 ->
+   withResource pool $ \conn2 -> f (conn1, conn2)
+
+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
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -1,9 +1,19 @@
 import           Test.Hspec
-import           Database.Hasql.Queue.SessionSpec as S
-import           Database.Hasql.Queue.IOSpec as I
+import           Hasql.Queue.High.ExactlyOnceSpec as HE
+import           Hasql.Queue.High.AtLeastOnceSpec as HL
+import           Hasql.Queue.High.AtMostOnceSpec as HM
+import           Hasql.Queue.Low.AtLeastOnceSpec as LL
+import           Hasql.Queue.Low.AtMostOnceSpec as LM
+import           Hasql.Queue.Low.AtMostOnceSpec as LE
+import           Hasql.Queue.MigrateSpec as M
 
 
 main :: IO ()
 main = hspec $ do
-  S.spec
-  I.spec
+  HE.spec
+  HM.spec
+  HL.spec
+  LL.spec
+  LM.spec
+  LE.spec
+  M.spec