diff --git a/ChangeLog.md b/ChangeLog.md
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -1,3 +1,8 @@
+# v0.5
+
+* Renamed internal modules under `Internal` prefix
+* Renamed `Database.Franz.Reconnect` to `Database.Franz.Client.Reconnect`
+
 # v0.4
 
 * Added `StreamName`
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -61,6 +61,8 @@
 
 ## Client API
 
+`Database.Franz.Client` exposes the client API.
+
 You can obtain a `Connection` to a remote franz file with `withConnection`.
 It tries to mount a squashfs image at `path`. This is shared between connections, and unmounts when the last client closes the connection.
 
@@ -108,3 +110,30 @@
 
 toList :: Contents -> [Item]
 ```
+
+## Writer API
+
+`Database.Franz.Writer` provides the writer interface.
+
+```haskell
+withWriter :: Foldable f
+  => f String
+  -> FilePath
+  -> (WriterHandle f -> IO a)
+  -> IO a
+```
+
+`withWriter` acquires a handle. The `f String` parameter represents a list of index names.
+
+```haskell
+write :: Foldable f
+  => WriterHandle f
+  -> f Int64 -- ^ index values
+  -> Builder -- ^ payload
+  -> IO Int
+flush :: WriterHandle f -> IO ()
+```
+
+`write` appends a payload to the stream. `f Int64` is the list of index values, and it has to have the same length as the one you specified in `withWriter`. Changes will be written to disk whenever the buffer gets full or you call `flush`.
+
+If you don't need the index mechanism, you can use `Database.Franz.Writer.Simple` instead.
diff --git a/app/client.hs b/app/client.hs
--- a/app/client.hs
+++ b/app/client.hs
@@ -1,6 +1,6 @@
 {-# LANGUAGE ApplicativeDo #-}
 module Main where
-import Database.Franz.URI
+import Database.Franz.Internal.URI
 import Database.Franz.Client
 import qualified Database.Franz.Contents as C
 
diff --git a/app/server.hs b/app/server.hs
--- a/app/server.hs
+++ b/app/server.hs
@@ -1,6 +1,5 @@
 module Main where
 
-import Database.Franz.Reader (FranzPrefix(..))
 import Database.Franz.Server
 import Options.Applicative
 
diff --git a/franz.cabal b/franz.cabal
--- a/franz.cabal
+++ b/franz.cabal
@@ -1,5 +1,5 @@
 cabal-version:  2.0
-version:        0.4
+version:        0.5
 name:           franz
 description:    Please see the README on GitHub at <https://github.com/tsurucapital/franz#readme>
 homepage:       https://github.com/tsurucapital/franz#readme
@@ -22,20 +22,22 @@
 
 library
   exposed-modules:
-      Database.Franz.Writer
-      Database.Franz.Writer.Simple
-      Database.Franz.Internal
+      Database.Franz.Client
+      Database.Franz.Client.Reconnect
       Database.Franz.Contents
-      Database.Franz.Reader
+      Database.Franz.Internal.Contents
+      Database.Franz.Internal.Fuse
+      Database.Franz.Internal.IO
+      Database.Franz.Internal.Protocol
+      Database.Franz.Internal.Reader
+      Database.Franz.Internal.URI
       Database.Franz.Server
-      Database.Franz.Client
-      Database.Franz.Protocol
-      Database.Franz.Reconnect
-      Database.Franz.URI
+      Database.Franz.Writer
+      Database.Franz.Writer.Simple
   hs-source-dirs:
       src
   build-depends:
-      base >=4.7 && <5
+      base >=4.13 && <5
     , bytestring
     , cereal
     , containers
diff --git a/src/Database/Franz/Client.hs b/src/Database/Franz/Client.hs
--- a/src/Database/Franz/Client.hs
+++ b/src/Database/Franz/Client.hs
@@ -37,12 +37,12 @@
 import Data.IORef.Unboxed
 import qualified Data.IntMap.Strict as IM
 import Data.Serialize hiding (getInt64le)
-import Database.Franz.Contents
-import Database.Franz.Internal
-import Database.Franz.Protocol
-import Database.Franz.Reader
-import Database.Franz.Server
-import Database.Franz.URI
+import Database.Franz.Internal.Contents
+import Database.Franz.Internal.Fuse
+import Database.Franz.Internal.IO
+import Database.Franz.Internal.Protocol
+import Database.Franz.Internal.Reader
+import Database.Franz.Internal.URI
 import qualified Network.Socket as S
 import qualified Network.Socket.ByteString as SB
 import System.Process (ProcessHandle)
@@ -159,7 +159,7 @@
       let tmpDir' = tmpDir </> "franz"
       createDirectoryIfMissing True tmpDir'
       dir <- createTempDirectory tmpDir' (takeBaseName path)
-      fuse <- mountFuse path dir
+      fuse <- mountFuse mempty (throwIO . InternalError) path dir
       pure (dir, Just fuse)
   pure LocalConnection{..}
 
@@ -169,7 +169,7 @@
   withMVar connSocket S.close
 disconnect LocalConnection{..} =
   closeFranzReader connReader
-  `finally` mapM_ (\p -> killFuse p connDir) connFuse
+  `finally` mapM_ (\p -> killFuse mempty p connDir) connFuse
 
 defQuery :: StreamName -> Query
 defQuery name = Query
diff --git a/src/Database/Franz/Client/Reconnect.hs b/src/Database/Franz/Client/Reconnect.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Client/Reconnect.hs
@@ -0,0 +1,96 @@
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE LambdaCase #-}
+module Database.Franz.Client.Reconnect
+  ( Pool
+  , poolLogFunc
+  , poolRetryPolicy
+  , withPool
+  , withReconnection
+  , Reconnect(..)
+  , atomicallyReconnecting
+  , fetchWithPool
+  )
+  where
+
+import Control.Retry (recovering, RetryPolicyM)
+import Control.Concurrent.MVar
+import Control.Concurrent.STM
+import Control.Exception (IOException)
+import Control.Monad.Catch
+import Database.Franz.Client
+
+data Pool = Pool
+  { poolPath :: FranzPath
+  , poolRef :: MVar (Int {- Connection number -}, Maybe Connection)
+  , poolRetryPolicy :: RetryPolicyM IO
+  , poolLogFunc :: String -> IO ()
+  }
+
+-- | A wrapper of 'fetch' which calls 'withReconnection' internally
+fetchWithPool
+  :: Pool
+  -> Query
+  -> (STM Response -> IO r)
+  -> IO r
+fetchWithPool pool q cont = withReconnection pool $ \conn -> fetch conn q cont
+
+-- | Run an action which takes a 'Connection', reconnecting whenever it throws an exception.
+withReconnection :: Pool -> (Connection -> IO a) -> IO a
+withReconnection Pool{..} cont = recovering
+  poolRetryPolicy
+  [const $ Handler $ \Reconnect -> pure True]
+  body
+  where
+
+    handler ex
+      | Just (ClientError err) <- fromException ex = Just err
+      | Just e <- fromException ex = Just (show (e :: IOException))
+      | Just Reconnect <- fromException ex = Just
+          $ "Connection to " <> fromFranzPath poolPath <> " timed out"
+      | otherwise = Nothing
+
+    body _ = do
+      (i, conn) <- modifyMVar poolRef $ \case
+        (i, Nothing) -> do
+            poolLogFunc $ unwords
+                [ "Connnecting to"
+                , fromFranzPath poolPath
+                ]
+            conn <- tryJust handler (connect poolPath)
+                >>= either (\e -> poolLogFunc e >> throwM Reconnect) pure
+            poolLogFunc $ "Connection #" <> show i <> " established"
+            pure ((i, Just conn), (i, conn))
+        v@(i, Just c) -> pure (v, (i, c))
+
+      tryJust handler (cont conn) >>= \case
+        Right a -> pure a
+        Left msg -> do
+            poolLogFunc msg
+            modifyMVar_ poolRef $ \case
+                -- Don't disconnect if the sequential number is different;
+                -- another thread already established a new connection
+                (j, Just _) | i == j -> (i + 1, Nothing) <$ disconnect conn
+                x -> pure x
+            throwM Reconnect
+
+data Reconnect = Reconnect deriving (Show, Eq)
+instance Exception Reconnect
+
+withPool :: RetryPolicyM IO
+    -> (String -> IO ()) -- ^ diagnostic output
+    -> FranzPath
+    -> (Pool -> IO a)
+    -> IO a
+withPool poolRetryPolicy poolLogFunc poolPath cont = do
+  poolRef <- newMVar (0, Nothing)
+  cont Pool{..} `finally` do
+    (_, conn) <- takeMVar poolRef
+    mapM_ disconnect conn
+
+-- | Run an 'STM' action, throwing 'Reconnect' when it exceeds the given timeout.
+atomicallyReconnecting :: Int -- ^ timeout in microseconds
+    -> STM a -> IO a
+atomicallyReconnecting timeout m = atomicallyWithin timeout m
+  >>= maybe (throwM Reconnect) pure
diff --git a/src/Database/Franz/Contents.hs b/src/Database/Franz/Contents.hs
--- a/src/Database/Franz/Contents.hs
+++ b/src/Database/Franz/Contents.hs
@@ -3,25 +3,20 @@
 {-# LANGUAGE RecordWildCards #-}
 module Database.Franz.Contents
   ( Contents
-  , Database.Franz.Contents.indexNames
+  , Database.Franz.Internal.Contents.indexNames
   , Item(..)
   , toList
   , last
   , length
   , index
   , lookupIndex
-  -- * Internal
-  , getResponse
-  , readContents
   ) where
 
-import Data.Serialize hiding (getInt64le)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.Vector as V
 import qualified Data.Vector.Unboxed as U
-import Database.Franz.Internal
-import Database.Franz.Protocol
-import Database.Franz.Reader
+import Database.Franz.Internal.Protocol
+import Database.Franz.Internal.Contents
 import Data.Int
 import Prelude hiding (length, last)
 
@@ -31,36 +26,9 @@
   , payload :: !B.ByteString
   } deriving (Show, Eq)
 
-data Contents = Contents
-  { indexNames :: !(V.Vector IndexName)
-  , payloads :: !B.ByteString
-  , indicess :: !IndexVec
-  , length :: !Int
-  , payloadOffset :: !Int
-  , seqnoOffset :: !Int
-  }
-
 toList :: Contents -> [Item]
 toList contents = [unsafeIndex contents i | i <- [0..length contents - 1]]
 
--- A vector containing file offsets and extra indices
-type IndexVec = V.Vector (Int, U.Vector Int64)
-
-getIndexVec :: V.Vector IndexName -> Int -> Get IndexVec
-getIndexVec names len = V.replicateM len
-  $ (,) <$> getInt64le <*> U.convert `fmap` traverse (const getInt64le) names
-
-getResponse :: Get Contents
-getResponse = do
-  PayloadHeader seqnoOffset s1 payloadOffset indexNames <- get
-  let length = s1 - seqnoOffset
-  if length <= 0
-    then pure Contents{ payloads = B.empty, indicess = V.empty, ..}
-    else do
-      indicess <- getIndexVec indexNames length
-      payloads <- getByteString $ fst (V.unsafeLast indicess) - payloadOffset
-      pure Contents{..}
-
 last :: Contents -> Maybe Item
 last contents
   | i >= 0 = Just $ unsafeIndex contents i
@@ -84,13 +52,3 @@
 lookupIndex :: Contents -> IndexName -> Maybe (Item -> Int64)
 lookupIndex Contents{indexNames} name
   = (\j Item{indices} -> indices U.! j) <$> V.elemIndex name indexNames
-
-readContents :: Stream -> QueryResult -> IO Contents
-readContents Stream{indexNames, payloadHandle, indexHandle} ((seqnoOffset, payloadOffset), (s1, p1)) = do
-  let length = s1 - seqnoOffset
-  -- byte offset + number of indices
-  let indexSize = 8 * (V.length indexNames + 1)
-  indexBS <- hGetRange indexHandle (indexSize * length) (indexSize * succ seqnoOffset)
-  payloads <- hGetRange payloadHandle (p1 - payloadOffset) payloadOffset
-  let indicess = either error id $ runGet (getIndexVec indexNames length) indexBS
-  pure Contents{..}
diff --git a/src/Database/Franz/Internal.hs b/src/Database/Franz/Internal.hs
deleted file mode 100644
--- a/src/Database/Franz/Internal.hs
+++ /dev/null
@@ -1,58 +0,0 @@
-{-# LANGUAGE RecordWildCards #-}
-module Database.Franz.Internal (getInt64le, runGetRecv, hGetRange) where
-
-import Data.IORef
-import Data.Serialize hiding (getInt64le)
-import Data.Typeable (cast)
-import Foreign.Ptr
-import Foreign.ForeignPtr
-import Foreign.C.Types
-import GHC.IO.Exception
-import GHC.IO.Handle.Internals (withHandle_)
-import GHC.IO.Handle.Types (Handle__(..), Handle)
-import Network.Socket as S
-import Network.Socket.ByteString as SB
-import System.Endian (fromLE64)
-import System.IO.Error
-import System.Posix.Types (Fd(..))
-import qualified Data.ByteString as B
-import qualified Data.ByteString.Internal as B
-import qualified GHC.IO.FD as FD
-import Data.Word (Word8)
-
--- | Better implementation of 'Data.Serialize.getInt64le'
-getInt64le :: Num a => Get a
-getInt64le = fromIntegral . fromLE64 <$> getWord64host
-{-# INLINE getInt64le #-}
-
-runGetRecv :: IORef B.ByteString -> S.Socket -> Get a -> IO (Either String a)
-runGetRecv refBuf sock m = do
-  lo <- readIORef refBuf
-  let go (Done a lo') = do
-        writeIORef refBuf lo'
-        return $ Right a
-      go (Partial cont) = SB.recv sock 4096 >>= go . cont
-      go (Fail str lo') = do
-        writeIORef refBuf lo'
-        return $ Left $ show sock ++ str
-  bs <- if B.null lo
-    then SB.recv sock 4096
-    else pure lo
-  go $ runGetPartial m bs
-
-withFd :: Handle -> (Fd -> IO a) -> IO a
-withFd h f = withHandle_ "withFd" h $ \ Handle__{..} -> do
-  case cast haDevice of
-    Nothing -> ioError (ioeSetErrorString (mkIOError IllegalOperation
-                                           "withFd" (Just h) Nothing)
-                        "handle is not a file descriptor")
-    Just fd -> f (Fd (fromIntegral (FD.fdFD fd)))
-
-foreign import ccall safe "pread"
-  c_pread :: Fd -> Ptr Word8 -> CSize -> CSize -> IO CSize
-
-hGetRange :: Handle -> Int -> Int -> IO B.ByteString
-hGetRange h len ofs = do
-  fptr <- B.mallocByteString len
-  count <- withFd h $ \fd -> withForeignPtr fptr $ \ptr -> c_pread fd ptr (fromIntegral len) (fromIntegral ofs)
-  pure $ B.PS fptr 0 $ fromIntegral count
diff --git a/src/Database/Franz/Internal/Contents.hs b/src/Database/Franz/Internal/Contents.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Internal/Contents.hs
@@ -0,0 +1,56 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE RecordWildCards #-}
+module Database.Franz.Internal.Contents
+    ( IndexVec
+    , Contents(..)
+    , getResponse
+    , readContents
+    )
+    where
+
+import Prelude hiding (length)
+import Data.Serialize hiding (getInt64le)
+import qualified Data.ByteString.Char8 as B
+import qualified Data.Vector as V
+import qualified Data.Vector.Unboxed as U
+import Database.Franz.Internal.Reader
+import Database.Franz.Internal.Protocol
+import Database.Franz.Internal.IO
+import Data.Int
+
+-- A vector containing file offsets and extra indices
+type IndexVec = V.Vector (Int, U.Vector Int64)
+
+data Contents = Contents
+  { indexNames :: !(V.Vector IndexName)
+  , payloads :: !B.ByteString
+  , indicess :: !IndexVec
+  , length :: !Int
+  , payloadOffset :: !Int
+  , seqnoOffset :: !Int
+  }
+
+getIndexVec :: V.Vector IndexName -> Int -> Get IndexVec
+getIndexVec names len = V.replicateM len
+  $ (,) <$> getInt64le <*> U.convert `fmap` traverse (const getInt64le) names
+
+getResponse :: Get Contents
+getResponse = do
+  PayloadHeader seqnoOffset s1 payloadOffset indexNames <- get
+  let length = s1 - seqnoOffset
+  if length <= 0
+    then pure Contents{ payloads = B.empty, indicess = V.empty, ..}
+    else do
+      indicess <- getIndexVec indexNames length
+      payloads <- getByteString $ fst (V.unsafeLast indicess) - payloadOffset
+      pure Contents{..}
+
+readContents :: Stream -> QueryResult -> IO Contents
+readContents Stream{indexNames, payloadHandle, indexHandle} ((seqnoOffset, payloadOffset), (s1, p1)) = do
+  let length = s1 - seqnoOffset
+  -- byte offset + number of indices
+  let indexSize = 8 * (V.length indexNames + 1)
+  indexBS <- hGetRange indexHandle (indexSize * length) (indexSize * succ seqnoOffset)
+  payloads <- hGetRange payloadHandle (p1 - payloadOffset) payloadOffset
+  let indicess = either error id $ runGet (getIndexVec indexNames length) indexBS
+  pure Contents{..}
diff --git a/src/Database/Franz/Internal/Fuse.hs b/src/Database/Franz/Internal/Fuse.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Internal/Fuse.hs
@@ -0,0 +1,38 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE RankNTypes #-}
+module Database.Franz.Internal.Fuse where
+
+import Control.Exception hiding (throw)
+import Control.Monad (when)
+import Control.Retry
+import System.Directory
+import System.Process (ProcessHandle, spawnProcess, cleanupProcess,
+  waitForProcess, getProcessExitCode)
+
+mountFuse :: ([String] -> IO ()) -- logger
+  -> (forall x. String -> IO x) -- throw an exception
+  -> FilePath -> FilePath -> IO ProcessHandle
+mountFuse logger throw src dest = do
+  createDirectoryIfMissing True dest
+  logger ["squashfuse", "-f", src, dest]
+  bracketOnError (spawnProcess "squashfuse" ["-f", src, dest]) (flip (killFuse logger) dest) $ \fuse -> do
+    -- It keeps process handles so that mounted directories are cleaned up
+    -- but there's no easy way to tell when squashfuse finished mounting.
+    -- Wait until the destination becomes non-empty.
+    notMounted <- retrying (limitRetries 5 <> exponentialBackoff 100000) (const pure)
+      $ \status -> getProcessExitCode fuse >>= \case
+        Nothing -> do
+          logger ["Waiting for squashfuse to mount", src, ":", show status]
+          null <$> listDirectory dest
+        Just e -> do
+          removeDirectory dest
+          throw $ "squashfuse exited with " <> show e
+    when notMounted $ throw $ "Failed to mount " <> src
+    return fuse
+
+killFuse :: ([String] -> IO ()) -> ProcessHandle -> FilePath -> IO ()
+killFuse logger fuse path = do
+  cleanupProcess (Nothing, Nothing, Nothing, fuse)
+  e <- waitForProcess fuse
+  logger ["squashfuse:", show e]
+  removeDirectory path
diff --git a/src/Database/Franz/Internal/IO.hs b/src/Database/Franz/Internal/IO.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Internal/IO.hs
@@ -0,0 +1,58 @@
+{-# LANGUAGE RecordWildCards #-}
+module Database.Franz.Internal.IO (getInt64le, runGetRecv, hGetRange) where
+
+import Data.IORef
+import Data.Serialize hiding (getInt64le)
+import Data.Typeable (cast)
+import Foreign.Ptr
+import Foreign.ForeignPtr
+import Foreign.C.Types
+import GHC.IO.Exception
+import GHC.IO.Handle.Internals (withHandle_)
+import GHC.IO.Handle.Types (Handle__(..), Handle)
+import Network.Socket as S
+import Network.Socket.ByteString as SB
+import System.Endian (fromLE64)
+import System.IO.Error
+import System.Posix.Types (Fd(..))
+import qualified Data.ByteString as B
+import qualified Data.ByteString.Internal as B
+import qualified GHC.IO.FD as FD
+import Data.Word (Word8)
+
+-- | Better implementation of 'Data.Serialize.getInt64le'
+getInt64le :: Num a => Get a
+getInt64le = fromIntegral . fromLE64 <$> getWord64host
+{-# INLINE getInt64le #-}
+
+runGetRecv :: IORef B.ByteString -> S.Socket -> Get a -> IO (Either String a)
+runGetRecv refBuf sock m = do
+  lo <- readIORef refBuf
+  let go (Done a lo') = do
+        writeIORef refBuf lo'
+        return $ Right a
+      go (Partial cont) = SB.recv sock 4096 >>= go . cont
+      go (Fail str lo') = do
+        writeIORef refBuf lo'
+        return $ Left $ show sock ++ str
+  bs <- if B.null lo
+    then SB.recv sock 4096
+    else pure lo
+  go $ runGetPartial m bs
+
+withFd :: Handle -> (Fd -> IO a) -> IO a
+withFd h f = withHandle_ "withFd" h $ \ Handle__{..} -> do
+  case cast haDevice of
+    Nothing -> ioError (ioeSetErrorString (mkIOError IllegalOperation
+                                           "withFd" (Just h) Nothing)
+                        "handle is not a file descriptor")
+    Just fd -> f (Fd (fromIntegral (FD.fdFD fd)))
+
+foreign import ccall safe "pread"
+  c_pread :: Fd -> Ptr Word8 -> CSize -> CSize -> IO CSize
+
+hGetRange :: Handle -> Int -> Int -> IO B.ByteString
+hGetRange h len ofs = do
+  fptr <- B.mallocByteString len
+  count <- withFd h $ \fd -> withForeignPtr fptr $ \ptr -> c_pread fd ptr (fromIntegral len) (fromIntegral ofs)
+  pure $ B.PS fptr 0 $ fromIntegral count
diff --git a/src/Database/Franz/Internal/Protocol.hs b/src/Database/Franz/Internal/Protocol.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Internal/Protocol.hs
@@ -0,0 +1,99 @@
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE GeneralisedNewtypeDeriving #-}
+module Database.Franz.Internal.Protocol
+  ( apiVersion
+  , defaultPort
+  , IndexName
+  , StreamName(..)
+  , streamNameToPath
+  , FranzException(..)
+  , RequestType(..)
+  , ItemRef(..)
+  , Query(..)
+  , RawRequest(..)
+  , ResponseId
+  , ResponseHeader(..)
+  , PayloadHeader(..)) where
+
+import Control.Exception (Exception)
+import qualified Data.ByteString.Char8 as B
+import Data.Serialize hiding (getInt64le)
+import Database.Franz.Internal.IO (getInt64le)
+import Data.Hashable (Hashable)
+import Data.String
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as T
+import Network.Socket (PortNumber)
+import GHC.Generics (Generic)
+import qualified Data.ByteString.FastBuilder as BB
+import qualified Data.Vector as V
+
+apiVersion :: B.ByteString
+apiVersion = B.pack "0"
+
+defaultPort :: PortNumber
+defaultPort = 1886
+
+type IndexName = B.ByteString
+
+newtype StreamName = StreamName { unStreamName :: B.ByteString }
+  deriving newtype (Show, Eq, Ord, Hashable, Serialize)
+
+streamNameToPath :: StreamName -> FilePath
+streamNameToPath = T.unpack . T.decodeUtf8 . unStreamName
+
+-- | UTF-8 encoded
+instance IsString StreamName where
+  fromString = StreamName . BB.toStrictByteString . BB.stringUtf8
+
+data FranzException = MalformedRequest !String
+  | StreamNotFound !FilePath
+  | IndexNotFound !IndexName ![IndexName]
+  | InternalError !String
+  | ClientError !String
+  deriving (Show, Generic)
+instance Serialize FranzException
+instance Exception FranzException
+
+data RequestType = AllItems | LastItem deriving (Show, Generic)
+instance Serialize RequestType
+
+data ItemRef = BySeqNum !Int -- ^ sequential number
+  | ByIndex !IndexName !Int -- ^ index name and value
+  deriving (Show, Generic)
+instance Serialize ItemRef
+
+data Query = Query
+  { reqStream :: !StreamName
+  , reqFrom :: !ItemRef -- ^ name of the index to search
+  , reqTo :: !ItemRef -- ^ name of the index to search
+  , reqType :: !RequestType
+  } deriving (Show, Generic)
+instance Serialize Query
+
+data RawRequest
+  = RawRequest !ResponseId !Query
+  | RawClean !ResponseId deriving Generic
+instance Serialize RawRequest
+
+type ResponseId = Int
+
+data ResponseHeader = Response !ResponseId
+    -- ^ response ID, number of streams; there are items satisfying the query
+    | ResponseWait !ResponseId -- ^ response ID; requested elements are not available right now
+    | ResponseError !ResponseId !FranzException -- ^ something went wrong
+    deriving (Show, Generic)
+instance Serialize ResponseHeader
+
+-- | Initial seqno, final seqno, base offset, index names
+data PayloadHeader = PayloadHeader !Int !Int !Int !(V.Vector IndexName)
+
+instance Serialize PayloadHeader where
+  put (PayloadHeader s t u xs) = f s *> f t *> f u
+    *> put (V.length xs) *> mapM_ put xs where
+    f = putInt64le . fromIntegral
+  get = PayloadHeader <$> getInt64le <*> getInt64le <*> getInt64le
+    <*> do
+      len <- get
+      V.replicateM len get
diff --git a/src/Database/Franz/Internal/Reader.hs b/src/Database/Franz/Internal/Reader.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Internal/Reader.hs
@@ -0,0 +1,330 @@
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE GeneralisedNewtypeDeriving #-}
+module Database.Franz.Internal.Reader where
+
+import Control.Concurrent
+import Control.Concurrent.STM
+import Control.Exception
+import Control.Monad
+import Control.Monad.State.Strict
+import Control.Monad.Trans.Maybe
+import Data.Hashable (Hashable)
+import Data.Serialize
+import Database.Franz.Internal.Protocol
+import qualified Data.ByteString.Char8 as B
+import qualified Data.HashMap.Strict as HM
+import qualified Data.IntMap.Strict as IM
+import qualified Data.Vector.Unboxed as U
+import qualified Data.Vector as V
+import Data.Maybe (isJust)
+import GHC.Clock (getMonotonicTime)
+import System.Directory
+import System.FilePath
+import System.IO
+import qualified System.FSNotify as FS
+
+data StreamStatus = CaughtUp | Outdated | Gone deriving Eq
+
+data Stream = Stream
+  { streamPath :: FilePath
+  , vOffsets :: !(TVar (IM.IntMap Int))
+  , indexNames :: !(V.Vector IndexName)
+  , indices :: !(HM.HashMap IndexName (TVar (IM.IntMap Int)))
+  , vCount :: !(TVar Int)
+  , vStatus :: !(TVar StreamStatus)
+  , followThread :: !ThreadId
+  , indexHandle :: !Handle
+  , payloadHandle :: !Handle
+  , vActivity :: !(TVar Activity)
+  }
+
+type Activity = Either Double Int
+
+addActivity :: Stream -> STM ()
+addActivity str = modifyTVar' (vActivity str) $ \case
+  Left _ -> Right 0
+  Right n -> Right (n + 1)
+
+removeActivity :: Stream -> IO ()
+removeActivity str = do
+  now <- getMonotonicTime
+  atomically $ modifyTVar' (vActivity str) $ \case
+    Left _ -> Left now
+    Right n
+      | n <= 0 -> Left now
+      | otherwise -> Right (n - 1)
+
+closeStream :: Stream -> IO ()
+closeStream Stream{..} = do
+  killThread followThread
+  hClose payloadHandle
+  hClose indexHandle
+
+createStream :: FS.WatchManager -> FilePath -> IO Stream
+createStream man path = do
+  let offsetPath = path </> "offsets"
+  let payloadPath = path </> "payloads"
+  exist <- doesFileExist offsetPath
+  unless exist $ throwIO $ StreamNotFound offsetPath
+  initialOffsetsBS <- B.readFile offsetPath
+  payloadHandle <- openBinaryFile payloadPath ReadMode
+  indexNames <- V.fromList . B.lines <$> B.readFile (path </> "indices")
+  let icount = 1 + V.length indexNames
+  let count = B.length initialOffsetsBS `div` (8 * icount)
+  let getI = fromIntegral <$> getInt64le
+  initialIndices <- either (throwIO . InternalError) pure
+    $ runGet (V.replicateM count $ U.replicateM icount getI) initialOffsetsBS
+  let initialOffsets = IM.fromList $ V.toList
+        $ V.zip (V.enumFromN 0 count) $ V.map U.head initialIndices
+  vOffsets <- newTVarIO $! initialOffsets
+  vStatus <- newTVarIO Outdated
+  vCount <- newTVarIO $! IM.size initialOffsets
+  _ <- FS.watchDir man path (\case
+    FS.Modified p _ _ | p == offsetPath -> True
+    FS.Removed p _ _ | p == offsetPath -> True
+    _ -> False)
+    $ \case
+      FS.Modified _ _ _ -> atomically $ writeTVar vStatus Outdated
+      FS.Removed _ _ _ -> atomically $ writeTVar vStatus Gone
+      _ -> pure ()
+
+  vIndices <- forM [1..V.length indexNames] $ \i -> newTVarIO
+    $ IM.fromList $ V.toList $ V.zip (V.map (U.! i) initialIndices) (V.enumFromN 0 count)
+
+  indexHandle <- openFile offsetPath ReadMode
+
+  let final :: Either SomeException () -> IO ()
+      final (Left exc) | Just ThreadKilled <- fromException exc = pure ()
+      final (Left exc) = logFollower [path, "terminated with", show exc]
+      final (Right _) = logFollower [path, "has been removed"]
+
+  -- TODO broadcast an exception if it exits?
+  followThread <- flip forkFinally final $ do
+    forM_ (IM.maxViewWithKey initialOffsets) $ \((i, _), _) ->
+      hSeek indexHandle AbsoluteSeek $ fromIntegral $ succ i * icount * 8
+    fix $ \self -> do
+      bs <- B.hGet indexHandle (8 * icount)
+      if B.null bs
+        then do
+          atomically $ modifyTVar' vStatus $ \case
+            Outdated -> CaughtUp
+            CaughtUp -> CaughtUp
+            Gone -> Gone
+          continue <- atomically $ readTVar vStatus >>= \case
+            CaughtUp -> retry
+            Outdated -> pure True
+            Gone -> pure False
+          when continue self
+        else do
+          ofs : indices <- either (throwIO . InternalError) pure $ runGet (replicateM icount getI) bs
+          atomically $ do
+            i <- readTVar vCount
+            modifyTVar' vOffsets $ IM.insert i ofs
+            forM_ (zip vIndices indices) $ \(v, x) -> modifyTVar' v $ IM.insert (fromIntegral x) i
+            writeTVar vCount $! i + 1
+          self
+
+  let indices = HM.fromList $ zip (V.toList indexNames) vIndices
+
+  vActivity <- getMonotonicTime >>= newTVarIO . Left
+  return Stream{ streamPath = path, ..}
+  where
+    logFollower = hPutStrLn stderr . unwords . (:) "[follower]"
+
+type QueryResult = ((Int, Int) -- starting SeqNo, byte offset
+    , (Int, Int)) -- ending SeqNo, byte offset
+
+range :: Int -- ^ from
+  -> Int -- ^ to
+  -> RequestType
+  -> IM.IntMap Int -- ^ offsets
+  -> (Bool, QueryResult)
+range begin end rt allOffsets = case rt of
+    AllItems -> (ready, (firstItem, maybe firstItem fst $ IM.maxViewWithKey body))
+    LastItem -> case IM.maxViewWithKey body of
+      Nothing -> (False, (zero, zero))
+      Just (ofs', r) -> case IM.maxViewWithKey (IM.union left r) of
+        Just (ofs, _) -> (ready, (ofs, ofs'))
+        Nothing -> (ready, (zero, ofs'))
+  where
+    zero = (-1, 0)
+    ready = isJust lastItem || not (null cont)
+    (wing, lastItem, cont) = IM.splitLookup end allOffsets
+    (left, body) = splitR begin $ maybe id (IM.insert end) lastItem wing
+    firstItem = maybe zero fst $ IM.maxViewWithKey left
+
+splitR :: Int -> IM.IntMap a -> (IM.IntMap a, IM.IntMap a)
+splitR i m = let (l, p, r) = IM.splitLookup i m in (l, maybe id (IM.insert i) p r)
+
+data FranzReader = FranzReader
+  { watchManager :: FS.WatchManager
+  , vStreams :: TVar (HM.HashMap FranzDirectory (HM.HashMap StreamName Stream))
+  }
+
+data ReaperState = ReaperState
+    { -- | How many streams we pruned.
+      prunedStreams :: !Int
+      -- | How many streams we saw in total.
+    , totalStreams :: !Int
+    }
+
+reaper :: Double -- interval
+  -> Double -- lifetime
+  -> FranzReader -> IO ()
+reaper int life FranzReader{..} = forever $ do
+  now <- getMonotonicTime
+  -- Check if stream's activity indicates that we should prune it.
+  let shouldPrune (Left t) = now - t >= life
+      shouldPrune _ = False
+
+      -- Try prunning stream at given filepath. Checks if stream
+      -- should really be pruned first.
+      tryPrune :: FranzDirectory -> StreamName -> STM (Maybe Stream)
+      tryPrune mPath sPath = runMaybeT $ do
+        currentAllStreams <- lift $ readTVar vStreams
+        currentStreams <- MaybeT . pure $ HM.lookup mPath currentAllStreams
+        currentStream <- MaybeT . pure $ HM.lookup sPath currentStreams
+        currentAct <- lift $ readTVar (vActivity currentStream)
+        guard $ shouldPrune currentAct
+        let newStreams = HM.delete sPath currentStreams
+        lift . writeTVar vStreams $ if HM.null newStreams
+          -- Stream we're deleting was the
+          -- last one around.
+          then HM.delete mPath currentAllStreams
+          -- Still have some other streams left for this mount path.
+          -- Keep those only.
+          else HM.insert mPath newStreams currentAllStreams
+        pure currentStream
+
+  -- Take a snapshots of all streams.
+  allStreams <- readTVarIO vStreams
+  -- Traverse the snapshot, looking for streams that currently seem
+  -- out of date. If we find an out-of-date stream, take outer lock
+  -- too, check again, delete it if necessary. Close stream promptly
+  -- after deletion.
+  --
+  -- We could first traverse the whole snapshot, gather potential
+  -- streams for deletion, take lock and delete all these streams from
+  -- the map. However, on an assumption that we normally reaps streams
+  -- and much lower rate than we use them and in favour of locking at
+  -- as small time intervals as possible, we simply traverse the
+  -- snapshot and if we find an out of date stream in the snapshot, we
+  -- take the lock on the whole stream data structure and delete the
+  -- stream. We also check that we aren't leaving an empty map entry
+  -- behind and if we are, we delete it straight away. This stops us
+  -- from having to traverse the whole map later to clean things up as
+  -- well as ensuring that we never leave empty values in the map for
+  -- others to see.
+  --
+  -- While doing all this, keep track of how many streams we saw and
+  -- how many we have closed which is used later to report statistics
+  -- to the user: this is in contrast to doing linear-time traversals
+  -- just to get counts of things we've already traversed.
+  stats <- flip execStateT (ReaperState 0 0) $
+    forM_ (HM.toList allStreams) $ \(mPath, streams) ->
+      forM_ (HM.toList streams) $ \(sPath, stream) -> do
+        modify' $ \s -> s { totalStreams = totalStreams s + 1 }
+        snapAct <- lift $ readTVarIO (vActivity stream)
+        when (shouldPrune snapAct) $ do
+          -- The stream indicates that it's old and should be pruned. Take
+          -- a lock on the outer map, check again inside a transaction and
+          -- amend the maps as necessary.
+          deletedStream'm <- lift . atomically $ tryPrune mPath sPath
+          forM_ deletedStream'm $ \prunedStream -> do
+            lift $ closeStream prunedStream
+            modify' $ \s -> s { prunedStreams = prunedStreams s + 1 }
+
+  when (prunedStreams stats > 0) $ hPutStrLn stderr $ unwords
+    [ "[reaper] closed"
+    , show (prunedStreams stats)
+    , "out of"
+    , show (totalStreams stats)
+    ]
+
+  threadDelay $ floor $ int * 1e6
+
+withFranzReader :: (FranzReader -> IO ()) -> IO ()
+withFranzReader = bracket newFranzReader closeFranzReader
+
+newFranzReader :: IO FranzReader
+newFranzReader = do
+  vStreams <- newTVarIO HM.empty
+  watchManager <- FS.startManager
+  pure FranzReader{..}
+
+closeFranzReader :: FranzReader -> IO ()
+closeFranzReader FranzReader{..} = do
+  FS.stopManager watchManager
+  readTVarIO vStreams >>= mapM_ (mapM_ closeStream)
+
+-- | Globally-configured path which contains franz directories.
+newtype FranzPrefix = FranzPrefix { unFranzPrefix :: FilePath } deriving (Eq, Hashable)
+
+-- | Directory which contains franz streams.
+-- Values of this type serve two purposes:
+--
+-- * Arbitrary prefix so that clients don't have to specify the full path 
+newtype FranzDirectory = FranzDirectory FilePath deriving (Eq, Hashable)
+
+getFranzDirectory :: FranzPrefix -> FranzDirectory -> FilePath
+getFranzDirectory (FranzPrefix prefix) (FranzDirectory dir) = prefix </> dir
+
+getFranzStreamPath :: FranzPrefix -> FranzDirectory -> StreamName -> FilePath
+getFranzStreamPath prefix dir name
+  = getFranzDirectory prefix dir </> streamNameToPath name
+
+handleQuery :: FranzPrefix
+  -> FranzReader
+  -> FranzDirectory
+  -> Query
+  -> (FranzException -> IO r)
+  -> (Stream -> STM (Bool, QueryResult) -> IO r) -> IO r
+handleQuery prefix FranzReader{..} dir (Query name begin_ end_ rt) onError cont
+  = bracket (try acquire) (either mempty removeActivity) $ \case
+  Left err -> onError err
+  Right stream@Stream{..} -> cont stream $ do
+    readTVar vStatus >>= \case
+        Outdated -> retry
+        CaughtUp -> pure ()
+        Gone -> throwSTM $ StreamNotFound streamPath
+    allOffsets <- readTVar vOffsets
+    let finalOffset = case IM.maxViewWithKey allOffsets of
+          Just ((k, _), _) -> k + 1
+          Nothing -> 0
+    let rotate i
+          | i < 0 = finalOffset + i
+          | otherwise = i
+    begin <- case begin_ of
+      BySeqNum i -> pure $ rotate i
+      ByIndex index val -> case HM.lookup index indices of
+        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices
+        Just v -> do
+          m <- readTVar v
+          let (_, wing) = splitR val m
+          return $! maybe maxBound fst $ IM.minView wing
+    end <- case end_ of
+      BySeqNum i -> pure $ rotate i
+      ByIndex index val -> case HM.lookup index indices of
+        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices
+        Just v -> do
+          m <- readTVar v
+          let (body, lastItem, _) = IM.splitLookup val m
+          let body' = maybe id (IM.insert val) lastItem body
+          return $! maybe minBound fst $ IM.maxView body'
+    return $! range begin end rt allOffsets
+  where
+    acquire = join $ atomically $ do
+      allStreams <- readTVar vStreams
+      let !streams = maybe mempty id $ HM.lookup dir allStreams
+      case HM.lookup name streams of
+        Nothing -> pure $ bracketOnError
+          (createStream watchManager $ getFranzStreamPath prefix dir name)
+          closeStream $ \s -> atomically $ do
+            addActivity s
+            modifyTVar' vStreams $ HM.insert dir $ HM.insert name s streams
+            pure s
+        Just s -> do
+          addActivity s
+          pure (pure s)
diff --git a/src/Database/Franz/Internal/URI.hs b/src/Database/Franz/Internal/URI.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Internal/URI.hs
@@ -0,0 +1,53 @@
+{-# LANGUAGE OverloadedStrings #-}
+module Database.Franz.Internal.URI
+  ( FranzPath(..)
+  , toFranzPath
+  , fromFranzPath
+  ) where
+
+import Data.List (stripPrefix)
+import Data.String
+import Network.Socket (HostName, PortNumber)
+import Text.Read (readMaybe)
+
+data FranzPath = FranzPath
+  { franzHost :: !HostName
+  , franzPort :: !PortNumber
+  , franzDir :: !FilePath
+  -- ^ Prefix of franz directories
+  }
+  | LocalFranzPath !FilePath
+  deriving (Show, Eq, Ord)
+
+localPrefix :: IsString a => a
+localPrefix = "franz-local:"
+
+remotePrefix :: IsString a => a
+remotePrefix = "franz://"
+
+-- | Parse a franz URI (franz://host:port/path or franz-local:path).
+toFranzPath :: String -> Either String FranzPath
+toFranzPath uri | Just path <- stripPrefix localPrefix uri = Right $ LocalFranzPath path
+toFranzPath uri = do
+  hostnamePath <- maybe (Left $ "Expecting " <> remotePrefix) Right $ stripPrefix remotePrefix uri
+  (host, path) <- case break (== '/') hostnamePath of
+    (h, '/' : p) -> Right (h, p)
+    _ -> Left "Expecting /"
+  case break (== ':') host of
+    (hostname, ':' : portStr)
+        | Just p <- readMaybe portStr -> Right $ FranzPath hostname p path
+        | otherwise -> Left "Failed to parse the port number"
+    _ -> Right $ FranzPath host 1886 path
+
+-- | Render 'FranzPath' as a franz URI.
+fromFranzPath :: (Monoid a, IsString a) => FranzPath -> a
+fromFranzPath (FranzPath host port path) = mconcat
+  [ remotePrefix
+  , fromString host
+  , ":"
+  , fromString (show port)
+  , "/"
+  , fromString path
+  ]
+
+fromFranzPath (LocalFranzPath path) = localPrefix <> fromString path
diff --git a/src/Database/Franz/Protocol.hs b/src/Database/Franz/Protocol.hs
deleted file mode 100644
--- a/src/Database/Franz/Protocol.hs
+++ /dev/null
@@ -1,99 +0,0 @@
-{-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE DerivingStrategies #-}
-{-# LANGUAGE GeneralisedNewtypeDeriving #-}
-module Database.Franz.Protocol
-  ( apiVersion
-  , defaultPort
-  , IndexName
-  , StreamName(..)
-  , streamNameToPath
-  , FranzException(..)
-  , RequestType(..)
-  , ItemRef(..)
-  , Query(..)
-  , RawRequest(..)
-  , ResponseId
-  , ResponseHeader(..)
-  , PayloadHeader(..)) where
-
-import Control.Exception (Exception)
-import qualified Data.ByteString.Char8 as B
-import Data.Serialize hiding (getInt64le)
-import Database.Franz.Internal (getInt64le)
-import Data.Hashable (Hashable)
-import Data.String
-import qualified Data.Text as T
-import qualified Data.Text.Encoding as T
-import Network.Socket (PortNumber)
-import GHC.Generics (Generic)
-import qualified Data.ByteString.FastBuilder as BB
-import qualified Data.Vector as V
-
-apiVersion :: B.ByteString
-apiVersion = B.pack "0"
-
-defaultPort :: PortNumber
-defaultPort = 1886
-
-type IndexName = B.ByteString
-
-newtype StreamName = StreamName { unStreamName :: B.ByteString }
-  deriving newtype (Show, Eq, Ord, Hashable, Serialize)
-
-streamNameToPath :: StreamName -> FilePath
-streamNameToPath = T.unpack . T.decodeUtf8 . unStreamName
-
--- | UTF-8 encoded
-instance IsString StreamName where
-  fromString = StreamName . BB.toStrictByteString . BB.stringUtf8
-
-data FranzException = MalformedRequest !String
-  | StreamNotFound !FilePath
-  | IndexNotFound !IndexName ![IndexName]
-  | InternalError !String
-  | ClientError !String
-  deriving (Show, Generic)
-instance Serialize FranzException
-instance Exception FranzException
-
-data RequestType = AllItems | LastItem deriving (Show, Generic)
-instance Serialize RequestType
-
-data ItemRef = BySeqNum !Int -- ^ sequential number
-  | ByIndex !IndexName !Int -- ^ index name and value
-  deriving (Show, Generic)
-instance Serialize ItemRef
-
-data Query = Query
-  { reqStream :: !StreamName
-  , reqFrom :: !ItemRef -- ^ name of the index to search
-  , reqTo :: !ItemRef -- ^ name of the index to search
-  , reqType :: !RequestType
-  } deriving (Show, Generic)
-instance Serialize Query
-
-data RawRequest
-  = RawRequest !ResponseId !Query
-  | RawClean !ResponseId deriving Generic
-instance Serialize RawRequest
-
-type ResponseId = Int
-
-data ResponseHeader = Response !ResponseId
-    -- ^ response ID, number of streams; there are items satisfying the query
-    | ResponseWait !ResponseId -- ^ response ID; requested elements are not available right now
-    | ResponseError !ResponseId !FranzException -- ^ something went wrong
-    deriving (Show, Generic)
-instance Serialize ResponseHeader
-
--- | Initial seqno, final seqno, base offset, index names
-data PayloadHeader = PayloadHeader !Int !Int !Int !(V.Vector IndexName)
-
-instance Serialize PayloadHeader where
-  put (PayloadHeader s t u xs) = f s *> f t *> f u
-    *> put (V.length xs) *> mapM_ put xs where
-    f = putInt64le . fromIntegral
-  get = PayloadHeader <$> getInt64le <*> getInt64le <*> getInt64le
-    <*> do
-      len <- get
-      V.replicateM len get
diff --git a/src/Database/Franz/Reader.hs b/src/Database/Franz/Reader.hs
deleted file mode 100644
--- a/src/Database/Franz/Reader.hs
+++ /dev/null
@@ -1,330 +0,0 @@
-{-# LANGUAGE RecordWildCards #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE BangPatterns #-}
-{-# LANGUAGE GeneralisedNewtypeDeriving #-}
-module Database.Franz.Reader where
-
-import Control.Concurrent
-import Control.Concurrent.STM
-import Control.Exception
-import Control.Monad
-import Control.Monad.State.Strict
-import Control.Monad.Trans.Maybe
-import Data.Hashable (Hashable)
-import Data.Serialize
-import Database.Franz.Protocol
-import qualified Data.ByteString.Char8 as B
-import qualified Data.HashMap.Strict as HM
-import qualified Data.IntMap.Strict as IM
-import qualified Data.Vector.Unboxed as U
-import qualified Data.Vector as V
-import Data.Maybe (isJust)
-import GHC.Clock (getMonotonicTime)
-import System.Directory
-import System.FilePath
-import System.IO
-import qualified System.FSNotify as FS
-
-data StreamStatus = CaughtUp | Outdated | Gone deriving Eq
-
-data Stream = Stream
-  { streamPath :: FilePath
-  , vOffsets :: !(TVar (IM.IntMap Int))
-  , indexNames :: !(V.Vector IndexName)
-  , indices :: !(HM.HashMap IndexName (TVar (IM.IntMap Int)))
-  , vCount :: !(TVar Int)
-  , vStatus :: !(TVar StreamStatus)
-  , followThread :: !ThreadId
-  , indexHandle :: !Handle
-  , payloadHandle :: !Handle
-  , vActivity :: !(TVar Activity)
-  }
-
-type Activity = Either Double Int
-
-addActivity :: Stream -> STM ()
-addActivity str = modifyTVar' (vActivity str) $ \case
-  Left _ -> Right 0
-  Right n -> Right (n + 1)
-
-removeActivity :: Stream -> IO ()
-removeActivity str = do
-  now <- getMonotonicTime
-  atomically $ modifyTVar' (vActivity str) $ \case
-    Left _ -> Left now
-    Right n
-      | n <= 0 -> Left now
-      | otherwise -> Right (n - 1)
-
-closeStream :: Stream -> IO ()
-closeStream Stream{..} = do
-  killThread followThread
-  hClose payloadHandle
-  hClose indexHandle
-
-createStream :: FS.WatchManager -> FilePath -> IO Stream
-createStream man path = do
-  let offsetPath = path </> "offsets"
-  let payloadPath = path </> "payloads"
-  exist <- doesFileExist offsetPath
-  unless exist $ throwIO $ StreamNotFound offsetPath
-  initialOffsetsBS <- B.readFile offsetPath
-  payloadHandle <- openBinaryFile payloadPath ReadMode
-  indexNames <- V.fromList . B.lines <$> B.readFile (path </> "indices")
-  let icount = 1 + V.length indexNames
-  let count = B.length initialOffsetsBS `div` (8 * icount)
-  let getI = fromIntegral <$> getInt64le
-  initialIndices <- either (throwIO . InternalError) pure
-    $ runGet (V.replicateM count $ U.replicateM icount getI) initialOffsetsBS
-  let initialOffsets = IM.fromList $ V.toList
-        $ V.zip (V.enumFromN 0 count) $ V.map U.head initialIndices
-  vOffsets <- newTVarIO $! initialOffsets
-  vStatus <- newTVarIO Outdated
-  vCount <- newTVarIO $! IM.size initialOffsets
-  _ <- FS.watchDir man path (\case
-    FS.Modified p _ _ | p == offsetPath -> True
-    FS.Removed p _ _ | p == offsetPath -> True
-    _ -> False)
-    $ \case
-      FS.Modified _ _ _ -> atomically $ writeTVar vStatus Outdated
-      FS.Removed _ _ _ -> atomically $ writeTVar vStatus Gone
-      _ -> pure ()
-
-  vIndices <- forM [1..V.length indexNames] $ \i -> newTVarIO
-    $ IM.fromList $ V.toList $ V.zip (V.map (U.! i) initialIndices) (V.enumFromN 0 count)
-
-  indexHandle <- openFile offsetPath ReadMode
-
-  let final :: Either SomeException () -> IO ()
-      final (Left exc) | Just ThreadKilled <- fromException exc = pure ()
-      final (Left exc) = logFollower [path, "terminated with", show exc]
-      final (Right _) = logFollower [path, "has been removed"]
-
-  -- TODO broadcast an exception if it exits?
-  followThread <- flip forkFinally final $ do
-    forM_ (IM.maxViewWithKey initialOffsets) $ \((i, _), _) ->
-      hSeek indexHandle AbsoluteSeek $ fromIntegral $ succ i * icount * 8
-    fix $ \self -> do
-      bs <- B.hGet indexHandle (8 * icount)
-      if B.null bs
-        then do
-          atomically $ modifyTVar' vStatus $ \case
-            Outdated -> CaughtUp
-            CaughtUp -> CaughtUp
-            Gone -> Gone
-          continue <- atomically $ readTVar vStatus >>= \case
-            CaughtUp -> retry
-            Outdated -> pure True
-            Gone -> pure False
-          when continue self
-        else do
-          ofs : indices <- either (throwIO . InternalError) pure $ runGet (replicateM icount getI) bs
-          atomically $ do
-            i <- readTVar vCount
-            modifyTVar' vOffsets $ IM.insert i ofs
-            forM_ (zip vIndices indices) $ \(v, x) -> modifyTVar' v $ IM.insert (fromIntegral x) i
-            writeTVar vCount $! i + 1
-          self
-
-  let indices = HM.fromList $ zip (V.toList indexNames) vIndices
-
-  vActivity <- getMonotonicTime >>= newTVarIO . Left
-  return Stream{ streamPath = path, ..}
-  where
-    logFollower = hPutStrLn stderr . unwords . (:) "[follower]"
-
-type QueryResult = ((Int, Int) -- starting SeqNo, byte offset
-    , (Int, Int)) -- ending SeqNo, byte offset
-
-range :: Int -- ^ from
-  -> Int -- ^ to
-  -> RequestType
-  -> IM.IntMap Int -- ^ offsets
-  -> (Bool, QueryResult)
-range begin end rt allOffsets = case rt of
-    AllItems -> (ready, (firstItem, maybe firstItem fst $ IM.maxViewWithKey body))
-    LastItem -> case IM.maxViewWithKey body of
-      Nothing -> (False, (zero, zero))
-      Just (ofs', r) -> case IM.maxViewWithKey (IM.union left r) of
-        Just (ofs, _) -> (ready, (ofs, ofs'))
-        Nothing -> (ready, (zero, ofs'))
-  where
-    zero = (-1, 0)
-    ready = isJust lastItem || not (null cont)
-    (wing, lastItem, cont) = IM.splitLookup end allOffsets
-    (left, body) = splitR begin $ maybe id (IM.insert end) lastItem wing
-    firstItem = maybe zero fst $ IM.maxViewWithKey left
-
-splitR :: Int -> IM.IntMap a -> (IM.IntMap a, IM.IntMap a)
-splitR i m = let (l, p, r) = IM.splitLookup i m in (l, maybe id (IM.insert i) p r)
-
-data FranzReader = FranzReader
-  { watchManager :: FS.WatchManager
-  , vStreams :: TVar (HM.HashMap FranzDirectory (HM.HashMap StreamName Stream))
-  }
-
-data ReaperState = ReaperState
-    { -- | How many streams we pruned.
-      prunedStreams :: !Int
-      -- | How many streams we saw in total.
-    , totalStreams :: !Int
-    }
-
-reaper :: Double -- interval
-  -> Double -- lifetime
-  -> FranzReader -> IO ()
-reaper int life FranzReader{..} = forever $ do
-  now <- getMonotonicTime
-  -- Check if stream's activity indicates that we should prune it.
-  let shouldPrune (Left t) = now - t >= life
-      shouldPrune _ = False
-
-      -- Try prunning stream at given filepath. Checks if stream
-      -- should really be pruned first.
-      tryPrune :: FranzDirectory -> StreamName -> STM (Maybe Stream)
-      tryPrune mPath sPath = runMaybeT $ do
-        currentAllStreams <- lift $ readTVar vStreams
-        currentStreams <- MaybeT . pure $ HM.lookup mPath currentAllStreams
-        currentStream <- MaybeT . pure $ HM.lookup sPath currentStreams
-        currentAct <- lift $ readTVar (vActivity currentStream)
-        guard $ shouldPrune currentAct
-        let newStreams = HM.delete sPath currentStreams
-        lift . writeTVar vStreams $ if HM.null newStreams
-          -- Stream we're deleting was the
-          -- last one around.
-          then HM.delete mPath currentAllStreams
-          -- Still have some other streams left for this mount path.
-          -- Keep those only.
-          else HM.insert mPath newStreams currentAllStreams
-        pure currentStream
-
-  -- Take a snapshots of all streams.
-  allStreams <- readTVarIO vStreams
-  -- Traverse the snapshot, looking for streams that currently seem
-  -- out of date. If we find an out-of-date stream, take outer lock
-  -- too, check again, delete it if necessary. Close stream promptly
-  -- after deletion.
-  --
-  -- We could first traverse the whole snapshot, gather potential
-  -- streams for deletion, take lock and delete all these streams from
-  -- the map. However, on an assumption that we normally reaps streams
-  -- and much lower rate than we use them and in favour of locking at
-  -- as small time intervals as possible, we simply traverse the
-  -- snapshot and if we find an out of date stream in the snapshot, we
-  -- take the lock on the whole stream data structure and delete the
-  -- stream. We also check that we aren't leaving an empty map entry
-  -- behind and if we are, we delete it straight away. This stops us
-  -- from having to traverse the whole map later to clean things up as
-  -- well as ensuring that we never leave empty values in the map for
-  -- others to see.
-  --
-  -- While doing all this, keep track of how many streams we saw and
-  -- how many we have closed which is used later to report statistics
-  -- to the user: this is in contrast to doing linear-time traversals
-  -- just to get counts of things we've already traversed.
-  stats <- flip execStateT (ReaperState 0 0) $
-    forM_ (HM.toList allStreams) $ \(mPath, streams) ->
-      forM_ (HM.toList streams) $ \(sPath, stream) -> do
-        modify' $ \s -> s { totalStreams = totalStreams s + 1 }
-        snapAct <- lift $ readTVarIO (vActivity stream)
-        when (shouldPrune snapAct) $ do
-          -- The stream indicates that it's old and should be pruned. Take
-          -- a lock on the outer map, check again inside a transaction and
-          -- amend the maps as necessary.
-          deletedStream'm <- lift . atomically $ tryPrune mPath sPath
-          forM_ deletedStream'm $ \prunedStream -> do
-            lift $ closeStream prunedStream
-            modify' $ \s -> s { prunedStreams = prunedStreams s + 1 }
-
-  when (prunedStreams stats > 0) $ hPutStrLn stderr $ unwords
-    [ "[reaper] closed"
-    , show (prunedStreams stats)
-    , "out of"
-    , show (totalStreams stats)
-    ]
-
-  threadDelay $ floor $ int * 1e6
-
-withFranzReader :: (FranzReader -> IO ()) -> IO ()
-withFranzReader = bracket newFranzReader closeFranzReader
-
-newFranzReader :: IO FranzReader
-newFranzReader = do
-  vStreams <- newTVarIO HM.empty
-  watchManager <- FS.startManager
-  pure FranzReader{..}
-
-closeFranzReader :: FranzReader -> IO ()
-closeFranzReader FranzReader{..} = do
-  FS.stopManager watchManager
-  readTVarIO vStreams >>= mapM_ (mapM_ closeStream)
-
--- | Globally-configured path which contains franz directories.
-newtype FranzPrefix = FranzPrefix { unFranzPrefix :: FilePath } deriving (Eq, Hashable)
-
--- | Directory which contains franz streams.
--- Values of this type serve two purposes:
---
--- * Arbitrary prefix so that clients don't have to specify the full path 
-newtype FranzDirectory = FranzDirectory FilePath deriving (Eq, Hashable)
-
-getFranzDirectory :: FranzPrefix -> FranzDirectory -> FilePath
-getFranzDirectory (FranzPrefix prefix) (FranzDirectory dir) = prefix </> dir
-
-getFranzStreamPath :: FranzPrefix -> FranzDirectory -> StreamName -> FilePath
-getFranzStreamPath prefix dir name
-  = getFranzDirectory prefix dir </> streamNameToPath name
-
-handleQuery :: FranzPrefix
-  -> FranzReader
-  -> FranzDirectory
-  -> Query
-  -> (FranzException -> IO r)
-  -> (Stream -> STM (Bool, QueryResult) -> IO r) -> IO r
-handleQuery prefix FranzReader{..} dir (Query name begin_ end_ rt) onError cont
-  = bracket (try acquire) (either mempty removeActivity) $ \case
-  Left err -> onError err
-  Right stream@Stream{..} -> cont stream $ do
-    readTVar vStatus >>= \case
-        Outdated -> retry
-        CaughtUp -> pure ()
-        Gone -> throwSTM $ StreamNotFound streamPath
-    allOffsets <- readTVar vOffsets
-    let finalOffset = case IM.maxViewWithKey allOffsets of
-          Just ((k, _), _) -> k + 1
-          Nothing -> 0
-    let rotate i
-          | i < 0 = finalOffset + i
-          | otherwise = i
-    begin <- case begin_ of
-      BySeqNum i -> pure $ rotate i
-      ByIndex index val -> case HM.lookup index indices of
-        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices
-        Just v -> do
-          m <- readTVar v
-          let (_, wing) = splitR val m
-          return $! maybe maxBound fst $ IM.minView wing
-    end <- case end_ of
-      BySeqNum i -> pure $ rotate i
-      ByIndex index val -> case HM.lookup index indices of
-        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices
-        Just v -> do
-          m <- readTVar v
-          let (body, lastItem, _) = IM.splitLookup val m
-          let body' = maybe id (IM.insert val) lastItem body
-          return $! maybe minBound fst $ IM.maxView body'
-    return $! range begin end rt allOffsets
-  where
-    acquire = join $ atomically $ do
-      allStreams <- readTVar vStreams
-      let !streams = maybe mempty id $ HM.lookup dir allStreams
-      case HM.lookup name streams of
-        Nothing -> pure $ bracketOnError
-          (createStream watchManager $ getFranzStreamPath prefix dir name)
-          closeStream $ \s -> atomically $ do
-            addActivity s
-            modifyTVar' vStreams $ HM.insert dir $ HM.insert name s streams
-            pure s
-        Just s -> do
-          addActivity s
-          pure (pure s)
diff --git a/src/Database/Franz/Reconnect.hs b/src/Database/Franz/Reconnect.hs
deleted file mode 100644
--- a/src/Database/Franz/Reconnect.hs
+++ /dev/null
@@ -1,96 +0,0 @@
-{-# LANGUAGE RankNTypes #-}
-{-# LANGUAGE RecordWildCards #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE LambdaCase #-}
-module Database.Franz.Reconnect
-  ( Pool
-  , poolLogFunc
-  , poolRetryPolicy
-  , withPool
-  , withReconnection
-  , Reconnect(..)
-  , atomicallyReconnecting
-  , fetchWithPool
-  )
-  where
-
-import Control.Retry (recovering, RetryPolicyM)
-import Control.Concurrent.MVar
-import Control.Concurrent.STM
-import Control.Exception (IOException)
-import Control.Monad.Catch
-import Database.Franz.Client
-
-data Pool = Pool
-  { poolPath :: FranzPath
-  , poolRef :: MVar (Int {- Connection number -}, Maybe Connection)
-  , poolRetryPolicy :: RetryPolicyM IO
-  , poolLogFunc :: String -> IO ()
-  }
-
--- | A wrapper of 'fetch' which calls 'withReconnection' internally
-fetchWithPool
-  :: Pool
-  -> Query
-  -> (STM Response -> IO r)
-  -> IO r
-fetchWithPool pool q cont = withReconnection pool $ \conn -> fetch conn q cont
-
--- | Run an action which takes a 'Connection', reconnecting whenever it throws an exception.
-withReconnection :: Pool -> (Connection -> IO a) -> IO a
-withReconnection Pool{..} cont = recovering
-  poolRetryPolicy
-  [const $ Handler $ \Reconnect -> pure True]
-  body
-  where
-
-    handler ex
-      | Just (ClientError err) <- fromException ex = Just err
-      | Just e <- fromException ex = Just (show (e :: IOException))
-      | Just Reconnect <- fromException ex = Just
-          $ "Connection to " <> fromFranzPath poolPath <> " timed out"
-      | otherwise = Nothing
-
-    body _ = do
-      (i, conn) <- modifyMVar poolRef $ \case
-        (i, Nothing) -> do
-            poolLogFunc $ unwords
-                [ "Connnecting to"
-                , fromFranzPath poolPath
-                ]
-            conn <- tryJust handler (connect poolPath)
-                >>= either (\e -> poolLogFunc e >> throwM Reconnect) pure
-            poolLogFunc $ "Connection #" <> show i <> " established"
-            pure ((i, Just conn), (i, conn))
-        v@(i, Just c) -> pure (v, (i, c))
-
-      tryJust handler (cont conn) >>= \case
-        Right a -> pure a
-        Left msg -> do
-            poolLogFunc msg
-            modifyMVar_ poolRef $ \case
-                -- Don't disconnect if the sequential number is different;
-                -- another thread already established a new connection
-                (j, Just _) | i == j -> (i + 1, Nothing) <$ disconnect conn
-                x -> pure x
-            throwM Reconnect
-
-data Reconnect = Reconnect deriving (Show, Eq)
-instance Exception Reconnect
-
-withPool :: RetryPolicyM IO
-    -> (String -> IO ()) -- ^ diagnostic output
-    -> FranzPath
-    -> (Pool -> IO a)
-    -> IO a
-withPool poolRetryPolicy poolLogFunc poolPath cont = do
-  poolRef <- newMVar (0, Nothing)
-  cont Pool{..} `finally` do
-    (_, conn) <- takeMVar poolRef
-    mapM_ disconnect conn
-
--- | Run an 'STM' action, throwing 'Reconnect' when it exceeds the given timeout.
-atomicallyReconnecting :: Int -- ^ timeout in microseconds
-    -> STM a -> IO a
-atomicallyReconnecting timeout m = atomicallyWithin timeout m
-  >>= maybe (throwM Reconnect) pure
diff --git a/src/Database/Franz/Server.hs b/src/Database/Franz/Server.hs
--- a/src/Database/Franz/Server.hs
+++ b/src/Database/Franz/Server.hs
@@ -5,9 +5,8 @@
 module Database.Franz.Server
   ( Settings(..)
   , startServer
+  , FranzPrefix(..)
   , defaultPort
-  , mountFuse
-  , killFuse
   ) where
 
 import Control.Concurrent
@@ -15,11 +14,11 @@
 import Control.Monad
 import Control.Monad.IO.Class
 import Control.Monad.Trans.Cont
-import Control.Retry
 import Control.Concurrent.STM
-import Database.Franz.Internal
-import Database.Franz.Protocol
-import Database.Franz.Reader
+import Database.Franz.Internal.Fuse
+import Database.Franz.Internal.IO
+import Database.Franz.Internal.Protocol
+import Database.Franz.Internal.Reader
 import Data.ConcurrentResourceMap
 import Data.Serialize
 import qualified Data.IntMap.Strict as IM
@@ -32,8 +31,7 @@
 import qualified Network.Socket as S
 import System.Directory
 import System.IO
-import System.Process (ProcessHandle, spawnProcess, cleanupProcess,
-  waitForProcess, getProcessExitCode, getPid)
+import System.Process (ProcessHandle, getPid)
 
 data Env = Env
   { prefix :: FranzPrefix
@@ -214,36 +212,10 @@
   -> FilePath
   -> IO a -> IO a
 withFuse vMounts release src dst body = withSharedResource vMounts dst
-  (mountFuse src dst)
+  (mountFuse logServer (throwIO . InternalError) src dst)
   (\fuse -> do
     release
-    killFuse fuse dst `finally` do
+    killFuse logServer fuse dst `finally` do
       pid <- getPid fuse
       forM_ pid $ \p -> logServer ["Undead squashfuse detected:", show p])
   (const body)
-
-mountFuse :: FilePath -> FilePath -> IO ProcessHandle
-mountFuse src dest = do
-  createDirectoryIfMissing True dest
-  logServer ["squashfuse", "-f", src, dest]
-  bracketOnError (spawnProcess "squashfuse" ["-f", src, dest]) (flip killFuse dest) $ \fuse -> do
-    -- It keeps process handles so that mounted directories are cleaned up
-    -- but there's no easy way to tell when squashfuse finished mounting.
-    -- Wait until the destination becomes non-empty.
-    notMounted <- retrying (limitRetries 5 <> exponentialBackoff 100000) (const pure)
-      $ \status -> getProcessExitCode fuse >>= \case
-        Nothing -> do
-          logServer ["Waiting for squashfuse to mount", src, ":", show status]
-          null <$> listDirectory dest
-        Just e -> do
-          removeDirectory dest
-          throwIO $ InternalError $ "squashfuse exited with " <> show e
-    when notMounted $ throwIO $ InternalError $ "Failed to mount " <> src
-    return fuse
-
-killFuse :: ProcessHandle -> FilePath -> IO ()
-killFuse fuse path = do
-  cleanupProcess (Nothing, Nothing, Nothing, fuse)
-  e <- waitForProcess fuse
-  logServer ["squashfuse:", show e]
-  removeDirectory path
diff --git a/src/Database/Franz/URI.hs b/src/Database/Franz/URI.hs
deleted file mode 100644
--- a/src/Database/Franz/URI.hs
+++ /dev/null
@@ -1,53 +0,0 @@
-{-# LANGUAGE OverloadedStrings #-}
-module Database.Franz.URI
-  ( FranzPath(..)
-  , toFranzPath
-  , fromFranzPath
-  ) where
-
-import Data.List (stripPrefix)
-import Data.String
-import Network.Socket (HostName, PortNumber)
-import Text.Read (readMaybe)
-
-data FranzPath = FranzPath
-  { franzHost :: !HostName
-  , franzPort :: !PortNumber
-  , franzDir :: !FilePath
-  -- ^ Prefix of franz directories
-  }
-  | LocalFranzPath !FilePath
-  deriving (Show, Eq, Ord)
-
-localPrefix :: IsString a => a
-localPrefix = "franz-local:"
-
-remotePrefix :: IsString a => a
-remotePrefix = "franz://"
-
--- | Parse a franz URI (franz://host:port/path or franz-local:path).
-toFranzPath :: String -> Either String FranzPath
-toFranzPath uri | Just path <- stripPrefix localPrefix uri = Right $ LocalFranzPath path
-toFranzPath uri = do
-  hostnamePath <- maybe (Left $ "Expecting " <> remotePrefix) Right $ stripPrefix remotePrefix uri
-  (host, path) <- case break (== '/') hostnamePath of
-    (h, '/' : p) -> Right (h, p)
-    _ -> Left "Expecting /"
-  case break (== ':') host of
-    (hostname, ':' : portStr)
-        | Just p <- readMaybe portStr -> Right $ FranzPath hostname p path
-        | otherwise -> Left "Failed to parse the port number"
-    _ -> Right $ FranzPath host 1886 path
-
--- | Render 'FranzPath' as a franz URI.
-fromFranzPath :: (Monoid a, IsString a) => FranzPath -> a
-fromFranzPath (FranzPath host port path) = mconcat
-  [ remotePrefix
-  , fromString host
-  , ":"
-  , fromString (show port)
-  , "/"
-  , fromString path
-  ]
-
-fromFranzPath (LocalFranzPath path) = localPrefix <> fromString path