fromBlocks :: Eq h => BlockStore h -> (u -> a -> a) -> Block a -> Block (Maybe u) -> IO (Journal a u)
fromBlocks bs apply checkpoint us = do
  current <- Block.get bs checkpoint
  us' <- (pure . catMaybes) =<< sequenceA =<< Block.gets bs us
  current <- atomically $ newTVar (foldl' (flip apply) current us')
  updateQ <- atomically $ newTQueue
  latestEnqueue <- atomically $ newTVar (pure ())
  err <- atomically $ newTVar Nothing
  get <- pure $ readTVar err >>= maybe (readTVar current) fail
  let flush = join $ get >> readTVar latestEnqueue
  append <- pure $ \vnow u -> do
    cur <- get
    done <- newTSem 0
    writeTVar latestEnqueue (waitTSem done <* get)
    writeTQueue updateQ (Just (u, vnow), signalTSem done)
    let cur' = apply u cur --maybe cur (`apply` cur) u
    (waitTSem done <* get) <$
      if vnow then cur' `seq` writeTVar current cur'
      else pure ()
  _ <- forkIO . forever $ do
    -- write updates to BlockStore asynchronously, notifying of completion or errors
    (uvnow, done) <- atomically $ readTQueue updateQ -- will die when the `Journal` gets GC'd
    id $
      let
        handle :: SomeException -> IO ()
        handle e = atomically $ modifyTVar' err (const (Just . show $ e))
      in case uvnow of
        Nothing -> do
          now <- atomically get
          _ <- catch (Block.modify' bs checkpoint (const now) >>
                      Block.modify' bs us (const Nothing) >>
                      pure ())
                     handle
          atomically done
        Just (u, vnow) ->
          catch (Block.append bs us (Just u) >> update) handle
          where
          update | not vnow  = atomically $ done >> modifyTVar' current (apply u)
                 | otherwise = atomically done
  pure $ Journal get (Updates flush append) (record updateQ get)
  where
  record updateQ get = do
    done <- newTSem 0
    writeTQueue updateQ (Nothing, signalTSem done)
    pure $ waitTSem done <* get

-- | Log a new checkpoint every `updateCount` updates made to the returned journal
fromBlocks :: Eq h => BlockStore h -> (u -> a -> a) -> Block a -> Block (Maybe u) -> IO (Journal a u)
fromBlocks bs apply checkpoint us = do
  current <- Block.get bs checkpoint
  us' <- (pure . catMaybes) =<< sequenceA =<< Block.gets bs us
  current <- atomically $ newTVar (foldl' (flip apply) current us')
  updateQ <- atomically $ newTQueue
  latestEnqueue <- atomically $ newTVar (pure ())
  err <- atomically $ newTVar Nothing
  get <- pure $ readTVar err >>= maybe (readTVar current) fail
  let flush = join $ get >> readTVar latestEnqueue
  append <- pure $ \vnow u -> do
    cur <- get
    done <- newTSem 0
    writeTVar latestEnqueue (waitTSem done <* get)
    writeTQueue updateQ (Just (u, vnow), signalTSem done)
    let cur' = apply u cur --maybe cur (`apply` cur) u
    (waitTSem done <* get) <$
      if vnow then cur' `seq` writeTVar current cur'
      else pure ()
  _ <- forkIO . forever $ do
    -- write updates to BlockStore asynchronously, notifying of completion or errors
    (uvnow, done) <- atomically $ readTQueue updateQ -- will die when the `Journal` gets GC'd
    id $
      let
        handle :: SomeException -> IO ()
        handle e = atomically $ modifyTVar' err (const (Just . show $ e))
      in case uvnow of
        Nothing -> do
          now <- atomically get
          _ <- catch (Block.modify' bs checkpoint (const now) >>
                      Block.modify' bs us (const Nothing) >>
                      pure ())
                     handle
          atomically done
        Just (u, vnow) ->
          catch (Block.append bs us (Just u) >> update) handle
          where
          update | not vnow  = atomically $ done >> modifyTVar' current (apply u)
                 | otherwise = atomically done
  pure $ Journal get (Updates flush append) (record updateQ get)
  where
  record updateQ get = do
    done <- newTSem 0
    writeTQueue updateQ (Nothing, signalTSem done)
    pure $ waitTSem done <* get

-- | Log a new checkpoint every `updateCount` updates made to the returned journal
fromBlocks bs apply checkpoint us = do
  current <- Block.get bs checkpoint
  us' <- (pure . catMaybes) =<< sequenceA =<< Block.gets bs us
  current <- atomically $ newTVar (foldl' (flip apply) current us')
  updateQ <- atomically $ newTQueue
  latestEnqueue <- atomically $ newTVar (pure ())
  err <- atomically $ newTVar Nothing
  get <- pure $ readTVar err >>= maybe (readTVar current) fail
  let flush = join $ get >> readTVar latestEnqueue
  append <- pure $ \vnow u -> do
    cur <- get
    done <- newTSem 0
    writeTVar latestEnqueue (waitTSem done <* get)
    writeTQueue updateQ (Just (u, vnow), signalTSem done)
    let cur' = apply u cur --maybe cur (`apply` cur) u
    (waitTSem done <* get) <$
      if vnow then cur' `seq` writeTVar current cur'
      else pure ()
  _ <- forkIO . forever $ do
    -- write updates to BlockStore asynchronously, notifying of completion or errors
    (uvnow, done) <- atomically $ readTQueue updateQ -- will die when the `Journal` gets GC'd
    id $
      let
        handle :: SomeException -> IO ()
        handle e = atomically $ modifyTVar' err (const (Just . show $ e))
      in case uvnow of
        Nothing -> do
          now <- atomically get
          _ <- catch (Block.modify' bs checkpoint (const now) >>
                      Block.modify' bs us (const Nothing) >>
                      pure ())
                     handle
          atomically done
        Just (u, vnow) ->
          catch (Block.append bs us (Just u) >> update) handle
          where
          update | not vnow  = atomically $ done >> modifyTVar' current (apply u)
                 | otherwise = atomically done
  pure $ Journal get (Updates flush append) (record updateQ get)
  where
  record updateQ get = do
    done <- newTSem 0
    writeTQueue updateQ (Nothing, signalTSem done)
    pure $ waitTSem done <* get

-- | Log a new checkpoint every `updateCount` updates made to the returned journal