Bolty.hs

-- | Haskell driver for Neo4j over the BOLT protocol (4.4+).
--
-- This is the main entry point for using bolty. It re-exports everything
-- needed to connect, run queries, manage transactions, and decode results.
--
-- @
-- import qualified Database.Bolty as Bolt
-- import           Data.Default   (def)
--
-- main :: IO ()
-- main = do
--   cfg  <- Bolt.'validateConfig' def{ Bolt.'scheme' = Bolt.'Basic' \"neo4j\" \"password\" }
--   conn <- Bolt.'connect' cfg
--   rows <- Bolt.'runBolt' conn $
--     Bolt.'queryWith' (Bolt.'field' \"n\" Bolt.'node') \"MATCH (n) RETURN n LIMIT 10\" mempty
--   print rows  -- Either DecodeError (Vector Node)
--   Bolt.'close' conn
-- @
module Database.Bolty
  ( -- * Configuration
    Config(..)
  , ValidatedConfig
  , validateConfig
  , Scheme(..)
  , Routing(..)
  , UserAgent(..)
  , Version(..)
    -- * Connection
  , Connection
  , connect
  , close
  , reset
  , ping
  , logon
  , logoff
  , connectionVersion
  , connectionAgent
  , connectionId
  , connectionTelemetryEnabled
  , connectionServerIdleTimeout
    -- * Telemetry
  , TelemetryApi(..)
  , sendTelemetry
    -- * Transactions
  , withTransaction
  , withTransaction'
  , withReadTransaction
  , withReadTransaction'
  , withTxMode
  , withRetryTransaction
  , withRetryTransactionMode
    -- * Retry configuration
  , RetryConfig(..)
  , defaultRetryConfig
    -- * Error helpers
  , isTransient
  , isRoutingError
    -- * Types
  , Bolt(..)
  , Record
  , Error(..)
    -- * Bolt value extractors
  , asNull, asBool, asInt, asFloat, asBytes, asText
  , asList, asDict, asNode, asRelationship, asPath
    -- * Query metadata
  , QueryMeta(..)
    -- * Connection pooling
  , BoltPool
  , PoolConfig(..)
  , defaultPoolConfig
  , ValidationStrategy(..)
  , PoolCounters(..)
  , createPool
  , destroyPool
  , withConnection
  , CheckedOutConnection(..)
  , acquireConnection
  , releaseConnection
  , releaseConnectionOnError
  , poolCounters
    -- * Routing
  , AccessMode(..)
  , RoutingTable(..)
  , RoutingPool
  , RoutingPoolConfig(..)
  , defaultRoutingPoolConfig
  , createRoutingPool
  , destroyRoutingPool
  , withRoutingConnection
  , acquireRoutingConnection
  , withRoutingTransaction
  , invalidateRoutingTable
  , getRoutingTable
    -- * Sessions
  , Session
  , SessionConfig(..)
  , defaultSessionConfig
  , BookmarkManager
  , createSession
  , createRoutingSession
  , readTransaction
  , writeTransaction
  , getLastBookmarks
    -- * Plan / Profile
  , PlanNode(..)
  , ProfileNode(..)
  , queryExplain
  , queryProfile
  , renderPlan
    -- * Query logging
  , QueryLog(..)
    -- * Notifications
  , Notification(..)
  , Severity(..)
  , Position(..)
    -- * Query statistics
  , QueryStats(..)
  , formatStatsLine
    -- * BoltM monad
  , BoltM
  , runBolt
    -- * Queries
  , query
  , queryWith
  , queryResult
  , queryMeta
  , queryMetaWith
  , queryResultMeta
  , execute
    -- * Record decoding
  , DecodeError(..)
  , Decode(..)
  , RowDecoder(..)
  , FromBolt(..)
  , ToBolt(..)
  , Database.Bolty.Decode.bool
  , Database.Bolty.Decode.int
  , int64
  , Database.Bolty.Decode.float
  , Database.Bolty.Decode.text
  , Database.Bolty.Decode.bytes
  , nullable, list, dict
  , Database.Bolty.Decode.uuid
  , utcTime, day, timeOfDay
  , aesonValue
  , nodeProperty, nodePropertyOptional, Database.Bolty.Decode.nodeLabels, Database.Bolty.Decode.nodeProperties
  , Database.Bolty.Decode.node
  , Database.Bolty.Decode.relationship
  , Database.Bolty.Decode.path
  , column, field
    -- * Result sets
  , ResultSet(..)
  , decodeResultSet
  , decodeHead
  , groupByField
    -- * PackStream re-exports
  , Ps(..)
    -- * Aeson interop
  , parsePs
  , psToAeson
  , propsToAeson
  , propsFromAeson
    -- * Admin / sandbox helpers (Database.Bolty.Admin)
  , ReadOutcome(..)
  , WriteOutcome(..)
  , ReadModeViolation(..)
  , WriteCountMismatch(..)
  , describeReadOutcome
  , describeWriteOutcome
  , withReadOnlyTransaction
  , withReadOnlyTransaction'
  , withConfirmedWriteTransaction
  , withConfirmedWriteTransaction'
  , totalAffected
  ) where

import           Database.Bolty.Connection          (queryPWithFieldsIO, queryPMetaIO, withTxMode)
import           Database.Bolty.Logging            (QueryLog(..))
import           Database.Bolty.Plan              (PlanNode(..), ProfileNode(..), renderPlan)
import           Database.Bolty.Notification      (Notification(..), Severity(..), Position(..))
import           Database.Bolty.Stats             (QueryStats(..), formatStatsLine)
import           Database.Bolty.Decode
import           Database.Bolty.Connection.Config   (validateConfig)
import qualified Database.Bolty.Connection.Pipe     as P
import           Database.Bolty.Connection.Pipe     (connectionVersion, connectionAgent,
                                                    connectionId, connectionTelemetryEnabled,
                                                    connectionServerIdleTimeout)
import           Database.Bolty.Connection.Type
import           Database.Bolty.Message.Request     (Begin(Begin), TelemetryApi(..))
import           Database.Bolty.Message.Response
import           Database.Bolty.Pool
import           Database.Bolty.Record
import           Database.Bolty.ResultSet
import           Database.Bolty.AccessMode          (AccessMode(..))
import           Database.Bolty.Routing (RoutingPool, RoutingPoolConfig(..),
                                        defaultRoutingPoolConfig, createRoutingPool,
                                        destroyRoutingPool, withRoutingConnection,
                                        acquireRoutingConnection,
                                        withRoutingTransaction, invalidateRoutingTable,
                                        getRoutingTable)
import           Database.Bolty.Session (Session, SessionConfig(..), defaultSessionConfig,
                                        BookmarkManager, createSession, createRoutingSession,
                                        readTransaction, writeTransaction, getLastBookmarks)
import           Database.Bolty.Value.Type          (Bolt(..), asNull, asBool, asInt, asFloat
                                                    , asBytes, asText, asList, asDict
                                                    , asNode, asRelationship, asPath
                                                    , parsePs, psToAeson
                                                    , propsToAeson, propsFromAeson)
import           Database.Bolty.Admin               (ReadOutcome(..), WriteOutcome(..)
                                                    , ReadModeViolation(..)
                                                    , WriteCountMismatch(..)
                                                    , describeReadOutcome, describeWriteOutcome
                                                    , withReadOnlyTransaction, withReadOnlyTransaction'
                                                    , withConfirmedWriteTransaction
                                                    , withConfirmedWriteTransaction'
                                                    , totalAffected)
import           Database.Bolty.Connection.Version  (Version(..))

import           Data.Kind                           (Type)
import           Control.Concurrent                  (threadDelay)
import           Control.Exception                  (SomeException, fromException, onException,
                                                     throwIO, try)
import           Control.Monad                      (void)
import           Control.Monad.Reader               (ReaderT(..))
import           Control.Monad.Trans                (MonadIO(..))
import           Data.Text                          (Text)
import           GHC.Stack                          (HasCallStack)
import qualified Data.HashMap.Lazy                  as H
import qualified Data.Vector                        as V
import           Data.PackStream.Ps                 (Ps(..))


-- | Connect to a Neo4j server using a validated configuration.
connect :: (MonadIO m, HasCallStack) => ValidatedConfig -> m Connection
connect = P.connect


-- | Close a connection to a Neo4j server.
close :: (MonadIO m, HasCallStack) => Connection -> m ()
close = P.close


-- | Reset the connection state after an error.
reset :: (MonadIO m, HasCallStack) => Connection -> m ()
reset = P.reset


-- | Check if a connection is alive by sending RESET.
-- Returns True if healthy, False otherwise.
ping :: MonadIO m => Connection -> m Bool
ping = P.ping


-- | Send LOGON with credentials (Bolt 5.1+).
-- Transitions from Authentication to Ready state.
logon :: (MonadIO m, HasCallStack) => Connection -> Scheme -> m ()
logon = P.logon


-- | Send LOGOFF (Bolt 5.1+).
-- Transitions from Ready to Authentication state.
logoff :: (MonadIO m, HasCallStack) => Connection -> m ()
logoff = P.logoff


-- | Send a TELEMETRY message if the server supports it (Bolt 5.4+).
-- No-op if the server does not support telemetry.
sendTelemetry :: (MonadIO m, HasCallStack) => Connection -> TelemetryApi -> m ()
sendTelemetry = P.sendTelemetry


-- ---------------------------------------------------------------------------
-- BoltM monad
-- ---------------------------------------------------------------------------

-- | A monad for running queries against a Neo4j connection.
-- Use 'runBolt' to execute a 'BoltM' action with a 'Connection'.
type BoltM :: Type -> Type
type role BoltM nominal
newtype BoltM a = BoltM (ReaderT Connection IO a)
  deriving newtype (Functor, Applicative, Monad, MonadIO)

-- | Run a 'BoltM' action with a 'Connection'.
runBolt :: Connection -> BoltM a -> IO a
runBolt conn (BoltM m) = runReaderT m conn


-- ---------------------------------------------------------------------------
-- Queries (BoltM)
-- ---------------------------------------------------------------------------

-- | Run a Cypher query, decode each record using the 'FromBolt' instance, and
-- return the decoded rows or a 'DecodeError'. Pass 'mempty' for no parameters.
query :: (FromBolt a, HasCallStack) => Text -> H.HashMap Text Ps -> BoltM (Either DecodeError (V.Vector a))
query = queryWith rowDecoder

-- | Like 'query', but with an explicit 'RowDecoder' instead of using 'FromBolt'.
queryWith :: HasCallStack => RowDecoder a -> Text -> H.HashMap Text Ps -> BoltM (Either DecodeError (V.Vector a))
queryWith decoder cypher params = BoltM $ ReaderT $ \conn -> do
  (cols, recs) <- queryPWithFieldsIO conn cypher params
  pure $ decodeRows decoder cols recs

-- | Run a Cypher query and return a 'ResultSet' (field names + records).
-- Use 'decodeResultSet' or 'groupByField' for multi-pass decoding.
-- Pass 'mempty' for no parameters.
queryResult :: HasCallStack => Text -> H.HashMap Text Ps -> BoltM ResultSet
queryResult cypher params = BoltM $ ReaderT $ \conn -> do
  (fs, rs) <- queryPWithFieldsIO conn cypher params
  pure $ ResultSet fs rs

-- | Run a Cypher query, decode each record using the 'FromBolt' instance, and
-- return the decoded rows paired with server metadata ('QueryMeta').
-- Pass 'mempty' for no parameters.
queryMeta :: (FromBolt a, HasCallStack) => Text -> H.HashMap Text Ps -> BoltM (Either DecodeError (V.Vector a), QueryMeta)
queryMeta = queryMetaWith rowDecoder

-- | Like 'queryMeta', but with an explicit 'RowDecoder' instead of using 'FromBolt'.
queryMetaWith :: HasCallStack => RowDecoder a -> Text -> H.HashMap Text Ps -> BoltM (Either DecodeError (V.Vector a), QueryMeta)
queryMetaWith decoder cypher params = BoltM $ ReaderT $ \conn -> do
  (fs, rs, qi) <- queryPMetaIO conn cypher params
  pure (decodeRows decoder fs rs, qi)

-- | Run a Cypher query and return a 'ResultSet' paired with server metadata
-- ('QueryMeta' — timing, statistics, notifications, plan/profile).
-- Pass 'mempty' for no parameters.
queryResultMeta :: HasCallStack => Text -> H.HashMap Text Ps -> BoltM (ResultSet, QueryMeta)
queryResultMeta cypher params = BoltM $ ReaderT $ \conn -> do
  (fs, rs, qi) <- queryPMetaIO conn cypher params
  pure (ResultSet fs rs, qi)


-- | Run a Cypher query for side effects only, discarding the result.
-- Pass 'mempty' for no parameters.
execute :: HasCallStack => Text -> H.HashMap Text Ps -> BoltM ()
execute cypher params = void $ queryResult cypher params


-- | Run EXPLAIN on a Cypher query, returning the query plan without executing it.
queryExplain :: HasCallStack => Text -> H.HashMap Text Ps -> BoltM (Maybe PlanNode)
queryExplain cypher params = do
  (_, meta) <- queryResultMeta ("EXPLAIN " <> cypher) params
  pure (parsedPlan meta)


-- | Run PROFILE on a Cypher query, returning both decoded rows and the profile tree
-- with actual execution statistics (db hits, rows, timing, etc.).
queryProfile :: (FromBolt a, HasCallStack)
             => Text -> H.HashMap Text Ps
             -> BoltM (Either DecodeError (V.Vector a), ProfileNode)
queryProfile cypher params = do
  (result, meta) <- queryMeta ("PROFILE " <> cypher) params
  case parsedProfile meta of
    Just prof -> pure (result, prof)
    Nothing   -> error "queryProfile: no profile in response"


-- | Run an action inside an explicit /write/ transaction. Automatically
-- commits on success and rolls back on failure.
--
-- Equivalent to @'withTxMode' 'WriteAccess'@. Use 'withReadTransaction'
-- for a read-only transaction that can be cluster-routed to a follower
-- and has the server enforce no writes.
withTransaction :: HasCallStack => Connection -> (Connection -> IO a) -> IO a
withTransaction = withTxMode WriteAccess


-- | Run an action inside an explicit /read/ transaction. Automatically
-- commits on success and rolls back on failure.
--
-- The transaction is opened with 'ReadAccess': in cluster setups the
-- server routes to a follower, and any direct write attempt is rejected
-- upfront with an @AccessMode@ error (a Neo4j 5+ feature). For application
-- code this is the right pair to 'withTransaction'.
--
-- For an admin-tool variant that additionally inspects 'parsedStats'
-- post-hoc and reports a rollback as a typed 'ReadOutcome' instead of
-- a thrown exception, see 'Database.Bolty.Admin.withReadOnlyTransaction'.
--
-- /Not the same as/ 'Database.Bolty.Session.readTransaction', which runs
-- a multi-step action against a session with bookmarks.
withReadTransaction :: HasCallStack => Connection -> (Connection -> IO a) -> IO a
withReadTransaction = withTxMode ReadAccess


-- | Run a transaction with automatic retry on transient failures.
-- Uses exponential backoff between retries.
withRetryTransaction :: HasCallStack
                     => RetryConfig -> Connection -> (Connection -> IO a) -> IO a
withRetryTransaction = withRetryTransactionMode WriteAccess


-- | Like 'withRetryTransaction' but parameterised on the 'AccessMode',
-- so the same retry machinery serves both 'withTransaction'' and
-- 'withReadTransaction''.
withRetryTransactionMode
  :: HasCallStack
  => AccessMode -> RetryConfig -> Connection -> (Connection -> IO a) -> IO a
withRetryTransactionMode mode RetryConfig{maxRetries, initialDelay, maxDelay} conn action =
    go maxRetries initialDelay
  where
    go 0 _     = withTxMode mode conn action
    go n delay = do
      result <- try $ withTxMode mode conn action
      case result of
        Right x -> pure x
        Left (e :: SomeException) -> case fromException e :: Maybe Error of
          Just err | isTransient err -> do
            threadDelay delay
            go (n - 1) (min maxDelay (delay * 2))
          _ -> throwIO e


-- | Convenience: acquire a pooled connection, run a retrying write
-- transaction, and release.
withTransaction' :: HasCallStack => BoltPool -> (Connection -> IO a) -> IO a
withTransaction' pool@BoltPool{bpRetryConfig} action =
  withConnection pool $ \conn ->
    withRetryTransactionMode WriteAccess bpRetryConfig conn action


-- | Convenience: acquire a pooled connection, run a retrying read
-- transaction, and release.
withReadTransaction' :: HasCallStack => BoltPool -> (Connection -> IO a) -> IO a
withReadTransaction' pool@BoltPool{bpRetryConfig} action =
  withConnection pool $ \conn ->
    withRetryTransactionMode ReadAccess bpRetryConfig conn action