diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,11 +1,16 @@
-## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.6...main)
+## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.2.0...main)
 
 None
 
+## [v1.0.2.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.6...v1.0.2.0)
+
+- Partial `BATCH` support (Enterprise only)
+- Support for `custom` field in Job payloads
+- Lower-level `buildJob` and `commandByteString` functions
+
 ## [v1.0.1.6](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.5...v1.0.1.6)
 
 - Relax dependencies upper bounds
-
 
 ## [v1.0.1.5](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.4...v1.0.1.5)
 
diff --git a/README.lhs b/README.lhs
--- a/README.lhs
+++ b/README.lhs
@@ -169,8 +169,9 @@
 stack build --pedantic --test
 ```
 
-*NOTE*: `FactorySpec` requires a local Faktory server is running, and it will
-flush all Jobs from this server as part of running the tests.
+- `FactorySpec` requires a local Faktory server is running, and it will flush
+  all Jobs from this server as part of running the tests.
+- The tests for `BATCH` require testing against an Enterprise Faktory image
 
 ---
 
diff --git a/faktory.cabal b/faktory.cabal
--- a/faktory.cabal
+++ b/faktory.cabal
@@ -1,6 +1,6 @@
 cabal-version:   1.18
 name:            faktory
-version:         1.0.1.6
+version:         1.0.2.0
 license:         MIT
 license-file:    LICENSE
 copyright:       2018 Freckle Education
@@ -53,7 +53,10 @@
     exposed-modules:
         Faktory.Client
         Faktory.Connection
+        Faktory.Ent.Batch
         Faktory.Job
+        Faktory.Job.Custom
+        Faktory.JobOptions
         Faktory.Prelude
         Faktory.Producer
         Faktory.Protocol
@@ -81,19 +84,22 @@
     build-depends:
         aeson >=1.5.5.1 && <1.6,
         aeson-casing >=0.2.0.0 && <0.3,
-        base >=4.11 && <5,
+        base >=4.13 && <5,
         bytestring >=0.10.12.0 && <0.12,
         connection >=0.3.1 && <0.4,
         cryptonite >=0.27 && <0.29,
         megaparsec >=9.0.1 && <9.1,
         memory >=0.15.0 && <0.16,
+        mtl >=2.2.2 && <2.3,
         network >=3.1.1.1 && <3.2,
         random >=1.1 && <1.3,
         safe-exceptions >=0.1.7.1 && <0.2,
         scanner >=0.3.1 && <0.4,
+        semigroups >=0.19.1 && <0.20,
         text >1.2 && <1.4,
         time >=1.9.3 && <1.12,
-        unix >=2.7.2.2 && <2.8
+        unix >=2.7.2.2 && <2.8,
+        unordered-containers >=0.2.13.0 && <0.3
 
     if impl(ghc >=8.10)
         ghc-options:
@@ -121,7 +127,7 @@
 
     build-depends:
         aeson >=1.5.5.1 && <1.6,
-        base >=4.11 && <5,
+        base >=4.13 && <5,
         faktory -any,
         safe-exceptions >=0.1.7.1 && <0.2
 
@@ -151,7 +157,7 @@
 
     build-depends:
         aeson >=1.5.5.1 && <1.6,
-        base >=4.11 && <5,
+        base >=4.13 && <5,
         faktory -any,
         safe-exceptions >=0.1.7.1 && <0.2
 
@@ -165,6 +171,7 @@
     hs-source-dirs:     tests
     other-modules:
         Faktory.ConnectionSpec
+        Faktory.JobOptionsSpec
         FaktorySpec
         Paths_faktory
 
@@ -185,9 +192,12 @@
         -Wno-monomorphism-restriction -Wno-missing-import-lists -rtsopts
 
     build-depends:
-        base >=4.11 && <5,
+        aeson >=1.5.5.1 && <1.6,
+        async >=2.2.2 && <2.3,
+        base >=4.13 && <5,
         faktory -any,
-        hspec >=2.7.8 && <2.8
+        hspec >=2.7.8 && <2.8,
+        time >=1.9.3 && <1.12
 
     if impl(ghc >=8.10)
         ghc-options:
@@ -217,7 +227,7 @@
 
     build-depends:
         aeson >=1.5.5.1 && <1.6,
-        base >=4.11 && <5,
+        base >=4.13 && <5,
         faktory -any,
         markdown-unlit >=0.5.1 && <0.6
 
diff --git a/library/Faktory/Client.hs b/library/Faktory/Client.hs
--- a/library/Faktory/Client.hs
+++ b/library/Faktory/Client.hs
@@ -9,6 +9,7 @@
   , command_
   , commandOK
   , commandJSON
+  , commandByteString
   ) where
 
 import Faktory.Prelude
@@ -120,15 +121,14 @@
 
 -- | Send a command, read and discard the response
 command_ :: Client -> ByteString -> [ByteString] -> IO ()
-command_ Client {..} cmd args = withMVar clientConnection $ \conn -> do
-  sendUnsafe clientSettings conn cmd args
-  void $ fromRightThrows =<< recvUnsafe clientSettings conn
+command_ client cmd args = do
+  response <- commandByteString client cmd args
+  void $ fromRightThrows response
 
 -- | Send a command, assert the response is @OK@
 commandOK :: HasCallStack => Client -> ByteString -> [ByteString] -> IO ()
-commandOK Client {..} cmd args = withMVar clientConnection $ \conn -> do
-  sendUnsafe clientSettings conn cmd args
-  response <- recvUnsafe clientSettings conn
+commandOK client cmd args = do
+  response <- commandByteString client cmd args
   unless (response == Right (Just "OK"))
     $ throwString
     $ "Server not OK. Reply was: "
@@ -141,10 +141,19 @@
   -> ByteString
   -> [ByteString]
   -> IO (Either String (Maybe a))
-commandJSON Client {..} cmd args = withMVar clientConnection $ \conn -> do
-  sendUnsafe clientSettings conn cmd args
-  emByteString <- recvUnsafe clientSettings conn
+commandJSON client cmd args = do
+  emByteString <- commandByteString client cmd args
   either (pure . Left) (pure . traverse eitherDecode) emByteString
+
+commandByteString
+  :: Client
+  -> ByteString
+  -> [ByteString]
+  -> IO (Either String (Maybe ByteString))
+commandByteString Client {..} cmd args = withMVar clientConnection $ \conn ->
+  do
+    sendUnsafe clientSettings conn cmd args
+    recvUnsafe clientSettings conn
 
 -- | Send a command to the Server socket
 --
diff --git a/library/Faktory/Ent/Batch.hs b/library/Faktory/Ent/Batch.hs
new file mode 100644
--- /dev/null
+++ b/library/Faktory/Ent/Batch.hs
@@ -0,0 +1,118 @@
+{-# LANGUAGE DerivingVia #-}
+
+-- | Support for the @BATCH@ command (Enterprise only)
+--
+-- <https://github.com/contribsys/faktory/wiki/Ent-Batches>
+--
+-- Batches allow multiple Jobs to be enqueued as a group, with a description
+-- (visible in the admin UI) and Jobs attached to run on completion of all Jobs
+-- within the group (always, or only if all were successful).
+--
+-- Usage:
+--
+-- @
+-- -- Build a Job to run at completion of the Batch. Arguments are the same as
+-- -- you would pass to 'perform' the Job.
+-- onComplete <- buildJob mempty producer myJob
+--
+-- 'runBatch' ('complete' onComplete <> 'description' "My Batch") producer $ do
+--   -- Use 'batchPerform' instead of 'perform'
+--   void $ 'batchPerform' mempty producer myBatchedJob1
+--   void $ 'batchPerform' mempty producer myBatchedJob2
+-- @
+--
+-- __/NOTE__: This module does not support batched Jobs dynamically adding more
+-- Jobs to the Batch. PRs welcome.
+--
+module Faktory.Ent.Batch
+    (
+    -- * Options
+      BatchOptions
+    , description
+    , complete
+    , success
+
+    -- * Running
+    , runBatch
+    , Batch
+    , batchPerform
+    )
+where
+
+import Faktory.Prelude
+
+import Control.Monad.Reader
+import Data.Aeson
+import Data.Aeson.Casing
+import Data.ByteString.Lazy as BSL
+import Data.Semigroup (Last(..))
+import Data.Semigroup.Generic
+import Data.Text.Encoding (decodeUtf8, encodeUtf8)
+import Faktory.Client
+import Faktory.Job
+import Faktory.Producer
+import GHC.Generics
+import GHC.Stack
+
+newtype Batch a = Batch (ReaderT BatchId IO a)
+  deriving newtype (Functor, Applicative, Monad, MonadIO, MonadReader BatchId)
+
+newtype BatchId = BatchId Text
+  deriving newtype ToJSON
+
+data BatchOptions arg = BatchOptions
+  { boDescription :: Maybe (Last Text)
+  , boSuccess :: Maybe (Last (Job arg))
+  , boComplete :: Maybe (Last (Job arg))
+  }
+  deriving stock Generic
+  deriving (Semigroup, Monoid) via GenericSemigroupMonoid (BatchOptions arg)
+
+instance ToJSON arg => ToJSON (BatchOptions arg) where
+  toJSON = genericToJSON $ aesonPrefix snakeCase
+  toEncoding = genericToEncoding $ aesonPrefix snakeCase
+
+description :: Text -> BatchOptions arg
+description d = mempty { boDescription = Just $ Last d }
+
+complete :: Job arg -> BatchOptions arg
+complete job = mempty { boComplete = Just $ Last job }
+
+success :: Job arg -> BatchOptions arg
+success job = mempty { boSuccess = Just $ Last job }
+
+runBatch :: ToJSON arg => BatchOptions arg -> Producer -> Batch a -> IO a
+runBatch options producer (Batch f) = do
+  bid <- newBatch producer options
+  result <- runReaderT f bid
+  result <$ commitBatch producer bid
+
+newtype CustomBatchId = CustomBatchId
+  { bid :: BatchId
+  }
+  deriving stock Generic
+  deriving anyclass ToJSON
+
+batchPerform
+  :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> Batch JobId
+batchPerform options producer arg = do
+  bid <- ask
+  Batch $ lift $ perform (options <> custom (CustomBatchId bid)) producer arg
+
+newBatch :: ToJSON arg => Producer -> BatchOptions arg -> IO BatchId
+newBatch producer options = do
+  result <- commandByteString
+    (producerClient producer)
+    "BATCH NEW"
+    [encode options]
+  case result of
+    Left err -> batchNewError err
+    Right Nothing -> batchNewError "No BatchId returned"
+    Right (Just bs) -> pure $ BatchId $ decodeUtf8 $ BSL.toStrict bs
+  where batchNewError err = throwString $ "BATCH NEW error: " <> err
+
+commitBatch :: Producer -> BatchId -> IO ()
+commitBatch producer (BatchId bid) = command_
+  (producerClient producer)
+  "BATCH COMMIT"
+  [BSL.fromStrict $ encodeUtf8 bid]
diff --git a/library/Faktory/Job.hs b/library/Faktory/Job.hs
--- a/library/Faktory/Job.hs
+++ b/library/Faktory/Job.hs
@@ -9,6 +9,8 @@
   , jobtype
   , at
   , in_
+  , custom
+  , buildJob
   , newJob
   , jobJid
   , jobArg
@@ -17,44 +19,27 @@
 import Faktory.Prelude
 
 import Data.Aeson
-import Data.Aeson.Casing
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as NE
-import Data.Time
+import Data.Time (UTCTime)
 import Faktory.Client (Client(..))
+import Faktory.Connection (ConnectionInfo(..))
+import Faktory.JobOptions
 import Faktory.Producer (Producer(..), pushJob)
-import Faktory.Settings
-import GHC.Generics
+import Faktory.Settings (Namespace, Settings(..))
 import GHC.Stack
 import System.Random
 
 data Job arg = Job
   { jobJid :: JobId
-  , jobJobtype :: String
+  , jobAt :: Maybe UTCTime
+  -- ^ Will be set based on 'JobOptions' when enqueued
   , jobArgs :: NonEmpty arg
   -- ^ Faktory needs to serialize args as a list, but we like a single-argument
   -- interface so that's what we expose. See @'jobArg'@.
-  , jobRetry :: Maybe Int
-  , jobQueue :: Maybe Queue
-  , jobAt :: Maybe UTCTime
+  , jobOptions :: JobOptions
   }
-  deriving stock Generic
 
--- | Individual changes to a @'Job'@ to be 'perform'ed
-data JobUpdate
-  = SetRetry Int
-  | SetQueue Queue
-  | SetJobtype String
-  | SetAt UTCTime
-  | SetIn NominalDiffTime
-
--- | Options for the execution of a job
---
--- See @'perform'@ for more details.
---
-newtype JobOptions = JobOptions [JobUpdate]
-  deriving newtype (Semigroup, Monoid)
-
 -- | Perform a Job with the given options
 --
 -- @
@@ -69,48 +54,26 @@
 perform
   :: (HasCallStack, ToJSON arg) => JobOptions -> Producer -> arg -> IO JobId
 perform options producer arg = do
-  let
-    namespace =
-      connectionInfoNamespace
-        $ settingsConnection
-        $ clientSettings
-        $ producerClient producer
-  job <- applyOptions namespace options =<< newJob arg
+  job <- buildJob options producer arg
   jobJid job <$ pushJob producer job
 
 applyOptions :: Namespace -> JobOptions -> Job arg -> IO (Job arg)
-applyOptions namespace (JobOptions patches) = go patches
- where
-  go [] job = pure job
-  go (set : sets) job = case set of
-    SetRetry n -> go sets $ job { jobRetry = Just n }
-    SetQueue q ->
-      go sets $ job { jobQueue = Just $ namespaceQueue namespace q }
-    SetJobtype jt -> go sets $ job { jobJobtype = jt }
-    SetAt time -> go sets $ job { jobAt = Just time }
-    SetIn diff -> do
-      now <- getCurrentTime
-      go sets $ job { jobAt = Just $ addUTCTime diff now }
-
-retry :: Int -> JobOptions
-retry n = JobOptions [SetRetry n]
-
--- | Equivalent to @'retry' (-1)@: no retries, and move to Dead on failure
-once :: JobOptions
-once = retry (-1)
-
-queue :: Queue -> JobOptions
-queue q = JobOptions [SetQueue q]
-
-jobtype :: String -> JobOptions
-jobtype jt = JobOptions [SetJobtype jt]
-
-at :: UTCTime -> JobOptions
-at t = JobOptions [SetAt t]
+applyOptions namespace options job = do
+  scheduledAt <- getAtFromSchedule options
+  let namespacedOptions = namespaceQueue namespace $ jobOptions job <> options
+  pure $ job { jobAt = scheduledAt, jobOptions = namespacedOptions }
 
-in_ :: NominalDiffTime -> JobOptions
-in_ i = JobOptions [SetIn i]
+-- | Construct a 'Job' and apply options and Producer settings
+buildJob :: JobOptions -> Producer -> arg -> IO (Job arg)
+buildJob options producer arg = applyOptions namespace options =<< newJob arg
+ where
+  namespace =
+    connectionInfoNamespace
+      $ settingsConnection
+      $ clientSettings
+      $ producerClient producer
 
+-- | Construct a 'Job' with default 'JobOptions'
 newJob :: arg -> IO (Job arg)
 newJob arg = do
   -- Ruby uses 12 random hex
@@ -118,21 +81,36 @@
 
   pure Job
     { jobJid = jobId
-    , jobJobtype = "Default"
-    , jobArgs = pure arg
-    , jobRetry = Nothing
-    , jobQueue = Nothing
     , jobAt = Nothing
+    , jobArgs = pure arg
+    , jobOptions = jobtype "Default"
     }
 
 jobArg :: Job arg -> arg
 jobArg Job {..} = NE.head jobArgs
 
 instance ToJSON args => ToJSON (Job args) where
-  toJSON = genericToJSON $ aesonPrefix snakeCase
-  toEncoding = genericToEncoding $ aesonPrefix snakeCase
+  toJSON = object . toPairs
+  toEncoding = pairs . mconcat . toPairs
 
+toPairs :: (KeyValue a, ToJSON arg) => Job arg -> [a]
+toPairs Job {..} =
+  [ "jid" .= jobJid
+  , "at" .= jobAt
+  , "args" .= jobArgs
+  , "jobtype" .= joJobtype jobOptions
+  , "retry" .= joRetry jobOptions
+  , "queue" .= joQueue jobOptions
+  , "custom" .= joCustom jobOptions
+  ]
+
+-- brittany-disable-next-binding
+
 instance FromJSON args => FromJSON (Job args) where
-  parseJSON = genericParseJSON $ aesonPrefix snakeCase
+  parseJSON = withObject "Job" $ \o -> Job
+    <$> o .: "jid"
+    <*> o .:? "at"
+    <*> o .: "args"
+    <*> parseJSON (Object o)
 
 type JobId = String
diff --git a/library/Faktory/Job/Custom.hs b/library/Faktory/Job/Custom.hs
new file mode 100644
--- /dev/null
+++ b/library/Faktory/Job/Custom.hs
@@ -0,0 +1,26 @@
+-- | Wrapper for the schema-less 'Value' used for @custom@ in Job payloads
+--
+-- This type's 'Semigroup' will merge two Objects. It is right-biased, as we are
+-- generally throughout this library, so called /last-wins/ semantics.
+--
+module Faktory.Job.Custom
+    ( Custom
+    , toCustom
+    ) where
+
+import Faktory.Prelude
+
+import Data.Aeson
+import qualified Data.HashMap.Strict as HashMap
+
+newtype Custom = Custom Value
+  deriving stock (Eq, Show)
+  deriving newtype (FromJSON, ToJSON)
+
+toCustom :: ToJSON a => a -> Custom
+toCustom = Custom . toJSON
+
+instance Semigroup Custom where
+  (Custom (Object a)) <> (Custom (Object b)) =
+    Custom $ Object $ HashMap.union b a
+  _ <> b = b
diff --git a/library/Faktory/JobOptions.hs b/library/Faktory/JobOptions.hs
new file mode 100644
--- /dev/null
+++ b/library/Faktory/JobOptions.hs
@@ -0,0 +1,96 @@
+{-# LANGUAGE DerivingVia #-}
+
+module Faktory.JobOptions
+  ( JobOptions(..)
+
+  -- * Modifiers
+  , retry
+  , once
+  , queue
+  , jobtype
+  , at
+  , in_
+  , custom
+
+  -- * Enqueue-time modifiers
+  , getAtFromSchedule
+  , namespaceQueue
+  ) where
+
+import Faktory.Prelude
+
+import Data.Aeson
+import Data.Semigroup (Last(..))
+import Data.Semigroup.Generic
+import Data.Time
+import Faktory.Job.Custom
+import Faktory.Settings (Namespace, Queue)
+import qualified Faktory.Settings as Settings
+import GHC.Generics
+
+-- | Options for the execution of a job
+--
+-- These can be constructed using '(<>)':
+--
+-- @
+-- let options = 'retry' 1 <> 'jobtype' "MyJob"
+-- @
+--
+-- To enqueue with defaults, use 'mempty'.
+--
+-- Options use 'Last' semantics, so (e.g.) @'retry' x <>@ will set retries to
+-- @x@ only if not already set, and @<> 'retry' x@ will override any
+-- already-present retries to @x@.
+--
+data JobOptions = JobOptions
+  { joJobtype :: Maybe (Last String)
+  , joRetry :: Maybe (Last Int)
+  , joQueue :: Maybe (Last Queue)
+  , joSchedule :: Maybe (Last (Either UTCTime NominalDiffTime))
+  , joCustom :: Maybe Custom
+  }
+  deriving stock Generic
+  deriving (Semigroup, Monoid) via GenericSemigroupMonoid JobOptions
+
+-- brittany-disable-next-binding
+
+instance FromJSON JobOptions where
+  parseJSON = withObject "JobOptions" $ \o ->
+    JobOptions
+      <$> o .:? "jobtype"
+      <*> o .:? "retry"
+      <*> o .:? "queue"
+      <*> (fmap (Last . Right) <$> o .:? "at")
+      <*> o .:? "custom"
+
+getAtFromSchedule :: JobOptions -> IO (Maybe UTCTime)
+getAtFromSchedule options = for (getLast <$> joSchedule options) $ \case
+  Left t -> pure t
+  Right nd -> addUTCTime nd <$> getCurrentTime
+
+namespaceQueue :: Namespace -> JobOptions -> JobOptions
+namespaceQueue namespace options = case joQueue options of
+  Nothing -> options
+  Just (Last q) -> options <> queue (Settings.namespaceQueue namespace q)
+
+retry :: Int -> JobOptions
+retry n = mempty { joRetry = Just $ Last n }
+
+-- | Equivalent to @'retry' (-1)@: no retries, and move to Dead on failure
+once :: JobOptions
+once = retry (-1)
+
+queue :: Queue -> JobOptions
+queue q = mempty { joQueue = Just $ Last q }
+
+jobtype :: String -> JobOptions
+jobtype jt = mempty { joJobtype = Just $ Last jt }
+
+at :: UTCTime -> JobOptions
+at t = mempty { joSchedule = Just $ Last $ Left t }
+
+in_ :: NominalDiffTime -> JobOptions
+in_ i = mempty { joSchedule = Just $ Last $ Right i }
+
+custom :: ToJSON a => a -> JobOptions
+custom v = mempty { joCustom = Just $ toCustom v }
diff --git a/library/Faktory/Settings.hs b/library/Faktory/Settings.hs
--- a/library/Faktory/Settings.hs
+++ b/library/Faktory/Settings.hs
@@ -73,6 +73,7 @@
     }
 
 newtype Queue = Queue Text
+  deriving stock (Eq, Show)
   deriving newtype (IsString, FromJSON, ToJSON)
 
 namespaceQueue :: Namespace -> Queue -> Queue
diff --git a/library/Faktory/Worker.hs b/library/Faktory/Worker.hs
--- a/library/Faktory/Worker.hs
+++ b/library/Faktory/Worker.hs
@@ -58,7 +58,11 @@
   toEncoding = genericToEncoding $ aesonPrefix snakeCase
 
 runWorker
-  :: FromJSON args => Settings -> WorkerSettings -> (args -> IO ()) -> IO ()
+  :: (HasCallStack, FromJSON args)
+  => Settings
+  -> WorkerSettings
+  -> (args -> IO ())
+  -> IO ()
 runWorker settings workerSettings f = do
   workerId <- maybe randomWorkerId pure $ settingsId workerSettings
   client <- newClient settings $ Just workerId
@@ -75,7 +79,7 @@
   runWorker settings workerSettings f
 
 processorLoop
-  :: FromJSON arg
+  :: (HasCallStack, FromJSON arg)
   => Client
   -> Settings
   -> WorkerSettings
diff --git a/tests/Faktory/JobOptionsSpec.hs b/tests/Faktory/JobOptionsSpec.hs
new file mode 100644
--- /dev/null
+++ b/tests/Faktory/JobOptionsSpec.hs
@@ -0,0 +1,100 @@
+module Faktory.JobOptionsSpec
+  ( spec
+  ) where
+
+import Faktory.Prelude
+
+import Data.Aeson
+import Data.Semigroup (Last(..))
+import Data.Time
+  ( DiffTime
+  , UTCTime(..)
+  , addUTCTime
+  , diffTimeToPicoseconds
+  , getCurrentTime
+  , secondsToDiffTime
+  )
+import Data.Time.Calendar (fromGregorian)
+import Faktory.Job.Custom
+import Faktory.JobOptions
+import Faktory.Settings (Namespace(..), Queue(..))
+import Test.Hspec
+
+spec :: Spec
+spec = do
+  describe "JobOptions" $ do
+    describe "(<>)" $ do
+      it "uses Last semantics" $ do
+        let options = retry 1 <> retry 2
+
+        fmap getLast (joRetry options) `shouldBe` Just 2
+
+      it "merges custom fields" $ do
+        let
+          custom1 = object ["a" .= True, "b" .= True, "c" .= True]
+          custom2 = object ["a" .= False, "d" .= False]
+          options = custom custom1 <> custom custom2
+
+        joCustom options
+          `shouldBe` Just
+                       (toCustom
+                       $ object
+                           [ "a" .= False
+                           , "b" .= True
+                           , "c" .= True
+                           , "d" .= False
+                           ]
+                       )
+
+    describe "getAtFromSchedule" $ do
+      it "sets at based on in" $ do
+        now <- getCurrentTime
+        let time = addUTCTime 30 now
+
+        timeAt <- getAtFromSchedule $ in_ 30
+
+        fmap truncateUTCTime timeAt `shouldBe` Just (truncateUTCTime time)
+
+      it "respects the last option given" $ do
+        now <- getCurrentTime
+        let time = makeUTCTime 2020 1 1 0
+
+        timeAt1 <- getAtFromSchedule $ in_ 5 <> at time
+        timeAt2 <- getAtFromSchedule $ at time <> in_ 5
+
+        timeAt1 `shouldBe` Just time
+        fmap truncateUTCTime timeAt2
+          `shouldBe` Just (truncateUTCTime $ addUTCTime 5 now)
+
+    describe "namespaceQueue" $ do
+      it "leaves no queue alone" $ do
+        let options = namespaceQueue (Namespace "namespace.") mempty
+
+        fmap getLast (joQueue options) `shouldBe` Nothing
+
+      it "namespaces a queue when given" $ do
+        let options = namespaceQueue (Namespace "namespace.") $ queue "queue"
+
+        fmap getLast (joQueue options) `shouldBe` Just (Queue "namespace.queue")
+
+makeUTCTime :: Integer -> Int -> Int -> DiffTime -> UTCTime
+makeUTCTime y m d s =
+  UTCTime { utctDay = fromGregorian y m d, utctDayTime = s }
+
+truncateUTCTime :: UTCTime -> UTCTime
+truncateUTCTime t = t
+  { utctDayTime =
+    secondsToDiffTime
+    $ picosecondsToSeconds
+    $ diffTimeToPicoseconds
+    $ utctDayTime t
+  }
+
+-- sameSecondAs :: UTCTime -> UTCTime -> Bool
+-- sameSecondAs a b =
+--   (utctDay a == utctDay b)
+--     &&
+--     == picosecondsToSeconds (diffTimeToPicoseconds $ utctDayTime b)
+
+picosecondsToSeconds :: Integer -> Integer
+picosecondsToSeconds = round @Double . (/ 1000000000000) . fromIntegral
diff --git a/tests/FaktorySpec.hs b/tests/FaktorySpec.hs
--- a/tests/FaktorySpec.hs
+++ b/tests/FaktorySpec.hs
@@ -4,8 +4,11 @@
 
 import Faktory.Prelude
 
-import Control.Concurrent (forkIO, threadDelay)
+import Control.Concurrent (threadDelay)
+import Control.Concurrent.Async
 import Control.Concurrent.MVar
+import Control.Monad.IO.Class (liftIO)
+import Faktory.Ent.Batch
 import Faktory.Job
 import Faktory.Producer
 import Faktory.Settings
@@ -15,42 +18,21 @@
 spec :: Spec
 spec = describe "Faktory" $ do
   it "can push and process jobs" $ do
-    bracket newProducerEnv closeProducer $ \producer -> do
-      void $ flush producer
+    jobs <- workerTestCase $ \producer -> do
       void $ perform @Text mempty producer "a"
       void $ perform @Text mempty producer "b"
-      void $ perform @Text mempty producer "HALT"
 
-    processedJobs <- newMVar ([] :: [Text])
-    runWorkerEnv $ \job -> do
-      modifyMVar_ processedJobs $ pure . (job :)
-      when (job == "HALT") $ throw WorkerHalt
-
-    jobs <- readMVar processedJobs
     jobs `shouldMatchList` ["a", "b", "HALT"]
 
   it "can push jobs with optional attributes" $ do
-    bracket newProducerEnv closeProducer $ \producer -> do
-      void $ flush producer
+    jobs <- workerTestCase $ \producer -> do
       void $ perform @Text once producer "a"
       void $ perform @Text (retry 0) producer "b"
-      void $ perform @Text mempty producer "HALT"
 
-    processedJobs <- newMVar ([] :: [Text])
-    runWorkerEnv $ \job -> do
-      modifyMVar_ processedJobs $ pure . (job :)
-      when (job == "HALT") $ throw WorkerHalt
-
-    jobs <- readMVar processedJobs
     jobs `shouldMatchList` ["a", "b", "HALT"]
 
   it "correctly handles fetch timeouts" $ do
-    settings <- envSettings
-    workerSettings' <- envWorkerSettings
-    let workerSettings = workerSettings' { settingsIdleDelay = 0 }
-
-    -- start a background thread that waits for longer than the fetch timeout,
-    -- then stops the worker.
+    -- Pause longer than the fetch timeout
     --
     -- https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#fetching-jobs
     --
@@ -58,16 +40,102 @@
     -- the Server and handles it correctly. Setting our own idle delay to 0
     -- ensures that we'll pick up the following HALT message immediately.
     --
-    void $ forkIO $ bracket (newProducer settings) closeProducer $ \producer ->
-      do
-        void $ flush producer
-        threadDelay $ 2 * 1000000 + 250000
-        void $ perform @Text mempty producer "HALT"
-
-    processedJobs <- newMVar ([] :: [Text])
-    runWorker settings workerSettings $ \job -> do
-      modifyMVar_ processedJobs $ pure . (job :)
-      when (job == "HALT") $ throw WorkerHalt
+    let editSettings ws = ws { settingsIdleDelay = 0 }
+    jobs <- workerTestCaseWith editSettings $ \_ -> do
+      threadDelay $ 2 * 1000000 + 250000
 
-    jobs <- readMVar processedJobs
     jobs `shouldMatchList` ["HALT"]
+
+  context "BATCH" $ do
+    it "runs a success job if all in-batch jobs succeed" $ do
+      jobs <- workerTestCase $ \producer -> do
+        c <- buildJob @Text mempty producer "c"
+        void $ runBatch (success c) producer $ do
+          void $ batchPerform @Text mempty producer "a"
+          void $ batchPerform @Text mempty producer "b"
+        -- Give a little time for Faktory to fire the callback
+        liftIO $ threadDelay 500000
+
+      jobs `shouldMatchList` ["a", "b", "c", "HALT"]
+
+    it "does not run a success job if all jobs don't succeed" $ do
+      jobs <- workerTestCase $ \producer -> do
+        c <- buildJob @Text mempty producer "c"
+        void $ runBatch (success c) producer $ do
+          void $ batchPerform @Text mempty producer "BOOM"
+          void $ batchPerform @Text mempty producer "b"
+        liftIO $ threadDelay 500000
+
+      jobs `shouldMatchList` ["BOOM", "b", "HALT"]
+
+    it "runs a job on complete" $ do
+      jobs <- workerTestCase $ \producer -> do
+        c <- buildJob @Text mempty producer "c"
+        void $ runBatch (complete c) producer $ do
+          void $ batchPerform @Text mempty producer "a"
+          void $ batchPerform @Text mempty producer "b"
+        liftIO $ threadDelay 500000
+
+      jobs `shouldMatchList` ["a", "b", "c", "HALT"]
+
+    it "runs a job on complete, even if in-batch jobs fail" $ do
+      jobs <- workerTestCase $ \producer -> do
+        c <- buildJob @Text mempty producer "c"
+        void $ runBatch (complete c) producer $ do
+          void $ batchPerform @Text mempty producer "BOOM"
+          void $ batchPerform @Text mempty producer "b"
+        liftIO $ threadDelay 500000
+
+      jobs `shouldMatchList` ["BOOM", "b", "c", "HALT"]
+
+    it "combines duplicate options in last-wins fashion" $ do
+      jobs <- workerTestCase $ \producer -> do
+        c <- buildJob @Text mempty producer "c"
+        d <- buildJob @Text mempty producer "d"
+        let options = description "foo" <> success c <> success d
+        void $ runBatch options producer $ do
+          void $ batchPerform @Text mempty producer "a"
+          void $ batchPerform @Text mempty producer "b"
+        liftIO $ threadDelay 500000
+
+      jobs `shouldMatchList` ["a", "b", "d", "HALT"]
+
+    it "runs success and complete if all Jobs were successful" $ do
+      jobs <- workerTestCase $ \producer -> do
+        c <- buildJob @Text mempty producer "c"
+        d <- buildJob @Text mempty producer "d"
+        let options = description "foo" <> complete c <> success d
+        void $ runBatch options producer $ do
+          void $ batchPerform @Text mempty producer "a"
+          void $ batchPerform @Text mempty producer "b"
+        liftIO $ threadDelay 500000
+
+      jobs `shouldMatchList` ["a", "b", "c", "d", "HALT"]
+
+workerTestCase :: HasCallStack => (Producer -> IO ()) -> IO [Text]
+workerTestCase = workerTestCaseWith id
+
+workerTestCaseWith
+  :: HasCallStack
+  => (WorkerSettings -> WorkerSettings)
+  -> (Producer -> IO ())
+  -> IO [Text]
+workerTestCaseWith editSettings run = do
+  bracket newProducerEnv closeProducer $ \producer -> do
+    void $ flush producer
+
+  settings <- envSettings
+  workerSettings <- editSettings <$> envWorkerSettings
+
+  processedJobs <- newMVar []
+  a <- async $ runWorker settings workerSettings $ \job -> do
+    modifyMVar_ processedJobs $ pure . (job :)
+    when (job == "BOOM") $ throw $ userError "BOOM"
+    when (job == "HALT") $ throw WorkerHalt
+
+  bracket newProducerEnv closeProducer $ \producer -> do
+    run producer
+    void $ perform @Text mempty producer "HALT"
+
+  void $ wait a
+  readMVar processedJobs