Types.hs

module Pgmq.Hasql.Statements.Types
  ( SendMessage (..),
    SendMessageForLater (..),
    BatchSendMessage (..),
    BatchSendMessageForLater (..),
    SendMessageWithHeaders (..),
    SendMessageWithHeadersForLater (..),
    BatchSendMessageWithHeaders (..),
    BatchSendMessageWithHeadersForLater (..),
    ReadMessage (..),
    PopMessage (..),
    MessageQuery (..),
    BatchMessageQuery (..),
    VisibilityTimeoutQuery (..),
    BatchVisibilityTimeoutQuery (..),
    -- Timestamp-based VT types (pgmq 1.10.0+)
    VisibilityTimeoutAtQuery (..),
    BatchVisibilityTimeoutAtQuery (..),
    ReadWithPollMessage (..),
    EnableNotifyInsert (..),
    CreatePartitionedQueue (..),
    QueueMetrics (..),
    -- FIFO read types (pgmq 1.8.0+)
    ReadGrouped (..),
    ReadGroupedWithPoll (..),
    -- Non-destructive inspection (no upstream function)
    PeekMessages (..),
    LookupMessage (..),
    -- Topic types (pgmq 1.11.0+)
    BindTopic (..),
    UnbindTopic (..),
    SendTopic (..),
    SendTopicWithHeaders (..),
    BatchSendTopic (..),
    BatchSendTopicForLater (..),
    BatchSendTopicWithHeaders (..),
    BatchSendTopicWithHeadersForLater (..),
    UpdateNotifyInsert (..),
  )
where

import Data.Aeson (KeyValue, Value, object, pairs, (.=))
import Pgmq.Hasql.Prelude hiding ((.=))
import Pgmq.Types (MessageBody, MessageHeaders, MessageId, QueueName, RoutingKey, TopicPattern)

type Delay = Int32

data SendMessage = SendMessage
  { queueName :: !QueueName,
    messageBody :: !MessageBody,
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

data VisibilityTimeoutQuery = VisibilityTimeoutQuery
  { queueName :: !QueueName,
    messageId :: !MessageId,
    visibilityTimeoutOffset :: !Int32
  }
  deriving stock (Generic)

-- | Batch visibility timeout update (pgmq 1.8.0+)
data BatchVisibilityTimeoutQuery = BatchVisibilityTimeoutQuery
  { queueName :: !QueueName,
    messageIds :: ![MessageId],
    visibilityTimeoutOffset :: !Int32
  }
  deriving stock (Generic)

-- | Set visibility timeout to absolute timestamp (pgmq 1.10.0+)
data VisibilityTimeoutAtQuery = VisibilityTimeoutAtQuery
  { queueName :: !QueueName,
    messageId :: !MessageId,
    visibilityTime :: !UTCTime
  }
  deriving stock (Generic)

-- | Batch set visibility timeout to absolute timestamp (pgmq 1.10.0+)
data BatchVisibilityTimeoutAtQuery = BatchVisibilityTimeoutAtQuery
  { queueName :: !QueueName,
    messageIds :: ![MessageId],
    visibilityTime :: !UTCTime
  }
  deriving stock (Generic)

data BatchMessageQuery = BatchMessageQuery
  { queueName :: !QueueName,
    messageIds :: ![MessageId]
  }
  deriving stock (Generic)

data MessageQuery = MessageQuery
  { queueName :: !QueueName,
    messageId :: !MessageId
  }
  deriving stock (Generic)

data SendMessageForLater = SendMessageForLater
  { queueName :: !QueueName,
    messageBody :: !MessageBody,
    scheduledAt :: !UTCTime
  }
  deriving stock (Generic)

data BatchSendMessage = BatchSendMessage
  { queueName :: !QueueName,
    messageBodies :: ![MessageBody],
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

data BatchSendMessageForLater = BatchSendMessageForLater
  { queueName :: !QueueName,
    messageBodies :: ![MessageBody],
    scheduledAt :: !UTCTime
  }
  deriving stock (Generic)

-- | Send message with headers (pgmq 1.5.0+)
data SendMessageWithHeaders = SendMessageWithHeaders
  { queueName :: !QueueName,
    messageBody :: !MessageBody,
    messageHeaders :: !MessageHeaders,
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

-- | Send message with headers for later (pgmq 1.5.0+)
data SendMessageWithHeadersForLater = SendMessageWithHeadersForLater
  { queueName :: !QueueName,
    messageBody :: !MessageBody,
    messageHeaders :: !MessageHeaders,
    scheduledAt :: !UTCTime
  }
  deriving stock (Generic)

-- | Batch send messages with headers (pgmq 1.5.0+)
data BatchSendMessageWithHeaders = BatchSendMessageWithHeaders
  { queueName :: !QueueName,
    messageBodies :: ![MessageBody],
    messageHeaders :: ![MessageHeaders],
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

-- | Batch send messages with headers for later (pgmq 1.5.0+)
data BatchSendMessageWithHeadersForLater = BatchSendMessageWithHeadersForLater
  { queueName :: !QueueName,
    messageBodies :: ![MessageBody],
    messageHeaders :: ![MessageHeaders],
    scheduledAt :: !UTCTime
  }
  deriving stock (Generic)

-- | Parameters for reading messages from a queue
-- Note: conditional field added in pgmq 1.5.0
data ReadMessage = ReadMessage
  { queueName :: !QueueName,
    delay :: !Delay,
    -- | Number of messages to read. Nothing = 1, applied via COALESCE in the
    -- statement (a bound SQL NULL never triggers the plpgsql DEFAULT, and NULL
    -- in a LIMIT clause means LIMIT ALL).
    batchSize :: !(Maybe Int32),
    -- | Optional JSONB containment filter (pgmq 1.5.0+). A message is returned
    -- only when its body contains this object (SQL @message \@> conditional@).
    -- Nothing means no filtering.
    conditional :: !(Maybe Value)
  }
  deriving stock (Generic)

data ReadWithPollMessage = ReadWithPollMessage
  { queueName :: !QueueName,
    delay :: !Delay,
    -- | Number of messages to read. Nothing = 1, applied via COALESCE in the
    -- statement, for the same reason as 'ReadMessage'.
    batchSize :: !(Maybe Int32),
    maxPollSeconds :: !Int32,
    pollIntervalMs :: !Int32,
    -- | Optional JSONB containment filter (pgmq 1.5.0+). Nothing means no
    -- filtering.
    conditional :: !(Maybe Value)
  }
  deriving stock (Generic)

-- | Parameters for popping messages from a queue (pgmq 1.7.0+)
data PopMessage = PopMessage
  { queueName :: !QueueName,
    -- | Number of messages to pop. Nothing = 1, applied via COALESCE in the
    -- statement (a bound SQL NULL never triggers the plpgsql DEFAULT, and NULL
    -- in a LIMIT clause means LIMIT ALL — which for pop would delete the whole
    -- queue).
    qty :: !(Maybe Int32)
  }
  deriving stock (Generic)

-- | Enable queue notifications (pgmq 1.7.0+, throttling in 1.8.0+)
data EnableNotifyInsert = EnableNotifyInsert
  { queueName :: !QueueName,
    -- | Minimum ms between notifications. Nothing = 250ms, applied via COALESCE
    -- in the statement so a bound SQL NULL never reaches the NOT NULL column
    -- (a column DEFAULT does not apply to an explicitly supplied NULL).
    throttleIntervalMs :: !(Maybe Int32)
  }
  deriving stock (Generic)

data CreatePartitionedQueue = CreatePartitionedQueue
  { queueName :: !QueueName,
    partitionInterval :: !Text,
    retentionInterval :: !Text
  }
  deriving stock (Generic)

-- | Queue metrics returned by pgmq.metrics() and pgmq.metrics_all()
-- Note: queueVisibleLength added in pgmq 1.5.0
data QueueMetrics = QueueMetrics
  { queueName :: !Text,
    queueLength :: !Int64,
    newestMsgAgeSec :: !(Maybe Int32),
    oldestMsgAgeSec :: !(Maybe Int32),
    totalMessages :: !Int64,
    scrapeTime :: !UTCTime,
    -- | Count of messages available for reading (pgmq 1.5.0+)
    queueVisibleLength :: !Int64,
    -- | Planner estimate of rows in queue and archive default partitions (1.13+).
    -- 'Nothing' means unavailable on 1.12 or inapplicable for ordinary queues;
    -- it must not be interpreted as zero. Estimates can lag writes.
    defaultPartitionLength :: !(Maybe Int64)
  }
  deriving stock (Generic, Show)

-- | Parameters for FIFO grouped read (pgmq 1.8.0+)
-- Used for read_grouped, read_grouped_rr and read_grouped_head functions.
-- Note: conditional parameter was removed in pgmq 1.9.0 (commit 9e9c3dc)
data ReadGrouped = ReadGrouped
  { queueName :: !QueueName,
    visibilityTimeout :: !Int32,
    qty :: !Int32
  }
  deriving stock (Generic)

-- | Parameters for FIFO grouped read with polling (pgmq 1.8.0+)
-- Used for both read_grouped_with_poll and read_grouped_rr_with_poll functions.
-- Note: conditional parameter was removed in pgmq 1.9.0 (commit 9e9c3dc)
data ReadGroupedWithPoll = ReadGroupedWithPoll
  { queueName :: !QueueName,
    visibilityTimeout :: !Int32,
    qty :: !Int32,
    maxPollSeconds :: !Int32,
    pollIntervalMs :: !Int32
  }
  deriving stock (Generic)

-- | Bind a topic pattern to a queue (pgmq 1.11.0+)
data BindTopic = BindTopic
  { topicPattern :: !TopicPattern,
    queueName :: !QueueName
  }
  deriving stock (Generic)

-- | Unbind a topic pattern from a queue (pgmq 1.11.0+)
data UnbindTopic = UnbindTopic
  { topicPattern :: !TopicPattern,
    queueName :: !QueueName
  }
  deriving stock (Generic)

-- | Send a message via topic routing (pgmq 1.11.0+)
data SendTopic = SendTopic
  { routingKey :: !RoutingKey,
    messageBody :: !MessageBody,
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

-- | Send a message via topic routing with headers (pgmq 1.11.0+)
data SendTopicWithHeaders = SendTopicWithHeaders
  { routingKey :: !RoutingKey,
    messageBody :: !MessageBody,
    messageHeaders :: !MessageHeaders,
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

-- | Batch send messages via topic routing (pgmq 1.11.0+)
data BatchSendTopic = BatchSendTopic
  { routingKey :: !RoutingKey,
    messageBodies :: ![MessageBody],
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

-- | Batch send messages via topic routing for later (pgmq 1.11.0+)
data BatchSendTopicForLater = BatchSendTopicForLater
  { routingKey :: !RoutingKey,
    messageBodies :: ![MessageBody],
    scheduledAt :: !UTCTime
  }
  deriving stock (Generic)

-- | Batch send messages via topic routing with headers (pgmq 1.11.0+)
data BatchSendTopicWithHeaders = BatchSendTopicWithHeaders
  { routingKey :: !RoutingKey,
    messageBodies :: ![MessageBody],
    messageHeaders :: ![MessageHeaders],
    delay :: !(Maybe Delay)
  }
  deriving stock (Generic)

-- | Batch send messages via topic routing with headers for later (pgmq 1.11.0+)
data BatchSendTopicWithHeadersForLater = BatchSendTopicWithHeadersForLater
  { routingKey :: !RoutingKey,
    messageBodies :: ![MessageBody],
    messageHeaders :: ![MessageHeaders],
    scheduledAt :: !UTCTime
  }
  deriving stock (Generic)

-- | Update notification throttle interval (pgmq 1.11.0+)
data UpdateNotifyInsert = UpdateNotifyInsert
  { queueName :: !QueueName,
    throttleIntervalMs :: !Int32
  }
  deriving stock (Generic)

-- | A keyset page over a queue table (@pgmq.q_\<name\>@) or, through
-- @peekArchivedMessages@, an archive table (@pgmq.a_\<name\>@).
--
-- @unvalidatedQueueName@ is any name the server accepts, including names
-- 'Pgmq.Types.parseQueueName' rejects: an inspection surface must show what
-- exists. Pass 'Pgmq.Types.queueNameToText' when you hold a validated name.
-- The physical table is resolved by @pgmq.format_table_name@ on the server.
--
-- @afterMessageId@ is an /exclusive/ cursor: the page starts strictly after
-- it; 'Nothing' starts at the beginning. @limit@ is passed straight to SQL
-- @LIMIT@, so it must be positive (@0@ returns nothing; a negative value is a
-- server error). Pages are ordered by @msg_id@ ascending and never use
-- @OFFSET@, so deleting earlier rows does not shift the page. Each call has
-- its own snapshot: lower IDs committed or archived later require a refresh
-- from the first page. This is not a replay feed. @LIMIT@ bounds rows returned,
-- not scan work or payload bytes; partitioned layouts may scan and sort. Use
-- finite statement/lock deadlines and inspection capacity separate from workers.
data PeekMessages = PeekMessages
  { unvalidatedQueueName :: !Text,
    afterMessageId :: !(Maybe MessageId),
    limit :: !Int32
  }
  deriving stock (Generic)

-- | One message by id in a queue table or, through @lookupArchivedMessage@, an
-- archive table. The name rules are those of t'PeekMessages'.
data LookupMessage = LookupMessage
  { unvalidatedQueueName :: !Text,
    messageId :: !MessageId
  }
  deriving stock (Generic)

-- JSON wire encodings ---------------------------------------------------------

-- | Wire keys are the @pgmq.metrics_result@ columns. Two of them are easy to
-- confuse and must not be: @queue_length@ is the total number of rows in the
-- queue table, and @queue_visible_length@ is the subset whose visibility
-- timeout has expired and are therefore available to a reader (see
-- @docs/design/002-queue-visible-length.md@). @default_partition_length@ is
-- @null@ when the metric is unavailable (PGMQ 1.12) or inapplicable (a
-- non-partitioned queue); @null@ never means zero. Encode-only. The encoding is
-- a published compatibility surface; see
-- @docs/design/020-json-wire-encodings.md@.
--
-- The pattern is positional because @queueName@ is a duplicate record field in
-- this module and its selector is ambiguous under GHC 9.12.
queueMetricsFields :: (KeyValue e kv) => QueueMetrics -> [kv]
queueMetricsFields (QueueMetrics qn len newest oldest total scrape visible dpl) =
  [ "queue_name" .= qn,
    "queue_length" .= len,
    "newest_msg_age_sec" .= newest,
    "oldest_msg_age_sec" .= oldest,
    "total_messages" .= total,
    "scrape_time" .= scrape,
    "queue_visible_length" .= visible,
    "default_partition_length" .= dpl
  ]

-- | JSON wire contract: @docs/design/020-json-wire-encodings.md@.
instance ToJSON QueueMetrics where
  toJSON = object . queueMetricsFields
  toEncoding = pairs . mconcat . queueMetricsFields