SQL.hs

{-# LANGUAGE MultilineStrings #-}

module Kiroku.Store.Subscription.CheckpointInventory.SQL
  ( getSubscriptionCheckpointInventoryStmt,
  )
where

import Data.Int (Int32, Int64)
import Data.Text (Text)
import Data.Time.Clock (UTCTime)
import Data.Vector (Vector)
import Data.Vector qualified as V
import Hasql.Decoders qualified as D
import Hasql.Encoders qualified as E
import Hasql.Statement (Statement, preparable)
import Hasql.Statement qualified as Statement
import Kiroku.Store.Subscription.Types
  ( SubscriptionCheckpoint (..),
    SubscriptionCheckpointInventory (..),
    SubscriptionName (..),
  )
import Kiroku.Store.Types (GlobalPosition (..))

data InventoryRow = InventoryRow
  { rowStorePosition :: !Int64,
    rowSubscriptionName :: !(Maybe Text),
    rowConsumerGroupMember :: !(Maybe Int32),
    rowCheckpointPosition :: !(Maybe Int64),
    rowCheckpointUpdatedAt :: !(Maybe UTCTime)
  }

getSubscriptionCheckpointInventoryStmt :: Statement () SubscriptionCheckpointInventory
getSubscriptionCheckpointInventoryStmt =
  Statement.refineResult finalizeInventory $
    preparable
      """
      SELECT store_head.stream_version,
             checkpoint.subscription_name,
             checkpoint.consumer_group_member,
             checkpoint.last_seen,
             checkpoint.updated_at
      FROM streams AS store_head
      LEFT JOIN subscriptions AS checkpoint ON TRUE
      WHERE store_head.stream_id = 0
      ORDER BY checkpoint.subscription_name ASC,
               checkpoint.consumer_group_member ASC
      """
      E.noParams
      (D.rowVector inventoryRow)

inventoryRow :: D.Row InventoryRow
inventoryRow =
  InventoryRow
    <$> D.column (D.nonNullable D.int8)
    <*> D.column (D.nullable D.text)
    <*> D.column (D.nullable D.int4)
    <*> D.column (D.nullable D.int8)
    <*> D.column (D.nullable D.timestamptz)

finalizeInventory :: Vector InventoryRow -> Either Text SubscriptionCheckpointInventory
finalizeInventory rows = case V.uncons rows of
  Nothing -> Left "subscription checkpoint inventory: missing $all stream row"
  Just (firstRow, remainingRows) ->
    let capturedPosition = rowStorePosition firstRow
     in case checkpointColumns firstRow of
          EmptyCheckpoint
            | V.null remainingRows ->
                Right $
                  SubscriptionCheckpointInventory
                    (GlobalPosition capturedPosition)
                    V.empty
            | otherwise ->
                Left "subscription checkpoint inventory: empty checkpoint row was not the only result"
          PartialCheckpoint ->
            Left "subscription checkpoint inventory: partially null checkpoint row"
          CompleteCheckpoint ->
            SubscriptionCheckpointInventory (GlobalPosition capturedPosition)
              <$> V.mapM (decodeCheckpoint capturedPosition) rows

data CheckpointColumns
  = EmptyCheckpoint
  | PartialCheckpoint
  | CompleteCheckpoint

checkpointColumns :: InventoryRow -> CheckpointColumns
checkpointColumns row =
  case ( rowSubscriptionName row,
         rowConsumerGroupMember row,
         rowCheckpointPosition row,
         rowCheckpointUpdatedAt row
       ) of
    (Nothing, Nothing, Nothing, Nothing) -> EmptyCheckpoint
    (Just _, Just _, Just _, Just _) -> CompleteCheckpoint
    _ -> PartialCheckpoint

decodeCheckpoint :: Int64 -> InventoryRow -> Either Text SubscriptionCheckpoint
decodeCheckpoint capturedPosition row
  | rowStorePosition row /= capturedPosition =
      Left "subscription checkpoint inventory: inconsistent repeated store position"
  | otherwise =
      case ( rowSubscriptionName row,
             rowConsumerGroupMember row,
             rowCheckpointPosition row,
             rowCheckpointUpdatedAt row
           ) of
        (Just name, Just member, Just position, Just updatedAt) ->
          Right $
            SubscriptionCheckpoint
              (SubscriptionName name)
              member
              (GlobalPosition position)
              updatedAt
        (Nothing, Nothing, Nothing, Nothing) ->
          Left "subscription checkpoint inventory: unexpected empty checkpoint row"
        _ -> Left "subscription checkpoint inventory: partially null checkpoint row"