HistoryRetention.hs

-- | Durable replay-history retention and transaction-scoped stream guards.
--
-- A lease protects the retained global event set from Kiroku hard delete and
-- authorized direct SQL deletion until its database-derived expiry. Transaction
-- combinators persist durable evidence but cannot emit process-local observability
-- events from inside an opaque caller-owned transaction.
--
-- Lease duration is one second through one hour. Expiry is passive: a crashed
-- owner blocks destructive work only until PostgreSQL time reaches @expiresAt@.
-- The owner is an accidental-mutation guard, not an authorization credential;
-- database roles and grants remain the security boundary. Direct destructive SQL
-- receives SQLSTATE @KR001@ while any lease is active.
--
-- When one transaction needs both a lease operation and a stream-history guard,
-- perform the lease operation first. This preserves Kiroku's coordinator-before-
-- stream-row lock order.
module Kiroku.Store.HistoryRetention
  ( module Kiroku.Store.HistoryRetention.Types,
    acquireHistoryRetentionLeaseTx,
    renewHistoryRetentionLeaseTx,
    releaseHistoryRetentionLeaseTx,
    historyRetentionLeaseInventoryTx,
    pruneHistoryRetentionLeasesTx,
    acquireHistoryRetentionLease,
    renewHistoryRetentionLease,
    releaseHistoryRetentionLease,
    historyRetentionLeaseInventory,
    pruneHistoryRetentionLeases,
    lockStreamHistoryForReplayTx,
    readStreamForwardTx,
  )
where

import Data.Int (Int32)
import Data.Time.Clock (UTCTime)
import Data.Vector (Vector)
import Effectful (Eff, (:>))
import Effectful.Dispatch.Dynamic (send)
import Hasql.Transaction qualified as Tx
import Kiroku.Store.Effect (Store (..))
import Kiroku.Store.HistoryRetention.Internal qualified as Internal
import Kiroku.Store.HistoryRetention.SQL qualified as HistoryRetentionSQL
import Kiroku.Store.HistoryRetention.Types
import Kiroku.Store.SQL qualified as SQL
import Kiroku.Store.Types (RecordedEvent, StreamInfo, StreamName (..), StreamVersion (..))

-- | Acquire a new lease and atomically capture the authoritative @$all@
-- frontier as the inclusive @protectedThrough@ rebuild ceiling. PostgreSQL
-- supplies every timestamp. Rollback leaves no lease row.
acquireHistoryRetentionLeaseTx :: HistoryRetentionLeaseRequest -> Tx.Transaction HistoryRetentionLease
acquireHistoryRetentionLeaseTx = Internal.acquireHistoryRetentionLeaseTx

-- | Renew a still-active lease without shortening its current expiry.
-- Unknown, wrong-owner, expired, and released leases return distinct typed
-- errors and are never resurrected.
renewHistoryRetentionLeaseTx ::
  HistoryRetentionLeaseHandle ->
  HistoryRetentionLeaseDuration ->
  Tx.Transaction (Either HistoryRetentionRenewalError HistoryRetentionLease)
renewHistoryRetentionLeaseTx = Internal.renewHistoryRetentionLeaseTx

-- | Release a live lease. Repetition is typed and does not rewrite the row.
releaseHistoryRetentionLeaseTx ::
  HistoryRetentionLeaseHandle ->
  Tx.Transaction HistoryRetentionReleaseResult
releaseHistoryRetentionLeaseTx = Internal.releaseHistoryRetentionLeaseTx

-- | Read a deterministic, database-time-derived, bounded inventory. Active,
-- expired, and released state is derived at statement time; no expiry worker is
-- required.
historyRetentionLeaseInventoryTx ::
  HistoryRetentionInventoryQuery ->
  Tx.Transaction (Vector HistoryRetentionLease)
historyRetentionLeaseInventoryTx = Internal.historyRetentionLeaseInventoryTx

-- | Remove only expired or released rows strictly older than the supplied cutoff.
pruneHistoryRetentionLeasesTx :: UTCTime -> Tx.Transaction HistoryRetentionPruneResult
pruneHistoryRetentionLeasesTx = Internal.pruneHistoryRetentionLeasesTx

-- | Acquire a lease through the mockable 'Store' effect. The production
-- interpreter emits the committed acquisition event after the transaction ends.
acquireHistoryRetentionLease :: (Store :> es) => HistoryRetentionLeaseRequest -> Eff es HistoryRetentionLease
acquireHistoryRetentionLease request = send (AcquireHistoryRetentionLease request)

-- | Renew a lease through the mockable 'Store' effect and emit only a committed success.
renewHistoryRetentionLease ::
  (Store :> es) =>
  HistoryRetentionLeaseHandle ->
  HistoryRetentionLeaseDuration ->
  Eff es (Either HistoryRetentionRenewalError HistoryRetentionLease)
renewHistoryRetentionLease handle duration = send (RenewHistoryRetentionLease handle duration)

-- | Release a lease through the mockable 'Store' effect. Repeated release emits no false transition.
releaseHistoryRetentionLease ::
  (Store :> es) =>
  HistoryRetentionLeaseHandle ->
  Eff es HistoryRetentionReleaseResult
releaseHistoryRetentionLease handle = send (ReleaseHistoryRetentionLease handle)

-- | Read the bounded durable inventory through the mockable 'Store' effect.
historyRetentionLeaseInventory ::
  (Store :> es) =>
  HistoryRetentionInventoryQuery ->
  Eff es (Vector HistoryRetentionLease)
historyRetentionLeaseInventory query = send (GetHistoryRetentionLeaseInventory query)

-- | Prune terminal rows through the mockable 'Store' effect.
pruneHistoryRetentionLeases ::
  (Store :> es) =>
  UTCTime ->
  Eff es HistoryRetentionPruneResult
pruneHistoryRetentionLeases cutoff = send (PruneHistoryRetentionLeases cutoff)

-- | Lock one application's stream row in share mode until the surrounding
-- transaction ends. The returned metadata includes soft-delete and logical
-- truncate state so a repair can reject incomplete history before mutating its
-- target. @$all@ is reserved.
--
-- When composing this guard with a lease operation, acquire or renew the lease
-- first and take this guard second.
lockStreamHistoryForReplayTx ::
  StreamName ->
  Tx.Transaction (Either StreamHistoryUnavailable StreamInfo)
lockStreamHistoryForReplayTx stream@(StreamName name)
  | name == "$all" = pure (Left (StreamHistoryReserved stream))
  | otherwise = do
      locked <- Tx.statement name HistoryRetentionSQL.lockStreamHistoryStmt
      pure $ maybe (Left (StreamHistoryNotFound stream)) Right locked

-- | Read a page inside the caller's transaction using the production
-- exclusive-lower cursor and ascending-order statement. This function deliberately
-- does not run 'Kiroku.Store.Settings.decodeHook': 'Tx.Transaction' has no 'IO'.
-- The caller must take 'lockStreamHistoryForReplayTx' earlier in the same
-- transaction when stable one-stream history is required.
readStreamForwardTx ::
  StreamName ->
  StreamVersion ->
  Int32 ->
  Tx.Transaction (Vector RecordedEvent)
readStreamForwardTx (StreamName name) (StreamVersion cursor) limit =
  Tx.statement (name, cursor, limit) SQL.readStreamForwardStmt