Transaction.hs

module Kafka.Effectful.Producer.Transaction
  ( commitOffsetMessageTransaction,
  )
where

import Effectful (Eff, (:>))
import Kafka.Consumer.Types (ConsumerRecord)
import Kafka.Effectful.Consumer.Effect (KafkaConsumer, askConsumerHandle)
import Kafka.Effectful.Producer.Effect (KafkaProducer, sendOffsetsToTransaction)
import Kafka.Transaction (TxError)
import Kafka.Types (Timeout)

-- | Commit the offset of a single 'ConsumerRecord' inside the current
-- transaction of the surrounding 'KafkaProducer'.
--
-- This is the cross-effect counterpart to the consumer's
-- @commitOffsetMessage@ — the record's offset is not stored via the
-- consumer's offset-commit path but instead shipped as part of the
-- producer's open transaction, which preserves the exactly-once
-- semantics required by Scenario 5 of @hw-kafka-client@'s producer
-- best practices.
--
-- Returns @Nothing@ on success or @Just TxError@ on failure. The
-- caller must dispatch on the three 'TxError' discriminators
-- (@kafkaErrorTxnRequiresAbort@ first, then @kafkaErrorIsRetriable@,
-- then @kafkaErrorIsFatal@) to decide whether to abort the
-- transaction, retry the commit, or crash.
--
-- The helper picks up the consumer handle automatically via
-- @askConsumerHandle@, so callers only need to have both 'KafkaProducer'
-- and 'KafkaConsumer' in the effect stack.
--
-- @since 0.2.0.0
commitOffsetMessageTransaction ::
  (KafkaProducer :> es, KafkaConsumer :> es) =>
  ConsumerRecord k v ->
  Timeout ->
  Eff es (Maybe TxError)
commitOffsetMessageTransaction record timeout = do
  consumer <- askConsumerHandle
  sendOffsetsToTransaction consumer record timeout