diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,30 @@
+Copyright Author name here (c) 2017
+
+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,28 @@
+# postgresql-simple-queue
+
+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
+
+```haskell
+createAccount userRecord = do
+  runDBTSerializable $ do
+    createUserDB userRecord
+    enqueueDB $ makeVerificationEmail userRecord
+```
+
+In another thread or process, the consumer would drain the queue.
+
+```haskell
+  forever $ do
+    -- Attempt get a payload or block until one is available
+    payload <- lock 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 conn $ pId payload
+```
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/examples/EmailQueue.hs b/examples/EmailQueue.hs
new file mode 100644
--- /dev/null
+++ b/examples/EmailQueue.hs
@@ -0,0 +1,48 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE DeriveGeneric     #-}
+{-# LANGUAGE DeriveAnyClass    #-}
+{-# LANGUAGE RecordWildCards   #-}
+import           Control.Exception.Lifted
+import           Control.Lens
+import           Control.Monad
+import           Control.Monad.IO.Class
+import           Data.Aeson as Aeson
+import           Data.Text
+import           Database.PostgreSQL.Simple.Queue
+import           Database.PostgreSQL.Simple.Queue.Main
+import           GHC.Generics
+import           Network.AWS               as AWS
+import           Network.AWS.SES.SendEmail as AWS
+import           Network.AWS.SES.Types     as AWS
+
+main :: IO ()
+main = do
+  env <- newEnv Discover
+  runResourceT $ runAWS env $ defaultMain "aws-email-queue-consumer" $ \payload _ -> do
+    case fromJSON $ pValue payload of
+      Aeson.Success email -> do
+        resp <- AWS.send $ makeEmail email
+        logFailedRequest resp
+      Aeson.Error x -> throwIO
+                     $ userError
+                     $ "Failed to decode payload as an Email: " ++ show x
+
+data Email = Email
+  { emailAddress :: Text
+  , emailSubject :: Text
+  , emailBody    :: Text
+  } deriving (Show, Eq, Generic, FromJSON, ToJSON)
+
+makeEmail :: Email -> SendEmail
+makeEmail Email {..}
+  = sendEmail emailAddress
+              (set dToAddresses [emailAddress] destination)
+  $ message (content emailSubject)
+  $ set bText (Just $ content emailBody) AWS.body
+
+logFailedRequest :: MonadIO m => SendEmailResponse -> m ()
+logFailedRequest resp = do
+    let stat = view sersResponseStatus resp
+
+    unless (stat >= 200 && stat < 300) $
+      liftIO $ putStrLn $ "SES failed with status: " ++ show stat
diff --git a/postgresql-simple-queue.cabal b/postgresql-simple-queue.cabal
new file mode 100644
--- /dev/null
+++ b/postgresql-simple-queue.cabal
@@ -0,0 +1,102 @@
+name:                postgresql-simple-queue
+version:             0.1.0.0
+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' $ 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' 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' conn $ 'pId' payload
+homepage:            https://github.com/jfischoff/postgresql-queue#readme
+license:             BSD3
+license-file:        LICENSE
+author:              Jonathan Fischoff
+maintainer:          jonathangfischoff@gmail.com
+copyright:           2017 Jonathan Fischoff
+category:            Web
+build-type:          Simple
+extra-source-files:  README.md
+cabal-version:       >=1.10
+
+library
+  hs-source-dirs: src
+  exposed-modules: Database.PostgreSQL.Simple.Queue
+                 , Database.PostgreSQL.Simple.Queue.Examples.EmailQueue
+                 , Database.PostgreSQL.Simple.Queue.Migrate
+                 , Database.PostgreSQL.Simple.Queue.Main
+  build-depends: base >= 4.7 && < 5
+               , postgresql-simple
+               , pg-transact
+               , aeson
+               , time
+               , uuid
+               , transformers
+               , random
+               , text
+               , monad-control
+               , postgresql-simple-opts
+               , resource-pool
+               , optparse-generic
+               , data-default
+               , monad-control
+               , lifted-base
+               , lifted-async
+               , exceptions
+               , bytestring
+  default-language:    Haskell2010
+  ghc-options: -Wall -Wno-unused-do-bind
+
+executable async-email-example
+  hs-source-dirs:      examples
+  main-is:             EmailQueue.hs
+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N
+  build-depends: base
+               , postgresql-simple-queue
+               , amazonka
+               , amazonka-ses
+               , lens
+               , text
+               , aeson
+               , lifted-base
+  default-language:    Haskell2010
+
+test-suite db-testing-example-test
+  type:                exitcode-stdio-1.0
+  hs-source-dirs:      test
+  main-is: Spec.hs
+  other-modules: Database.PostgreSQL.Simple.QueueSpec
+  build-depends: base
+               , postgresql-simple-queue
+               , hspec
+               , hspec-discover
+               , postgresql-simple
+               , pg-transact
+               , bytestring
+               , aeson
+               , hspec-expectations-lifted
+               , hspec-pg-transact
+               , async
+  ghc-options: -Wall -Wno-unused-do-bind -O2 -threaded -rtsopts -with-rtsopts=-N
+  default-language:    Haskell2010
+
+source-repository head
+  type:     git
+  location: https://github.com/jfischoff/postgresql-queue
diff --git a/src/Database/PostgreSQL/Simple/Queue.hs b/src/Database/PostgreSQL/Simple/Queue.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Simple/Queue.hs
@@ -0,0 +1,282 @@
+{-| 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' $ 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' 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' conn $ 'pId' payload
+ @
+
+ For a more complete example or a consumer, utilizing the provided
+ 'Database.PostgreSQL.Simple.Queue.Main.defaultMain', see
+ 'Database.PostgreSQL.Simple.Queue.Examples.EmailQueue.EmailQueue'.
+
+This modules provides two flavors of functions, a DB API and an IO API.
+Most operations are provided in both flavors, with the exception of 'lock'.
+'lock' blocks and would not be that useful as part of a larger transaction
+since it would keep the transaction open for a potentially long time. Although
+both flavors are provided, in general one versions is more useful for typical
+use cases.
+
+-}
+{-# LANGUAGE FlexibleContexts           #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE LambdaCase                 #-}
+{-# LANGUAGE OverloadedStrings          #-}
+{-# LANGUAGE QuasiQuotes                #-}
+{-# LANGUAGE RecordWildCards            #-}
+{-# LANGUAGE ScopedTypeVariables        #-}
+module Database.PostgreSQL.Simple.Queue
+  ( -- * Types
+    PayloadId (..)
+  , State (..)
+  , Payload (..)
+  -- * DB API
+  , enqueueDB
+  , tryLockDB
+  , unlockDB
+  , dequeueDB
+  , getCountDB
+  -- * IO API
+  , enqueue
+  , tryLock
+  , lock
+  , unlock
+  , dequeue
+  , getCount
+  ) where
+import           Control.Monad
+import           Control.Monad.Catch
+import           Control.Monad.IO.Class
+import           Data.Aeson
+import qualified Data.ByteString                         as BS
+import           Data.Function
+import           Data.Int
+import           Data.Maybe
+import           Data.Text                               (Text)
+import           Data.Time
+import           Data.UUID
+import           Database.PostgreSQL.Simple              (Connection, Only (..))
+import qualified Database.PostgreSQL.Simple              as Simple
+import           Database.PostgreSQL.Simple.FromField
+import           Database.PostgreSQL.Simple.FromRow
+import           Database.PostgreSQL.Simple.Notification
+import           Database.PostgreSQL.Simple.SqlQQ
+import           Database.PostgreSQL.Simple.ToField
+import           Database.PostgreSQL.Simple.ToRow
+import           Database.PostgreSQL.Simple.Transaction
+import           Database.PostgreSQL.Transact
+import           System.Random
+-------------------------------------------------------------------------------
+---  Types
+-------------------------------------------------------------------------------
+newtype PayloadId = PayloadId { unPayloadId :: UUID }
+  deriving (Eq, Show, FromField, ToField)
+
+instance FromRow PayloadId where
+  fromRow = fromOnly <$> fromRow
+
+instance ToRow PayloadId where
+  toRow = toRow . Only
+
+-- | A 'Payload' can exist in three states in the queue, 'Enqueued', 'Locked'
+--   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 | Locked | Dequeued
+  deriving (Show, Eq, Ord, Enum, Bounded)
+
+instance ToField State where
+  toField = toField . \case
+    Enqueued -> "enqueued" :: Text
+    Locked   -> "locked"
+    Dequeued -> "dequeued"
+
+-- Converting from enumerations is annoying :(
+instance FromField State where
+  fromField f y = do
+     n <- typename f
+     if n == "state_t" then case y of
+       Nothing -> returnError UnexpectedNull f "state can't be NULL"
+       Just y' -> case y' of
+         "enqueued" -> return Enqueued
+         "locked"   -> return Locked
+         "dequeued" -> return Dequeued
+         x          -> returnError ConversionFailed f (show x)
+     else
+       returnError Incompatible f $
+         "Expect type name to be state but it was " ++ show n
+
+-- The fundemental record stored in the queue. The queue is a single table
+-- and each row consists of a 'Payload'
+data Payload = Payload
+  { pId         :: PayloadId
+  , pValue      :: Value
+  -- ^ The JSON value of a payload
+  , pState      :: State
+  , pCreatedAt  :: UTCTime
+  , pModifiedAt :: UTCTime
+  } deriving (Show, Eq)
+
+instance FromRow Payload where
+  fromRow = Payload <$> field <*> field <*> field <*> field <*> field
+-------------------------------------------------------------------------------
+---  DB API
+-------------------------------------------------------------------------------
+retryOnUniqueViolation :: MonadCatch m => m a -> m a
+retryOnUniqueViolation act = try act >>= \case
+  Right x -> return x
+  Left e ->
+    if Simple.sqlState e == "23505" &&
+       "duplicate key" `BS.isPrefixOf` Simple.sqlErrorMsg e then
+      act
+    else
+      throwM e
+
+{-| Enqueue a new JSON value into the queue. This particularly function
+    can be composed as part of a larger database transaction. For instance,
+    a single transaction could create a user and enqueue a email message.
+
+ @
+   createAccount userRecord = do
+      'runDBTSerializable' $ do
+         createUserDB userRecord
+         'enqueueDB' $ makeVerificationEmail userRecord
+ @
+-}
+enqueueDB :: Value -> DB PayloadId
+enqueueDB value = retryOnUniqueViolation $ do
+  pid <- liftIO randomIO
+  execute [sql| INSERT INTO payloads (id, value)
+                VALUES (?, ?);
+                NOTIFY enqueue;
+          |]
+          (pid, value)
+  return $ PayloadId pid
+
+{-| Return a the oldest 'Payload' in the 'Enqueued' state, or 'Nothing'
+    if there are no payloads. This function is not necessarily useful by
+    itself, since there are not many use cases where it needs to be combined
+    with other transactions. See 'tryLock' the IO API version, or for a
+    blocking version utilizing PostgreSQL's NOTIFY and LISTEN, see 'lock'
+-}
+tryLockDB :: DB (Maybe Payload)
+tryLockDB = listToMaybe <$> query_
+  [sql| UPDATE payloads
+        SET state='locked'
+        WHERE id in
+          ( SELECT id
+            FROM payloads
+            WHERE state='enqueued'
+            ORDER BY created_at ASC
+            LIMIT 1
+          )
+        RETURNING id, value, state, created_at, modified_at
+  |]
+
+{-| Transition a 'Payload' from the 'Locked' state to the 'Enqueued' state.
+    Useful for responding to asynchronous exceptions during a unexpected
+    shutdown. In general the IO API version, 'unlock', is probably more
+    useful. The DB version is provided for completeness.
+-}
+unlockDB :: PayloadId -> DB ()
+unlockDB payloadId = void $ execute
+  [sql| UPDATE payloads
+        SET state='enqueued'
+        WHERE id=? AND state='locked'
+  |]
+  payloadId
+
+-- | Transition a 'Payload' to the 'Dequeued' state.
+dequeueDB :: PayloadId -> DB ()
+dequeueDB payloadId = void $ execute
+  [sql| UPDATE payloads
+        SET state='dequeued'
+        WHERE id=?
+  |]
+  payloadId
+
+-- | Get the number of rows in the 'Enqueued' state.
+getCountDB :: DB Int64
+getCountDB = fromOnly . head <$> query_
+  [sql| SELECT count(*)
+        FROM payloads
+        WHERE state='enqueued'
+  |]
+-------------------------------------------------------------------------------
+---  IO API
+-------------------------------------------------------------------------------
+{-| Enqueue a new JSON value into the queue. See 'enqueueDB' for a version
+    which can be composed with other queries in a single transaction.
+-}
+enqueue :: Connection -> Value -> IO PayloadId
+enqueue conn value = runDBT (enqueueDB value) ReadCommitted conn
+
+{-| Return a the oldest 'Payload' in the 'Enqueued' state or 'Nothing'
+    if there are no payloads. For a blocking version utilizing PostgreSQL's
+    NOTIFY and LISTEN, see 'lock'. This functions runs 'tryLockDB' as a
+    'Serializable' transaction.
+-}
+tryLock :: Connection -> IO (Maybe Payload)
+tryLock = runDBTSerializable tryLockDB
+
+notifyPayload :: Connection -> IO ()
+notifyPayload conn = do
+  Notification {..} <- getNotification conn
+  unless (notificationChannel == "enqueue") $ notifyPayload conn
+
+{-| Return the oldest 'Payload' in the 'Enqueued' state or block until a
+    payload arrives. This function utilizes PostgreSQL's LISTEN and NOTIFY
+    functionality to avoid excessively polling of the DB while
+    waiting for new payloads, without scarficing promptness.
+-}
+lock :: Connection -> IO Payload
+lock conn = bracket_
+  (Simple.execute_ conn "LISTEN enqueue")
+  (Simple.execute_ conn "UNLISTEN enqueue")
+  $ fix $ \continue -> do
+      m <- tryLock conn
+      case m of
+        Nothing -> do
+          notifyPayload conn
+          continue
+        Just x -> return x
+
+{-| Transition a 'Payload' from the 'Locked' state to the 'Enqueued' state.
+    Useful for responding to asynchronous exceptions during a unexpected
+    shutdown. For a DB API version see 'unlockDB'
+-}
+unlock :: Connection -> PayloadId -> IO ()
+unlock conn x = runDBTSerializable (unlockDB x) conn
+
+-- | Transition a 'Payload' to the 'Dequeued' state. his functions runs
+--   'dequeueDB' as a 'Serializable' transaction.
+dequeue :: Connection -> PayloadId -> IO ()
+dequeue conn x = runDBTSerializable (dequeueDB x) conn
+
+{-| Get the number of rows in the 'Enqueued' state. This function runs
+    'getCountDB' in a 'ReadCommitted' transaction.
+-}
+getCount :: Connection -> IO Int64
+getCount = runDBT getCountDB ReadCommitted
diff --git a/src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs b/src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Simple/Queue/Examples/EmailQueue.hs
@@ -0,0 +1,59 @@
+{-|
+ This is a documentation only module that exposes the source for
+ the async-email-example. The main file is located in examples/EmailQueue
+ and is reprinted here to show an example of using 'defaultMain' from
+ 'Database.PostgreSQL.Simple.Queue.Main' to quickly build a queue consumer executable.
+
+ @
+ import           Control.Exception.Lifted
+ import           Control.Lens
+ import           Control.Monad
+ import           Control.Monad.IO.Class
+ import           Data.Aeson as Aeson
+ import           Data.Text
+ import           Database.PostgreSQL.Simple.Queue
+ import           Database.PostgreSQL.Simple.Queue.Main
+ import           GHC.Generics
+ import           Network.AWS               as AWS
+ import           Network.AWS.SES.SendEmail as AWS
+ import           Network.AWS.SES.Types     as AWS
+
+ main :: IO ()
+ main = do
+   env <- newEnv Discover
+   runResourceT $ runAWS env $ 'defaultMain' "aws-email-queue-consumer" $ \payload _ -> do
+     case fromJSON $ 'pValue' payload of
+       Aeson.Success email -> do
+         resp <- AWS.send $ makeEmail email
+         logFailedRequest resp
+       Aeson.Error x -> throwIO
+                      $ userError
+                      $ "Failed to decode payload as an Email: " ++ show x
+
+ data Email = Email
+   { emailAddress :: Text
+   , emailSubject :: Text
+   , emailBody    :: Text
+   } deriving (Show, Eq, Generic, FromJSON, ToJSON)
+
+ makeEmail :: Email -> SendEmail
+ makeEmail Email {..}
+   = sendEmail emailAddress
+               (set dToAddresses [emailAddress] destination)
+   $ message (content emailSubject)
+   $ set bText (Just $ content emailBody) AWS.body
+
+ logFailedRequest :: MonadIO m => SendEmailResponse -> m ()
+ logFailedRequest resp = do
+     let stat = view sersResponseStatus resp
+
+     unless (stat >= 200 && stat < 300) $
+       liftIO $ putStrLn $ "SES failed with status: " ++ show stat
+ @
+
+-}
+{-# OPTIONS_GHC -w #-}
+module Database.PostgreSQL.Simple.Queue.Examples.EmailQueue where
+import Database.PostgreSQL.Simple.Queue
+import Database.PostgreSQL.Simple.Queue.Main
+
diff --git a/src/Database/PostgreSQL/Simple/Queue/Main.hs b/src/Database/PostgreSQL/Simple/Queue/Main.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Simple/Queue/Main.hs
@@ -0,0 +1,175 @@
+{-# LANGUAGE DeriveDataTypeable    #-}
+{-# LANGUAGE FlexibleContexts      #-}
+{-# LANGUAGE LambdaCase            #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE NamedFieldPuns        #-}
+{-# LANGUAGE RecordWildCards       #-}
+{-# LANGUAGE ScopedTypeVariables   #-}
+{-# LANGUAGE OverloadedStrings     #-}
+{-|
+ This module provides a simple way to create executables for consuming queue
+ payloads. It provides a `defaultMain' function which takes in a executable
+ name and processing function. It returns a main function which will parse
+ database options on from the command line and spawn a specified number of
+ worker threads.
+
+ Here is a simple example that logs out queue payloads
+
+ @
+  defaultMain "queue-logger" $ \payload count -> do
+    print payload
+    print count
+ @
+
+ The worker threads do not attempt to handle exceptions. If an exception is
+ thrown on any threads, all threads are cancel and 'defaultMain' returns. The
+ assumption is the queue executable will get run by a process watcher that
+ can log failures.
+
+ For a more complicated example, see the code for the async-email-example
+ documented in 'Database.PostgreSQL.Simple.Queue.Examples.EmailQueue.EmailQueue'.
+-}
+module Database.PostgreSQL.Simple.Queue.Main
+  ( -- * Options
+    PartialOptions
+  , Options
+  , completeOptions
+    -- * Entry Points
+  , defaultMain
+  , run
+  ) where
+import           Control.Concurrent.Async.Lifted
+import           Control.Monad
+import           Control.Monad.IO.Class
+import           Control.Monad.Trans.Control
+import           Data.Default
+import           Data.Int
+import           Data.Monoid
+import           Data.Pool
+import           Data.Typeable
+import           Database.PostgreSQL.Simple
+import qualified Database.PostgreSQL.Simple.Options as PostgreSQL
+import           Database.PostgreSQL.Simple.Queue
+import           Options.Generic
+import           System.Exit
+
+{-| The 'PartialOptions' provide a 'Monoid' for combining options used by
+    'defaultMain'. The following command line arguments are used to
+    construct a 'PartialOptions'
+
+ @
+   --thread-count INT
+
+   Either a connection string
+   --connectString ARG
+   or individual db properties are provided
+   --host STRING
+   --port INT
+   --user STRING
+   --password STRING
+   --database STRING
+ @
+-}
+data PartialOptions = PartialOptions
+  { threadCount :: Last Int
+  , dbOptions   :: PostgreSQL.PartialOptions
+  } deriving (Show, Eq, Typeable)
+
+instance Monoid PartialOptions where
+  mempty = PartialOptions mempty mempty
+  mappend x y =
+    PartialOptions
+      (threadCount x <> threadCount y)
+      (dbOptions   x <> dbOptions   y)
+
+-- | The default 'threadCount' is 1.
+--   The default db options are specified in
+--   'Database.PostgreSQL.Simple.Options.PartialOptions'
+instance Default PartialOptions where
+  def = PartialOptions (return 1) def
+
+instance ParseRecord PartialOptions where
+  parseRecord
+     =  PartialOptions
+    <$> parseFields Nothing (Just "thread-count")
+    <*> parseRecord
+
+-- | Final Options used by 'run'.
+data Options = Options
+  { oThreadCount :: Int
+  , oDBOptions   :: PostgreSQL.Options
+  } deriving (Show, Eq)
+
+-- | Convert a 'PartialOptions' to a final 'Options'
+completeOptions :: PartialOptions -> Either [String] Options
+completeOptions = \case
+  PartialOptions { threadCount = Last (Just oThreadCount), dbOptions } ->
+    Options oThreadCount <$> PostgreSQL.completeOptions dbOptions
+  _ -> Left ["Missing threadCount"]
+
+{-| This function is a helper for creating queue consumer executables.
+ It takes in a executable
+ name and processing function. It returns a main function which will parse
+ database options on from the command line and spawn a specified number of
+ worker threads. See 'PartialOptions' for command line documentation.
+
+ Here is a simple example that logs out queue payloads
+
+ @
+  defaultMain "queue-logger" $ \payload count -> do
+    print payload
+    print count
+ @
+
+ The worker threads do not attempt to handle exceptions. If an exception is
+ thrown on any threads, all threads are cancel and 'defaultMain' returns. The
+ assumption is the queue executable will get run by a process watcher that
+ can log failures.
+
+ For a more complicated example, see the code for the async-email-example
+ documented in 'Database.PostgreSQL.Simple.Queue.Examples.EmailQueue.EmailQueue'.
+
+-}
+defaultMain :: (MonadIO m, MonadBaseControl IO m)
+            => Text
+            -- ^ Executable name. Used when command line help
+            --   is printed
+            -> (Payload -> Int64 -> m ())
+            -- ^ Processing function. Takes a 'Payload' to process
+            --   and the current number of 'Enqueued' 'Payload's
+            --   remaining.
+            -> m ()
+defaultMain name f = do
+  poptions <- liftIO $ getRecord name
+  options  <- liftIO
+            $ either (\err -> putStrLn (unlines err) >> exitWith (ExitFailure 64))
+                     return
+                     $ completeOptions $ def <> poptions
+  run f options
+
+-- | 'run' is called by 'defaultMain' after command line argument parsing
+--   is performed. Useful is wants to embed consumer threads inside a
+--   larger application.
+run :: forall m. (MonadIO m, MonadBaseControl IO m)
+    => (Payload -> Int64 -> m ())
+    -- ^ Processing function. Takes a 'Payload' to process
+    --   and the current number of 'Enqueued' 'Payload's
+    --   remaining.
+    -> Options
+    -- ^ Options for thread creation and db connections.
+    -> m ()
+run f Options {..} = do
+  -- Creates a pool with half as many connections as threads. Should
+  -- probably make the number of connection configurable.
+  connectionPool <- liftIO $ createPool
+    (PostgreSQL.run oDBOptions) close 1 60 (max 1 $ oThreadCount `div` 2)
+
+  threads :: [Async (StM m ())] <- replicateM oThreadCount $ async $ void $
+    forever $ do
+      payload <- liftIO $ withResource connectionPool lock
+      count   <- liftIO $ withResource connectionPool getCount
+      f payload count
+      liftIO $ withResource connectionPool $ flip dequeue (pId payload)
+
+  _ :: (Async (StM m ()), ()) <- waitAnyCancel threads
+  return ()
diff --git a/src/Database/PostgreSQL/Simple/Queue/Migrate.hs b/src/Database/PostgreSQL/Simple/Queue/Migrate.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/PostgreSQL/Simple/Queue/Migrate.hs
@@ -0,0 +1,65 @@
+{-# LANGUAGE QuasiQuotes #-}
+module Database.PostgreSQL.Simple.Queue.Migrate where
+import           Control.Monad
+import           Database.PostgreSQL.Simple
+import           Database.PostgreSQL.Simple.SqlQQ
+
+{-| This function creates a table and enumeration type that is
+    appriopiate for the queue. The following sql is used.
+
+ @
+ CREATE TYPE state_t AS ENUM ('enqueued', 'locked', 'dequeued');
+
+ CREATE TABLE payloads
+ ( id uuid PRIMARY KEY
+ , value jsonb NOT NULL
+ , state state_t NOT NULL DEFAULT 'enqueued'
+ , created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()
+ , modified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()
+ );
+
+ CREATE INDEX state_idx ON payloads (state);
+
+ CREATE OR REPLACE FUNCTION update_row_modified_function_()
+ RETURNS TRIGGER
+ AS
+ $$
+ BEGIN
+     -- ASSUMES the table has a column named exactly "modified_at".
+     -- Fetch date-time of actual current moment from clock,
+     -- rather than start of statement or start of transaction.
+     NEW.modified_at = clock_timestamp();
+     RETURN NEW;
+ END;
+ $$
+ language 'plpgsql';
+ @
+-}
+migrate :: Connection -> IO ()
+migrate conn = void $ execute_ conn [sql|
+  CREATE TYPE state_t AS ENUM ('enqueued', 'locked', 'dequeued');
+
+  CREATE TABLE payloads
+  ( id uuid PRIMARY KEY
+  , value jsonb NOT NULL
+  , state state_t NOT NULL DEFAULT 'enqueued'
+  , created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()
+  , modified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT clock_timestamp()
+  );
+
+  CREATE INDEX state_idx ON payloads (state);
+
+  CREATE OR REPLACE FUNCTION update_row_modified_function_()
+  RETURNS TRIGGER
+  AS
+  $$
+  BEGIN
+      -- ASSUMES the table has a column named exactly "modified_at".
+      -- Fetch date-time of actual current moment from clock,
+      -- rather than start of statement or start of transaction.
+      NEW.modified_at = clock_timestamp();
+      RETURN NEW;
+  END;
+  $$
+  language 'plpgsql';
+  |]
diff --git a/test/Database/PostgreSQL/Simple/QueueSpec.hs b/test/Database/PostgreSQL/Simple/QueueSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Database/PostgreSQL/Simple/QueueSpec.hs
@@ -0,0 +1,94 @@
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE RecordWildCards   #-}
+module Database.PostgreSQL.Simple.QueueSpec (spec, main) where
+import           Control.Concurrent
+import           Control.Concurrent.Async
+import           Control.Monad
+import           Data.Aeson
+import           Data.Function
+import           Data.IORef
+import           Data.List
+import           Database.PostgreSQL.Simple.Queue
+import           Database.PostgreSQL.Simple.Queue.Migrate
+import           Test.Hspec                     (Spec, hspec, it)
+import           Test.Hspec.Expectations.Lifted
+import           Test.Hspec.DB
+
+main :: IO ()
+main = hspec spec
+
+spec :: Spec
+spec = describeDB migrate "Database.Queue" $ do
+  itDB "empty locks nothing" $
+    tryLockDB `shouldReturn` Nothing
+
+  itDB "empty gives count 0" $
+    getCountDB `shouldReturn` 0
+
+  itDB "enqueues/tryLocks/unlocks" $ do
+    payloadId <- enqueueDB $ String "Hello"
+    getCountDB `shouldReturn` 1
+
+    Just Payload {..} <- tryLockDB
+    getCountDB `shouldReturn` 0
+
+    pId `shouldBe` payloadId
+    pValue `shouldBe` String "Hello"
+    tryLockDB `shouldReturn` Nothing
+
+    unlockDB pId
+    getCountDB `shouldReturn` 1
+
+  itDB "tryLocks/dequeues" $ do
+    Just Payload {..} <- tryLockDB
+    getCountDB `shouldReturn` 0
+
+    dequeueDB pId `shouldReturn` ()
+    tryLockDB `shouldReturn` Nothing
+
+  it "enqueues and dequeues concurrently tryLock" $ \testDB -> do
+    let withPool' = withPool testDB
+        elementCount = 1000 :: Int
+        expected = [0 .. elementCount - 1]
+
+    ref <- newIORef []
+
+    loopThreads <- replicateM 10 $ async $ fix $ \next -> do
+      mpayload <- withPool' tryLock
+      case mpayload of
+        Nothing -> next
+        Just Payload {..}  -> do
+          lastCount <- atomicModifyIORef ref
+                     $ \xs -> (pValue : xs, length xs + 1)
+          withPool' $ flip dequeue pId
+          when (lastCount < elementCount) next
+
+    -- Fork a hundred threads and enqueue an index
+    forM_ [0 .. elementCount - 1] $ \i ->
+      forkIO $ void $ withPool' $ flip enqueue $ toJSON i
+
+    waitAnyCancel loopThreads
+    Just decoded <- mapM (decode . encode) <$> readIORef ref
+    sort decoded `shouldBe` sort expected
+
+  it "enqueues and dequeues concurrently lock" $ \testDB -> do
+    let withPool' = withPool testDB
+        elementCount = 1000 :: Int
+        expected = [0 .. elementCount - 1]
+
+    ref <- newIORef []
+
+    loopThreads <- replicateM 10 $ async $ fix $ \next -> do
+      Payload {..} <- withPool' lock
+      lastCount <- atomicModifyIORef ref
+                 $ \xs -> (pValue : xs, length xs + 1)
+      withPool' $ flip dequeue pId
+      when (lastCount < elementCount) next
+
+    -- Fork a hundred threads and enqueue an index
+    forM_ [0 .. elementCount - 1] $ \i -> forkIO $ void $ withPool' $
+      flip enqueue $ toJSON i
+
+    waitAnyCancel loopThreads
+    Just decoded <- mapM (decode . encode) <$> readIORef ref
+    sort decoded `shouldBe` sort expected
diff --git a/test/Spec.hs b/test/Spec.hs
new file mode 100644
--- /dev/null
+++ b/test/Spec.hs
@@ -0,0 +1,1 @@
+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}