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
  , withRetryTransaction
  , withTransaction'
    -- * 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
    -- * Query logging
  , QueryLog(..)
    -- * Notifications
  , Notification(..)
  , Severity(..)
  , Position(..)
    -- * Query statistics
  , QueryStats(..)
    -- * 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(..)
  ) where

import           Database.Bolty.Connection          (queryPWithFieldsIO, queryPMetaIO)
import           Database.Bolty.Logging            (QueryLog(..))
import           Database.Bolty.Plan              (PlanNode(..), ProfileNode(..))
import           Database.Bolty.Notification      (Notification(..), Severity(..), Position(..))
import           Database.Bolty.Stats             (QueryStats(..))
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.Routing (AccessMode(..), 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)
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 transaction. Automatically commits on success
-- and rolls back on failure.
withTransaction :: HasCallStack => Connection -> (Connection -> IO a) -> IO a
withTransaction conn action = do
  P.beginTx conn defaultBegin
  result <- action conn `onException` P.tryRollback conn
  _ <- P.commitTx conn
  pure result
  where
    defaultBegin = Begin V.empty Nothing H.empty 'w' Nothing Nothing


-- | Run a transaction with automatic retry on transient failures.
-- Uses exponential backoff between retries.
withRetryTransaction :: HasCallStack
                     => RetryConfig -> Connection -> (Connection -> IO a) -> IO a
withRetryTransaction RetryConfig{maxRetries, initialDelay, maxDelay} conn action =
    go maxRetries initialDelay
  where
    go 0 _     = withTransaction conn action
    go n delay = do
      result <- try $ withTransaction 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 transaction, and release.
withTransaction' :: HasCallStack => BoltPool -> (Connection -> IO a) -> IO a
withTransaction' pool@BoltPool{bpRetryConfig} action =
  withConnection pool $ \conn ->
    withRetryTransaction bpRetryConfig conn action