Async.hs

{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeFamilies #-}
{-# LANGUAGE TypeOperators #-}
{-# LANGUAGE UnicodeSyntax #-}

module Database.Async (withAsyncHandle) where

import Control.Concurrent.Async
import Control.Concurrent.STM.TMChan
import Control.Exception.Safe
import Database.Handle
import Imm.Logger (LogLevel (..), log)
import qualified Imm.Logger as Logger
import Imm.Pretty
import Output (putDocLn)
import qualified Output

-- | Run an existing database handle in a separate thread
withAsyncHandle ∷ m ~ IO ⇒ Logger.Handle m → Output.Handle m → Handle m → (Handle m → m ()) → m ()
withAsyncHandle logger output database f = do
  channel ← newTMChanIO

  thread ← async $ fix $ \recurse → do
    maybeMessage ← atomically $ readTMChan channel
    forM_ maybeMessage $ \message → do
      interpret database message
      recurse

  log logger Debug "Database thread started"
  catchAny (f $ command channel) $ \e → do
    log logger Error $ pretty $ displayException e
    putDocLn output $ pretty $ displayException e
  log logger Debug "Closing database thread..."
  atomically (closeTMChan channel) >> wait thread

withMVar ∷ MonadIO m ⇒ TMChan a → ((b → m ()) → a) → m b
withMVar channel message = do
  result ← newEmptyMVar
  atomically $ writeTMChan channel (message $ putMVar result)
  takeMVar result

withMVarE
  ∷ (MonadIO m, MonadThrow m, Exception e)
  ⇒ TMChan a
  → ((Either e b → m ()) → a)
  → m b
withMVarE channel message = do
  result ← newEmptyMVar
  atomically $ writeTMChan channel (message $ putMVar result)
  takeMVar result >>= either throwM return

asyncCommand ∷ MonadIO m ⇒ TMChan a → (m () → a) → m ()
asyncCommand channel message = do
  atomically $ writeTMChan channel (message $ return ())

command ∷ MonadIO m ⇒ MonadThrow m ⇒ TMChan (HandleF (m ())) → Handle m
command channel =
  Handle
    { _describeDatabase = withMVar channel DescribeDatabase
    , _fetchAllFeeds = withMVar channel FetchAllFeeds
    , _fetchFeed = withMVarE channel . FetchFeed
    , _fetchAllItems = withMVar channel FetchAllItems
    , _fetchItems = withMVar channel . FetchItems
    , _fetchItem = withMVarE channel . FetchItem
    , _updateFeedDefinition = asyncCommand channel . UpdateFeedDefinition
    , _updateFeedStatus = asyncCommand channel . UpdateFeedStatus
    , _updateItemStatus = asyncCommand channel . UpdateItemStatus
    , _insertFeed = withMVarE channel . InsertFeed
    , _insertItem = withMVar channel . InsertItem
    , _deleteFeed = asyncCommand channel . DeleteFeed
    , _purge = asyncCommand channel Purge
    , _commit = asyncCommand channel Commit
    }