packages feed

sqlite-simple-effectful-0.1.0.0: src/Effectful/SQLite/Simple/RW.hs

{-# LANGUAGE OverloadedRecordDot #-}
{-# LANGUAGE OverloadedStrings #-}
{-# OPTIONS_GHC -Wno-redundant-constraints #-}

{- ORMOLU_DISABLE -}
{- | A dynamic effect that lets us use a __pooled__ SQLite `Database.SQLite.Simple.Connection`.

SQLite allows multiple connections to read/write concurrently, but concurrent writes will lead to contention, performance degradation, and @SQLITE_BUSY@ errors.
We avoid this by:

  * Having separate pools for reading and writing.
  * Configuring the write pool to have a maximum of 1 connection, thus serializing all writes.

__WARNING__: This interpreter sets the database's journal mode to [WAL](https://sqlite.org/wal.html),
so that readers will not block the writer and the writer will not block readers.

Note that even in WAL mode, [@SQLITE_BUSY@ errors can still occur](https://sqlite.org/wal.html#sometimes_queries_return_sqlite_busy_in_wal_mode).

To connect to multiple databases, see "Effectful.SQLite.Simple.RW.Labeled".

>>> import Effectful
>>> import Effectful.Concurrent (runConcurrent)
>>> import Effectful.SQLite.Simple.RW (SQLite)
>>> import Effectful.SQLite.Simple.RW qualified as SQL

>>> :{
app :: (SQLite :> es) => Eff es ()
app = do
  SQL.useWriteConnection \conn -> do
    SQL.withImmediateTransaction conn do
      SQL.query_ @User conn "SELECT * FROM users"
      SQL.execute conn "DELETE FROM users WHERE username = ?" (SQL.Only "dcastro")
:}

>>> :{
main :: IO ()
main = do
  let dbPath = "users.db"
  pools <-
    SQL.newPools
      =<< SQL.newPoolsConfig
        (SQL.open dbPath) -- Action to create a new read connection.
        60                -- Read connections' idle timeout in seconds.
        32                -- Max number of read connections.
        (SQL.open dbPath) -- Action to create a new write connection.
        60                -- Write connections' idle timeout in seconds.
  app
    & SQL.runSQLiteWithPools pools
    & runConcurrent
    & runEff
:}

-}
{- ORMOLU_ENABLE -}
module Effectful.SQLite.Simple.RW
  ( -- * Effects
    SQLite (..),

    -- * Use connection
    -- $useConnection
    useReadConnection,
    useWriteConnection,
    RWConnection (..),
    ConnMode (..),

    -- * Interpreters
    runSQLiteWithPools,
    Pools (..),
    newPools,
    PoolsConfig (..),
    newPoolsConfig,

    -- * Connections
    S.open,
    S.close,
    S.withConnection,
    S.setTrace,

    -- * Queries that return results
    query,
    query_,
    queryWith,
    queryWith_,
    queryNamed,
    lastInsertRowId,
    changes,
    totalChanges,

    -- * Queries that stream results
    fold,
    fold_,
    foldNamed,

    -- * Statements that do not return results
    execute,
    execute_,
    executeMany,
    executeNamed,
    S.field,

    -- * Transactions
    withTransaction,
    withImmediateTransaction,
    withExclusiveTransaction,
    withSavepoint,

    -- * Low-level statement API for stream access and prepared statements
    openStatement,
    closeStatement,
    withStatement,
    bind,
    bindNamed,
    reset,
    columnName,
    columnCount,
    withBind,
    nextRow,

    -- ** Exceptions
    S.FormatError (..),
    S.ResultError (..),
    S.SQLError (..),
    S.Error (..),

    -- * Types
    S.Query (..),
    S.Connection (..),
    S.ToRow (..),
    S.FromRow (..),
    S.Only (..),
    (S.:.) (..),
    S.SQLData (..),
    S.Statement (..),
    S.ColumnIndex (..),
    S.NamedParam (..),
  )
where

import Data.Function ((&))
import Data.Int (Int64)
import Data.Text (Text)
import Database.SQLite.Simple (ColumnIndex, FromRow, NamedParam, Query, Statement, ToRow)
import Database.SQLite.Simple qualified as S
import Database.SQLite.Simple.FromRow (RowParser)
import Effectful
import Effectful.Dispatch.Dynamic (interpret, localSeqUnlift, send)
import Effectful.Dispatch.Static (seqUnliftIO, unsafeEff, unsafeEff_)
import GHC.Stack (HasCallStack)
import UnliftIO.Pool qualified as Pool

-- $setup
-- Clear all imports before running doctests
-- >>> :m
-- >>> :set -XOverloadedStrings
-- >>> import Data.Function ((&))
-- >>> import Data.Text (Text)
-- >>> import Effectful.SQLite.Simple (FromRow(fromRow))
-- >>> data User
-- >>> instance FromRow User where fromRow = undefined

----------------------------------------------------------------------------
-- Effect
----------------------------------------------------------------------------

{-
  NOTE: The rationale for having separate "read" and "write" pools has been documented here: @(ref:concurrency)
-}

-- | The mode in which a connection is acquired.
data ConnMode = Read | Write

-- | A connection that can be acquired in "read" or "write" mode.
-- This determines which kind of operations can be performed with the connection.
newtype RWConnection (mode :: ConnMode) = RWConnection {getConn :: S.Connection}

data SQLite :: Effect where
  UseReadConnection :: (RWConnection 'Read -> m a) -> SQLite m a
  UseWriteConnection :: (RWConnection 'Write -> m a) -> SQLite m a

type instance DispatchOf SQLite = 'Dynamic

{- ORMOLU_DISABLE -}
{- $useConnection

The `useReadConnection` and `useWriteConnection` operations retrieve a pooled connection from the context that can be used to run "read" or "write" operations.

If the action throws an exception of any type, the connection is closed and not returned to the pool.

__WARNING__:

* The connection must not be manually closed.
* The connection must not escape the scope of `useReadConnection` or `useWriteConnection`.
* `useWriteConnection` calls must not be nested.
* When `useWriteConnection` is used together with other locking primitives, the locks must always be acquired in the same order to avoid deadlocks.

-}
{- ORMOLU_ENABLE -}

-- | Retrieve the connection from the context and run the given "read" operations with it.
useReadConnection :: (HasCallStack, SQLite :> es) => (RWConnection 'Read -> Eff es a) -> Eff es a
useReadConnection = send . UseReadConnection

-- | Retrieve the connection from the context and run the given "read" or "write" operations with it.
useWriteConnection :: (HasCallStack, SQLite :> es) => (RWConnection 'Write -> Eff es a) -> Eff es a
useWriteConnection = send . UseWriteConnection

----------------------------------------------------------------------------
-- Interpreters
----------------------------------------------------------------------------

-- | Interprets the 'SQLite' effect by using 2 connection pools for reading and writing.
--
-- __WARNING__: This interpreter sets the database's journal mode to [WAL](https://sqlite.org/wal.html),
-- so that readers will not block the writer and the writer will not block readers.
--
-- Note that even in WAL mode, [@SQLITE_BUSY@ errors can still occur](https://sqlite.org/wal.html#sometimes_queries_return_sqlite_busy_in_wal_mode).
runSQLiteWithPools :: (HasCallStack, IOE :> es) => Pools -> Eff (SQLite ': es) a -> Eff es a
runSQLiteWithPools pools action = do
  {-
    NOTE: we set the WAL mode upfront.

    To avoid SQLITE_BUSY errors, all connections must be made in WAL mode.
    We can't set WAL mode lazily (e.g. every time a pool creates a new connection),
    because having the "read pool" execute `PRAGMA journal_mode=WAL` would cause it to acquire
    an exclusive lock on the database, which it must not do.
    Therefore, we must do it eagerly, here.
  -}
  Pool.withResource pools.writePool \conn -> do
    liftIO $ S.execute_ conn.getConn "PRAGMA journal_mode=WAL"

  action & interpret \env -> \case
    UseReadConnection action -> do
      localSeqUnlift env \unlift -> do
        Pool.withResource pools.readPool \conn -> do
          unlift $ action conn
    UseWriteConnection action -> do
      localSeqUnlift env \unlift -> do
        Pool.withResource pools.writePool \conn -> do
          unlift $ action conn

data Pools = Pools
  { readPool :: Pool.Pool (RWConnection 'Read),
    writePool :: Pool.Pool (RWConnection 'Write)
  }

newPools :: (MonadUnliftIO m) => PoolsConfig -> m Pools
newPools (PoolsConfig readPoolConfig writePoolConfig) = do
  readPool <- Pool.newPool readPoolConfig
  writePool <- Pool.newPool writePoolConfig
  pure Pools {readPool, writePool}

data PoolsConfig = PoolsConfig
  { readPoolConfig :: Pool.PoolConfig (RWConnection 'Read),
    writePoolConfig :: Pool.PoolConfig (RWConnection 'Write)
  }

newPoolsConfig ::
  (MonadUnliftIO m) =>
  -- | The action to create a new connection for reading from the database.
  m S.Connection ->
  -- | The number of seconds for which an unused read connection is kept around. The smallest acceptable value is 0.5.
  --
  -- Note: the elapsed time before destroying a connection may be a little longer than requested, as the collector thread wakes at 1-second intervals.
  Double ->
  -- | The maximum number of read connections to keep open at once. The smallest acceptable value is 1.
  Int ->
  -- | The action to create a new connection for writing to the database.
  m S.Connection ->
  -- | The number of seconds for which an unused write connection is kept around. The smallest acceptable value is 0.5.
  --
  -- Note: the elapsed time before destroying a connection may be a little longer than requested, as the collector thread wakes at 1-second intervals.
  Double ->
  m PoolsConfig
newPoolsConfig mkReadConn readTTLSeconds readMaxConns mkWriteConn writeTTLSeconds = do
  readPoolConfig <-
    Pool.mkDefaultPoolConfig
      (RWConnection @'Read <$> mkReadConn)
      (liftIO . S.close . getConn)
      readTTLSeconds
      readMaxConns

  let writeMaxConns = 1
  writePoolConfig <-
    Pool.mkDefaultPoolConfig
      (RWConnection @'Write <$> mkWriteConn)
      (liftIO . S.close . getConn)
      writeTTLSeconds
      writeMaxConns
  pure PoolsConfig {readPoolConfig, writePoolConfig}

----------------------------------------------------------------------------
-- Operations
----------------------------------------------------------------------------

query :: forall q r es mode. (SQLite :> es) => (ToRow q, FromRow r) => RWConnection mode -> Query -> q -> Eff es [r]
query conn q params = unsafeEff_ $ S.query conn.getConn q params

query_ :: forall r es mode. (SQLite :> es) => (FromRow r) => RWConnection mode -> Query -> Eff es [r]
query_ conn q = unsafeEff_ $ S.query_ conn.getConn q

queryWith :: forall q r es mode. (SQLite :> es) => (ToRow q) => RowParser r -> RWConnection mode -> Query -> q -> Eff es [r]
queryWith parser conn q params = unsafeEff_ $ S.queryWith parser conn.getConn q params

queryWith_ :: forall r es mode. (SQLite :> es) => RowParser r -> RWConnection mode -> Query -> Eff es [r]
queryWith_ parser conn q = unsafeEff_ $ S.queryWith_ parser conn.getConn q

queryNamed :: forall r es mode. (SQLite :> es) => (FromRow r) => RWConnection mode -> Query -> [NamedParam] -> Eff es [r]
queryNamed conn q params = unsafeEff_ $ S.queryNamed conn.getConn q params

lastInsertRowId :: forall es mode. (SQLite :> es) => RWConnection mode -> Eff es Int64
lastInsertRowId = unsafeEff_ . S.lastInsertRowId . getConn

changes :: forall es mode. (SQLite :> es) => RWConnection mode -> Eff es Int
changes = unsafeEff_ . S.changes . getConn

totalChanges :: forall es mode. (SQLite :> es) => RWConnection mode -> Eff es Int
totalChanges = unsafeEff_ . S.totalChanges . getConn

fold :: forall row params a es mode. (SQLite :> es) => (FromRow row, ToRow params) => RWConnection mode -> Query -> params -> a -> (a -> row -> Eff es a) -> Eff es a
fold conn q params initialState action =
  unsafeEffWithUnlift \unlift -> do
    S.fold conn.getConn q params initialState \a row ->
      unlift $ action a row

fold_ :: forall row a es mode. (SQLite :> es) => (FromRow row) => RWConnection mode -> Query -> a -> (a -> row -> Eff es a) -> Eff es a
fold_ conn q initialState action =
  unsafeEffWithUnlift \unlift -> do
    S.fold_ conn.getConn q initialState \a row ->
      unlift $ action a row

foldNamed :: forall row a es mode. (SQLite :> es) => (FromRow row) => RWConnection mode -> Query -> [NamedParam] -> a -> (a -> row -> Eff es a) -> Eff es a
foldNamed conn q params initialState action =
  unsafeEffWithUnlift \unlift -> do
    S.foldNamed conn.getConn q params initialState \a row ->
      unlift $ action a row

execute :: forall q es. (SQLite :> es) => (ToRow q) => RWConnection 'Write -> Query -> q -> Eff es ()
execute conn q params = unsafeEff_ $ S.execute conn.getConn q params

execute_ :: forall es. (SQLite :> es) => RWConnection 'Write -> Query -> Eff es ()
execute_ conn q = unsafeEff_ $ S.execute_ conn.getConn q

executeMany :: forall q es. (SQLite :> es) => (ToRow q) => RWConnection 'Write -> Query -> [q] -> Eff es ()
executeMany conn q params = unsafeEff_ $ S.executeMany conn.getConn q params

executeNamed :: forall es. (SQLite :> es) => RWConnection 'Write -> Query -> [NamedParam] -> Eff es ()
executeNamed conn q params = unsafeEff_ $ S.executeNamed conn.getConn q params

withTransaction :: forall a es mode. (SQLite :> es) => RWConnection mode -> Eff es a -> Eff es a
withTransaction conn action =
  unsafeEffWithUnlift \unlift -> do
    S.withTransaction conn.getConn $ unlift action

withImmediateTransaction :: forall a es. (SQLite :> es) => RWConnection 'Write -> Eff es a -> Eff es a
withImmediateTransaction conn action =
  unsafeEffWithUnlift \unlift -> do
    S.withImmediateTransaction conn.getConn $ unlift action

withExclusiveTransaction :: forall a es. (SQLite :> es) => RWConnection 'Write -> Eff es a -> Eff es a
withExclusiveTransaction conn action =
  unsafeEffWithUnlift \unlift -> do
    S.withExclusiveTransaction conn.getConn $ unlift action

withSavepoint :: forall a es. (SQLite :> es) => RWConnection 'Write -> Eff es a -> Eff es a
withSavepoint conn action =
  unsafeEffWithUnlift \unlift -> do
    S.withSavepoint conn.getConn $ unlift action

openStatement :: forall es. (SQLite :> es) => RWConnection 'Write -> Query -> Eff es Statement
openStatement conn q = unsafeEff_ $ S.openStatement conn.getConn q

closeStatement :: forall es. (SQLite :> es) => Statement -> Eff es ()
closeStatement stmt = unsafeEff_ $ S.closeStatement stmt

withStatement :: forall a es. (SQLite :> es) => RWConnection 'Write -> Query -> (Statement -> Eff es a) -> Eff es a
withStatement conn q action =
  unsafeEffWithUnlift \unlift -> do
    S.withStatement conn.getConn q (unlift . action)

bind :: forall params es. (SQLite :> es) => (ToRow params) => Statement -> params -> Eff es ()
bind stmt params = unsafeEff_ $ S.bind stmt params

bindNamed :: forall es. (SQLite :> es) => Statement -> [NamedParam] -> Eff es ()
bindNamed stmt params = unsafeEff_ $ S.bindNamed stmt params

reset :: forall es. (SQLite :> es) => Statement -> Eff es ()
reset stmt = unsafeEff_ $ S.reset stmt

columnName :: forall es. (SQLite :> es) => Statement -> ColumnIndex -> Eff es Text
columnName stmt idx = unsafeEff_ $ S.columnName stmt idx

columnCount :: forall es. (SQLite :> es) => Statement -> Eff es ColumnIndex
columnCount stmt = unsafeEff_ $ S.columnCount stmt

withBind :: forall params a es. (SQLite :> es) => (ToRow params) => Statement -> params -> Eff es a -> Eff es a
withBind stmt params action =
  unsafeEffWithUnlift \unlift -> do
    S.withBind stmt params $ unlift action

nextRow :: forall r es. (SQLite :> es) => (FromRow r) => Statement -> Eff es (Maybe r)
nextRow stmt = unsafeEff_ $ S.nextRow stmt

----------------------------------------------------------------------------
-- Utils
----------------------------------------------------------------------------

unsafeEffWithUnlift :: forall a es. (SQLite :> es) => ((forall x. Eff es x -> IO x) -> IO a) -> Eff es a
unsafeEffWithUnlift action =
  unsafeEff \env -> do
    seqUnliftIO env \unlift -> do
      action unlift