Interpreter.hs

module Pgmq.Effectful.Interpreter
  ( -- * Interpreters
    runPgmq,

    -- * Error Types
    PgmqRuntimeError (..),
    fromUsageError,
    isTransient,
    isAmbiguousReply,

    -- * Legacy Error Types (deprecated; will be removed in 0.3.0)
    PgmqError (..),
  )
where

import Control.Exception (Exception)
import Data.Text (Text)
import Data.Text qualified as T
import Effectful (Eff, IOE, (:>))
import Effectful qualified
import Effectful.Dispatch.Dynamic (interpret)
import Effectful.Error.Static (Error, throwError)
import GHC.Generics (Generic)
import Hasql.Errors qualified as HasqlErrors
import Hasql.Pool (Pool, UsageError)
import Hasql.Pool qualified as Pool
import Hasql.Session qualified
import Pgmq.Effectful.Effect (Pgmq (..))
import Pgmq.Hasql.Sessions qualified as Sessions

-- | Structured runtime error for pgmq-effectful operations.
--
-- Constructors mirror 'Hasql.Pool.UsageError' so the mapping is direct.
-- The inner 'HasqlErrors.ConnectionError' and 'HasqlErrors.SessionError'
-- types are sum types with detailed constructors; see the Hasql
-- documentation for their full shapes.
data PgmqRuntimeError
  = -- | Timed out waiting for a connection from the pool.
    PgmqAcquisitionTimeout
  | -- | Failed to establish a connection to PostgreSQL.
    PgmqConnectionError HasqlErrors.ConnectionError
  | -- | Error during session execution (SQL statement, decoding, etc.).
    PgmqSessionError HasqlErrors.SessionError
  deriving stock (Show, Eq, Generic)

instance Exception PgmqRuntimeError

-- | Convert hasql-pool's 'UsageError' into the structured
-- 'PgmqRuntimeError'.
fromUsageError :: UsageError -> PgmqRuntimeError
fromUsageError = \case
  Pool.AcquisitionTimeoutUsageError -> PgmqAcquisitionTimeout
  Pool.ConnectionUsageError e -> PgmqConnectionError e
  Pool.SessionUsageError e -> PgmqSessionError e

-- | Is this error plausibly transient — i.e., worth retrying?
--
-- Returns 'True' for acquisition timeouts, networking connection errors,
-- uncategorized libpq connection errors, session-level connection drops,
-- server-reported statement or script errors whose SQLSTATE names a
-- transient condition: @40001@ (serialization_failure), @40P01@
-- (deadlock_detected), @55P03@ (lock_not_available), @57P01@
-- (admin_shutdown), @57P02@ (crash_shutdown), @57P03@ (cannot_connect_now),
-- and class @53@ (insufficient resources); server errors in statements or
-- scripts with no SQLSTATE (client-side connection faults, see
-- 'isAmbiguousReply'); and the driver's stray-result shape
-- @UnexpectedRowCountStatementError 1 1 1@. All other errors
-- (authentication failure, compatibility mismatches, missing types, other
-- server errors, genuine decode and row-count mismatches, driver bugs) are
-- treated as permanent.
--
-- A connection lost while waiting for a reply leaves the dead connection in
-- the pool, so the next attempt typically fails once more with a transient
-- 'HasqlErrors.ConnectionSessionError' (which evicts it) before a fresh
-- connection succeeds.
--
-- Note: 'HasqlErrors.OtherConnectionError' is classed as transient here
-- despite hasql's documentation calling it \"not transient by default\",
-- because it is the catch-all for unrecognized libpq errors and classing
-- the unknown as transient errs toward letting retries happen.
isTransient :: PgmqRuntimeError -> Bool
isTransient = \case
  PgmqAcquisitionTimeout -> True
  PgmqConnectionError e -> case e of
    HasqlErrors.NetworkingConnectionError _ -> True
    HasqlErrors.AuthenticationConnectionError _ -> False
    HasqlErrors.CompatibilityConnectionError _ -> False
    HasqlErrors.OtherConnectionError _ -> True
  PgmqSessionError e -> case e of
    HasqlErrors.ConnectionSessionError _ -> True
    HasqlErrors.StatementSessionError _ _ _ _ _ statementError ->
      isLostReplyStatementError statementError || isTransientStatementError statementError
    HasqlErrors.ScriptSessionError _ serverError ->
      isLostReplyServerError serverError || isTransientServerError serverError
    HasqlErrors.MissingTypesSessionError _ -> False
    HasqlErrors.DriverSessionError _ -> False

-- | Was the statement delivered but its reply lost, so that the caller cannot
-- tell whether PostgreSQL applied it?
--
-- 'True' for the two values hasql 1.10 produces when the connection dies while
-- it is waiting for a reply: a server error with no SQLSTATE (libpq's own
-- connection-loss result, which carries no fields) inside a statement or script
-- error, and @UnexpectedRowCountStatementError 1 1 1@ (a stray second result;
-- hasql discards the first, so the value cannot show whether that was a server
-- fatal error or a completed command). 'False' for everything else: a
-- 'HasqlErrors.ConnectionSessionError' means the send failed and nothing reached
-- the server, and every other error carries the server's verdict or the
-- driver's own decode verdict.
--
-- Every ambiguous reply is transient, so @isAmbiguousReply err@ implies
-- @isTransient err@. Gate the retry of non-idempotent operations on it: an
-- ambiguous reply to a send may have committed, so check the durable message
-- keys before sending again.
isAmbiguousReply :: PgmqRuntimeError -> Bool
isAmbiguousReply = \case
  PgmqSessionError (HasqlErrors.StatementSessionError _ _ _ _ _ statementError) ->
    isLostReplyStatementError statementError
  PgmqSessionError (HasqlErrors.ScriptSessionError _ serverError) ->
    isLostReplyServerError serverError
  _ -> False

-- | The statement-level shapes of a lost reply (see 'isAmbiguousReply').
--
-- hasql 1.10 reports a stray second result as @UnexpectedRowCountStatementError 1 1 1@
-- (@Hasql.Engine.Errors.fromRecvError@, @TooManyResultsError@ branch). A genuine
-- row-count mismatch can never carry an actual count inside its own bounds, so
-- this exact value is reserved for the driver's stray-result case, which for the
-- single-command roundtrips this library issues only happens when libpq appends
-- its own connection-loss result after a server fatal error (for example
-- @57P01@ from @pg_terminate_backend@). Every other actual count is a real
-- decoder mismatch and stays permanent.
isLostReplyStatementError :: HasqlErrors.StatementError -> Bool
isLostReplyStatementError = \case
  HasqlErrors.ServerStatementError serverError -> isLostReplyServerError serverError
  HasqlErrors.UnexpectedRowCountStatementError 1 1 1 -> True
  _ -> False

-- | Every @ErrorResponse@ PostgreSQL sends carries a SQLSTATE, so an empty code
-- means libpq manufactured the error on the client — \"server closed the
-- connection unexpectedly\" and its relatives — which hasql 1.10 surfaces with
-- every field empty.
isLostReplyServerError :: HasqlErrors.ServerError -> Bool
isLostReplyServerError (HasqlErrors.ServerError code _ _ _ _) = T.null code

-- | Server-reported statement errors worth retrying: those whose SQLSTATE is in
-- the transient whitelist.
isTransientStatementError :: HasqlErrors.StatementError -> Bool
isTransientStatementError = \case
  HasqlErrors.ServerStatementError serverError -> isTransientServerError serverError
  _ -> False

-- | Server errors (from a statement or a script) whose SQLSTATE is in the
-- transient whitelist.
isTransientServerError :: HasqlErrors.ServerError -> Bool
isTransientServerError (HasqlErrors.ServerError code _ _ _ _) = isTransientSqlState code

-- | SQLSTATEs that indicate a transient, retry-worthy condition: @40001@
-- serialization_failure, @40P01@ deadlock_detected, @55P03@
-- lock_not_available, @57P01@ admin_shutdown, @57P02@ crash_shutdown,
-- @57P03@ cannot_connect_now, and class @53@ (insufficient resources —
-- 53000\/53100\/53200\/53300\/53400). These arrive as server errors inside
-- 'HasqlErrors.StatementSessionError' or 'HasqlErrors.ScriptSessionError' and are precisely the errors retries
-- exist for. Everything else reported by the server is permanent for retry
-- purposes.
isTransientSqlState :: Text -> Bool
isTransientSqlState code =
  code `elem` ["40001", "40P01", "55P03", "57P01", "57P02", "57P03"]
    || "53" `T.isPrefixOf` code

-- | Legacy error type. Retained for one release cycle; migrate to
-- 'PgmqRuntimeError'. Will be removed in pgmq-effectful 0.3.0.
newtype PgmqError = PgmqPoolError UsageError
  deriving stock (Show)
{-# DEPRECATED PgmqError, PgmqPoolError "Use PgmqRuntimeError instead (available from Pgmq.Effectful). The legacy types will be removed in pgmq-effectful 0.3.0." #-}

-- | Run the Pgmq effect using a connection pool.
-- Errors are thrown via the 'Error' effect as 'PgmqRuntimeError'.
runPgmq ::
  (IOE :> es, Error PgmqRuntimeError :> es) =>
  Pool ->
  Eff (Pgmq : es) a ->
  Eff es a
runPgmq pool = interpret $ \_ -> \case
  -- Queue Management
  CreateQueue q -> runSession pool $ Sessions.createQueue q
  DropQueue q -> runSession pool $ Sessions.dropQueue q
  CreatePartitionedQueue q -> runSession pool $ Sessions.createPartitionedQueue q
  CreatePartitionedQueueWithPremake q n -> runSession pool $ Sessions.createPartitionedQueueWithPremake q n
  CreateUnloggedQueue q -> runSession pool $ Sessions.createUnloggedQueue q
  DetachArchive _q -> pure ()
  EnableNotifyInsert config -> runSession pool $ Sessions.enableNotifyInsert config
  DisableNotifyInsert q -> runSession pool $ Sessions.disableNotifyInsert q
  CreateFifoIndex q -> runSession pool $ Sessions.createFifoIndex q
  CreateFifoIndexesAll -> runSession pool Sessions.createFifoIndexesAll
  -- Message Operations
  SendMessage msg -> runSession pool $ Sessions.sendMessage msg
  SendMessageForLater msg -> runSession pool $ Sessions.sendMessageForLater msg
  BatchSendMessage msgs -> runSession pool $ Sessions.batchSendMessage msgs
  BatchSendMessageForLater msgs -> runSession pool $ Sessions.batchSendMessageForLater msgs
  SendMessageWithHeaders msg -> runSession pool $ Sessions.sendMessageWithHeaders msg
  SendMessageWithHeadersForLater msg -> runSession pool $ Sessions.sendMessageWithHeadersForLater msg
  BatchSendMessageWithHeaders msgs -> runSession pool $ Sessions.batchSendMessageWithHeaders msgs
  BatchSendMessageWithHeadersForLater msgs -> runSession pool $ Sessions.batchSendMessageWithHeadersForLater msgs
  ReadMessage query -> runSession pool $ Sessions.readMessage query
  DeleteMessage query -> runSession pool $ Sessions.deleteMessage query
  BatchDeleteMessages query -> runSession pool $ Sessions.batchDeleteMessages query
  ArchiveMessage query -> runSession pool $ Sessions.archiveMessage query
  BatchArchiveMessages query -> runSession pool $ Sessions.batchArchiveMessages query
  DeleteAllMessagesFromQueue q -> runSession pool $ Sessions.deleteAllMessagesFromQueue q
  ChangeVisibilityTimeout query -> runSession pool $ Sessions.changeVisibilityTimeout query
  BatchChangeVisibilityTimeout query -> runSession pool $ Sessions.batchChangeVisibilityTimeout query
  -- Timestamp-based VT (pgmq 1.10.0+)
  SetVisibilityTimeoutAt query -> runSession pool $ Sessions.setVisibilityTimeoutAt query
  BatchSetVisibilityTimeoutAt query -> runSession pool $ Sessions.batchSetVisibilityTimeoutAt query
  ReadWithPoll query -> runSession pool $ Sessions.readWithPoll query
  Pop query -> runSession pool $ Sessions.pop query
  -- FIFO Read (pgmq 1.8.0+)
  ReadGroupedHead query -> runSession pool $ Sessions.readGroupedHead query
  ReadGroupedHeadWithPoll query -> runSession pool $ Sessions.readGroupedHeadWithPoll query
  ReadGrouped query -> runSession pool $ Sessions.readGrouped query
  ReadGroupedWithPoll query -> runSession pool $ Sessions.readGroupedWithPoll query
  -- Round-robin FIFO Read (pgmq 1.9.0+)
  ReadGroupedRoundRobin query -> runSession pool $ Sessions.readGroupedRoundRobin query
  ReadGroupedRoundRobinWithPoll query -> runSession pool $ Sessions.readGroupedRoundRobinWithPoll query
  -- Topic Management (pgmq 1.11.0+)
  BindTopic params -> runSession pool $ Sessions.bindTopic params
  UnbindTopic params -> runSession pool $ Sessions.unbindTopic params
  ValidateRoutingKey key -> runSession pool $ Sessions.validateRoutingKey key
  ValidateTopicPattern pat -> runSession pool $ Sessions.validateTopicPattern pat
  TestRouting key -> runSession pool $ Sessions.testRouting key
  ListTopicBindings -> runSession pool Sessions.listTopicBindings
  ListTopicBindingsForQueue q -> runSession pool $ Sessions.listTopicBindingsForQueue q
  -- Topic Sending (pgmq 1.11.0+)
  SendTopic msg -> runSession pool $ Sessions.sendTopic msg
  SendTopicWithHeaders msg -> runSession pool $ Sessions.sendTopicWithHeaders msg
  BatchSendTopic msgs -> runSession pool $ Sessions.batchSendTopic msgs
  BatchSendTopicForLater msgs -> runSession pool $ Sessions.batchSendTopicForLater msgs
  BatchSendTopicWithHeaders msgs -> runSession pool $ Sessions.batchSendTopicWithHeaders msgs
  BatchSendTopicWithHeadersForLater msgs -> runSession pool $ Sessions.batchSendTopicWithHeadersForLater msgs
  -- Notification Management (pgmq 1.11.0+)
  ListNotifyInsertThrottles -> runSession pool Sessions.listNotifyInsertThrottles
  UpdateNotifyInsert params -> runSession pool $ Sessions.updateNotifyInsert params
  -- Queue Observability
  ListQueues -> runSession pool Sessions.listQueues
  ListQueuesUnvalidated -> runSession pool Sessions.listQueuesUnvalidated
  ListFifoIndexQueueNames -> runSession pool Sessions.listFifoIndexQueueNames
  QueueMetrics q -> runSession pool $ Sessions.queueMetrics q
  AllQueueMetrics -> runSession pool Sessions.allQueueMetrics
  -- Non-destructive inspection
  PeekMessages args -> runSession pool $ Sessions.peekMessages args
  PeekArchivedMessages args -> runSession pool $ Sessions.peekArchivedMessages args
  LookupMessage args -> runSession pool $ Sessions.lookupMessage args
  LookupArchivedMessage args -> runSession pool $ Sessions.lookupArchivedMessage args
  QueueMetricsUnvalidated q -> runSession pool $ Sessions.queueMetricsUnvalidated q

-- Internal helper
runSession ::
  (IOE :> es, Error PgmqRuntimeError :> es) =>
  Pool ->
  Hasql.Session.Session a ->
  Eff es a
runSession pool session = do
  result <- Effectful.liftIO $ Pool.use pool session
  case result of
    Left err -> throwError $ fromUsageError err
    Right a -> pure a