SQL.hs

{-# LANGUAGE MultilineStrings #-}

module Kiroku.Store.Subscription.DeadLetter.SQL
  ( listSubscriptionDeadLettersSession,
    listSubscriptionDeadLettersStmt,
    listSubscriptionDeadLettersFromStartStmt,
    listSubscriptionMemberDeadLettersStmt,
    listSubscriptionMemberDeadLettersFromStartStmt,
  )
where

import Contravariant.Extras (contrazip2, contrazip3, contrazip4, contrazip5)
import Control.Lens ((^.))
import Data.Generics.Labels ()
import Data.Int (Int32, Int64)
import Data.Text (Text)
import Data.Vector (Vector)
import Data.Vector qualified as V
import Hasql.Decoders qualified as D
import Hasql.Encoders qualified as E
import Hasql.Session (Session)
import Hasql.Session qualified as Session
import Hasql.Statement (Statement, preparable)
import Kiroku.Store.Subscription.Types
import Kiroku.Store.Types (EventId (..), GlobalPosition (..))

listSubscriptionDeadLettersFromStartStmt :: Statement (Text, Int32) (Vector SubscriptionDeadLetter)
listSubscriptionDeadLettersFromStartStmt =
  preparable
    """
    WITH RECURSIVE members(member) AS (
        SELECT (SELECT consumer_group_member FROM dead_letters
                WHERE subscription_name = $1 ORDER BY consumer_group_member LIMIT 1)
        UNION ALL
        SELECT (SELECT consumer_group_member FROM dead_letters
                WHERE subscription_name = $1 AND consumer_group_member > m.member
                ORDER BY consumer_group_member LIMIT 1)
        FROM members m WHERE m.member IS NOT NULL
    )
    SELECT d.dead_letter_id, d.subscription_name, d.consumer_group_member, d.global_position, d.event_id, d.reason, d.reason_summary, d.attempt_count, d.created_at
    FROM members m
    CROSS JOIN LATERAL (
    SELECT dead_letter_id, subscription_name, consumer_group_member, global_position, event_id, reason, reason_summary, attempt_count, created_at
    FROM dead_letters
    WHERE subscription_name = $1 AND consumer_group_member = m.member

    ORDER BY global_position DESC, dead_letter_id DESC LIMIT $2
    ) d
    WHERE m.member IS NOT NULL
    ORDER BY d.global_position DESC, d.dead_letter_id DESC LIMIT $2
    """
    (contrazip2 text int4)
    (D.rowVector deadLetterRow)

listSubscriptionDeadLettersStmt :: Statement (Text, Int64, Int64, Int32) (Vector SubscriptionDeadLetter)
listSubscriptionDeadLettersStmt =
  preparable
    """
    WITH RECURSIVE members(member) AS (
        SELECT (SELECT consumer_group_member FROM dead_letters
                WHERE subscription_name = $1 ORDER BY consumer_group_member LIMIT 1)
        UNION ALL
        SELECT (SELECT consumer_group_member FROM dead_letters
                WHERE subscription_name = $1 AND consumer_group_member > m.member
                ORDER BY consumer_group_member LIMIT 1)
        FROM members m WHERE m.member IS NOT NULL
    )
    SELECT d.dead_letter_id, d.subscription_name, d.consumer_group_member, d.global_position, d.event_id, d.reason, d.reason_summary, d.attempt_count, d.created_at
    FROM members m
    CROSS JOIN LATERAL (
    SELECT dead_letter_id, subscription_name, consumer_group_member, global_position, event_id, reason, reason_summary, attempt_count, created_at
    FROM dead_letters
    WHERE subscription_name = $1 AND consumer_group_member = m.member
    AND (global_position, dead_letter_id) < ($2, $3)
    ORDER BY global_position DESC, dead_letter_id DESC LIMIT $4
    ) d
    WHERE m.member IS NOT NULL
    ORDER BY d.global_position DESC, d.dead_letter_id DESC LIMIT $4
    """
    (contrazip4 text int8 int8 int4)
    (D.rowVector deadLetterRow)

listSubscriptionMemberDeadLettersFromStartStmt :: Statement (Text, Int32, Int32) (Vector SubscriptionDeadLetter)
listSubscriptionMemberDeadLettersFromStartStmt =
  preparable
    """
    SELECT dead_letter_id, subscription_name, consumer_group_member, global_position, event_id, reason, reason_summary, attempt_count, created_at
    FROM dead_letters
    WHERE subscription_name = $1 AND consumer_group_member = $2

    ORDER BY global_position DESC, dead_letter_id DESC LIMIT $3
    """
    (contrazip3 text int4 int4)
    (D.rowVector deadLetterRow)

listSubscriptionMemberDeadLettersStmt :: Statement (Text, Int32, Int64, Int64, Int32) (Vector SubscriptionDeadLetter)
listSubscriptionMemberDeadLettersStmt =
  preparable
    """
    SELECT dead_letter_id, subscription_name, consumer_group_member, global_position, event_id, reason, reason_summary, attempt_count, created_at
    FROM dead_letters
    WHERE subscription_name = $1 AND consumer_group_member = $2
    AND (global_position, dead_letter_id) < ($3, $4)
    ORDER BY global_position DESC, dead_letter_id DESC LIMIT $5
    """
    (contrazip5 text int4 int8 int8 int4)
    (D.rowVector deadLetterRow)

text :: E.Params Text
text = E.param (E.nonNullable E.text)

int4 :: E.Params Int32
int4 = E.param (E.nonNullable E.int4)

int8 :: E.Params Int64
int8 = E.param (E.nonNullable E.int8)

deadLetterRow :: D.Row SubscriptionDeadLetter
deadLetterRow =
  SubscriptionDeadLetter
    <$> D.column (D.nonNullable D.int8)
    <*> (SubscriptionName <$> D.column (D.nonNullable D.text))
    <*> D.column (D.nonNullable D.int4)
    <*> (GlobalPosition <$> D.column (D.nonNullable D.int8))
    <*> (EventId <$> D.column (D.nonNullable D.uuid))
    <*> D.column (D.nonNullable D.jsonb)
    <*> D.column (D.nonNullable D.text)
    <*> D.column (D.nonNullable D.int4)
    <*> D.column (D.nonNullable D.timestamptz)

listSubscriptionDeadLettersSession :: SubscriptionDeadLetterQuery -> Session SubscriptionDeadLetterPage
listSubscriptionDeadLettersSession query = do
  let SubscriptionName name = query ^. #subscriptionName
      pageSize = subscriptionDeadLetterLimitValue (query ^. #limit)
      fetch = pageSize + 1
  rows <- case (query ^. #consumerGroupMember, query ^. #after) of
    (Nothing, Nothing) -> Session.statement (name, fetch) listSubscriptionDeadLettersFromStartStmt
    (Just selectedMember, Nothing) -> Session.statement (name, selectedMember, fetch) listSubscriptionMemberDeadLettersFromStartStmt
    (Nothing, Just (SubscriptionDeadLetterCursor (GlobalPosition position) ident)) ->
      Session.statement (name, position, ident, fetch) listSubscriptionDeadLettersStmt
    (Just selectedMember, Just (SubscriptionDeadLetterCursor (GlobalPosition position) ident)) ->
      Session.statement (name, selectedMember, position, ident, fetch) listSubscriptionMemberDeadLettersStmt
  let page = V.take (fromIntegral pageSize) rows
      cursor = if V.length rows > fromIntegral pageSize then Just (subscriptionDeadLetterCursor (V.last page)) else Nothing
  pure (SubscriptionDeadLetterPage page cursor)