{-# LANGUAGE DataKinds       #-}
{-# LANGUAGE GADTs           #-}
{-# LANGUAGE KindSignatures  #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE TypeFamilies    #-}
--------------------------------------------------------------------------------
-- |
-- Module : Database.EventStore.Internal.Operation.Transaction
-- Copyright : (C) 2015 Yorick Laupa
-- License : (see the file LICENSE)
--
-- Maintainer : Yorick Laupa <yo.eight@gmail.com>
-- Stability : provisional
-- Portability : non-portable
--
--------------------------------------------------------------------------------
module Database.EventStore.Internal.Operation.Transaction
    ( transactionStart
    , transactionWrite
    , transactionCommit
    ) where

--------------------------------------------------------------------------------
import Data.Int
import Data.Maybe

--------------------------------------------------------------------------------
import Data.ProtocolBuffers

--------------------------------------------------------------------------------
import Database.EventStore.Internal.Command
import Database.EventStore.Internal.Control (publishWith)
import Database.EventStore.Internal.Communication (Transmit(..))
import Database.EventStore.Internal.Exec (Exec)
import Database.EventStore.Internal.Operation (OpResult(..))
import Database.EventStore.Internal.Operation
import Database.EventStore.Internal.Operation.Transaction.Message
import Database.EventStore.Internal.Operation.Write.Common
import Database.EventStore.Internal.Prelude
import Database.EventStore.Internal.Settings
import Database.EventStore.Internal.Stream
import Database.EventStore.Internal.Types

--------------------------------------------------------------------------------
-- | Start transaction operation.
transactionStart
  :: Settings
  -> Exec
  -> Text
  -> ExpectedVersion
  -> Maybe Credentials
  -> IO (Async Int64)
transactionStart :: Settings
-> Exec
-> Text
-> ExpectedVersion
-> Maybe Credentials
-> IO (Async Int64)
transactionStart Settings{Bool
Maybe Text
Maybe TLSSettings
Maybe Credentials
LogType
NominalDiffTime
LoggerFilter
MonitoringBackend
Retry
s_defaultUserCredentials :: Settings -> Maybe Credentials
s_defaultConnectionName :: Settings -> Maybe Text
s_monitoring :: Settings -> MonitoringBackend
s_operationRetry :: Settings -> Retry
s_operationTimeout :: Settings -> NominalDiffTime
s_loggerDetailed :: Settings -> Bool
s_loggerFilter :: Settings -> LoggerFilter
s_loggerType :: Settings -> LogType
s_ssl :: Settings -> Maybe TLSSettings
s_reconnect_delay :: Settings -> NominalDiffTime
s_retry :: Settings -> Retry
s_requireMaster :: Settings -> Bool
s_heartbeatTimeout :: Settings -> NominalDiffTime
s_heartbeatInterval :: Settings -> NominalDiffTime
s_defaultUserCredentials :: Maybe Credentials
s_defaultConnectionName :: Maybe Text
s_monitoring :: MonitoringBackend
s_operationRetry :: Retry
s_operationTimeout :: NominalDiffTime
s_loggerDetailed :: Bool
s_loggerFilter :: LoggerFilter
s_loggerType :: LogType
s_ssl :: Maybe TLSSettings
s_reconnect_delay :: NominalDiffTime
s_retry :: Retry
s_requireMaster :: Bool
s_heartbeatTimeout :: NominalDiffTime
s_heartbeatInterval :: NominalDiffTime
..} Exec
exec Text
stream ExpectedVersion
exp_v Maybe Credentials
cred
  = do Mailbox
m <- forall (m :: * -> *). MonadBase IO m => m Mailbox
mailboxNew
       forall (m :: * -> *) a.
MonadBaseControl IO m =>
m a -> m (Async (StM m a))
async forall a b. (a -> b) -> a -> b
$
         do let req :: Start
req = Text -> Int64 -> Bool -> Start
newStart Text
stream (ExpectedVersion -> Int64
expVersionInt64 ExpectedVersion
exp_v) Bool
s_requireMaster
            Package
pkg <- forall msg (m :: * -> *).
(Encode msg, MonadIO m) =>
Command -> Maybe Credentials -> msg -> m Package
createPkg Command
transactionStartCmd Maybe Credentials
cred Start
req

            forall (m :: * -> *) a. Monad m => m (Loop a) -> m a
keepLooping forall a b. (a -> b) -> a -> b
$
              do forall p a (m :: * -> *).
(Pub p, Typeable a, MonadIO m) =>
p -> a -> m ()
publishWith Exec
exec (Mailbox -> Lifetime -> Package -> Transmit
Transmit Mailbox
m Lifetime
OneTime Package
pkg)
                 Either OperationError Started
outcome <- forall (m :: * -> *) resp.
(MonadBase IO m, Decode resp) =>
Mailbox -> m (Either OperationError resp)
mailboxReadDecoded Mailbox
m
                 case Either OperationError Started
outcome of
                   Left OperationError
e
                     -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw OperationError
e
                   Right Started
resp
                     -> let tid :: FieldType (Required 1 (Value Int64))
tid = forall a. HasField a => a -> FieldType a
getField forall a b. (a -> b) -> a -> b
$ Started -> Required 1 (Value Int64)
_transId Started
resp
                            r :: FieldType (Required 2 (Enumeration OpResult))
r   = forall a. HasField a => a -> FieldType a
getField forall a b. (a -> b) -> a -> b
$ Started -> Required 2 (Enumeration OpResult)
_result Started
resp in
                        case FieldType (Required 2 (Enumeration OpResult))
r of
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_PREPARE_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_FORWARD_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_COMMIT_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_WRONG_EXPECTED_VERSION -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ Text -> ExpectedVersion -> OperationError
WrongExpectedVersion Text
stream ExpectedVersion
exp_v
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_STREAM_DELETED -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ StreamName -> OperationError
StreamDeleted forall a b. (a -> b) -> a -> b
$ Text -> StreamName
StreamName Text
stream
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_INVALID_TRANSACTION -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw OperationError
InvalidTransaction
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_ACCESS_DENIED -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ forall t. StreamId t -> OperationError
AccessDenied forall a b. (a -> b) -> a -> b
$ Text -> StreamName
StreamName Text
stream
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_SUCCESS -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a b. (a -> b) -> a -> b
$ forall a. a -> Loop a
Break FieldType (Required 1 (Value Int64))
tid

--------------------------------------------------------------------------------
-- | Transactional write operation.
transactionWrite
  :: Settings
  -> Exec
  -> Text
  -> ExpectedVersion
  -> Int64
  -> [Event]
  -> Maybe Credentials
  -> IO (Async ())
transactionWrite :: Settings
-> Exec
-> Text
-> ExpectedVersion
-> Int64
-> [Event]
-> Maybe Credentials
-> IO (Async ())
transactionWrite Settings{Bool
Maybe Text
Maybe TLSSettings
Maybe Credentials
LogType
NominalDiffTime
LoggerFilter
MonitoringBackend
Retry
s_defaultUserCredentials :: Maybe Credentials
s_defaultConnectionName :: Maybe Text
s_monitoring :: MonitoringBackend
s_operationRetry :: Retry
s_operationTimeout :: NominalDiffTime
s_loggerDetailed :: Bool
s_loggerFilter :: LoggerFilter
s_loggerType :: LogType
s_ssl :: Maybe TLSSettings
s_reconnect_delay :: NominalDiffTime
s_retry :: Retry
s_requireMaster :: Bool
s_heartbeatTimeout :: NominalDiffTime
s_heartbeatInterval :: NominalDiffTime
s_defaultUserCredentials :: Settings -> Maybe Credentials
s_defaultConnectionName :: Settings -> Maybe Text
s_monitoring :: Settings -> MonitoringBackend
s_operationRetry :: Settings -> Retry
s_operationTimeout :: Settings -> NominalDiffTime
s_loggerDetailed :: Settings -> Bool
s_loggerFilter :: Settings -> LoggerFilter
s_loggerType :: Settings -> LogType
s_ssl :: Settings -> Maybe TLSSettings
s_reconnect_delay :: Settings -> NominalDiffTime
s_retry :: Settings -> Retry
s_requireMaster :: Settings -> Bool
s_heartbeatTimeout :: Settings -> NominalDiffTime
s_heartbeatInterval :: Settings -> NominalDiffTime
..} Exec
exec Text
stream ExpectedVersion
exp_v Int64
trans_id [Event]
evts Maybe Credentials
cred
  = do Mailbox
m <- forall (m :: * -> *). MonadBase IO m => m Mailbox
mailboxNew
       forall (m :: * -> *) a.
MonadBaseControl IO m =>
m a -> m (Async (StM m a))
async forall a b. (a -> b) -> a -> b
$
         do [NewEvent]
nevts <- forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
traverse Event -> IO NewEvent
eventToNewEventIO [Event]
evts
            let req :: Write
req = Int64 -> [NewEvent] -> Bool -> Write
newWrite Int64
trans_id [NewEvent]
nevts Bool
s_requireMaster
            Package
pkg <- forall msg (m :: * -> *).
(Encode msg, MonadIO m) =>
Command -> Maybe Credentials -> msg -> m Package
createPkg Command
transactionWriteCmd Maybe Credentials
cred Write
req
            forall (m :: * -> *) a. Monad m => m (Loop a) -> m a
keepLooping forall a b. (a -> b) -> a -> b
$
              do forall p a (m :: * -> *).
(Pub p, Typeable a, MonadIO m) =>
p -> a -> m ()
publishWith Exec
exec (Mailbox -> Lifetime -> Package -> Transmit
Transmit Mailbox
m Lifetime
OneTime Package
pkg)
                 Either OperationError Written
outcome <- forall (m :: * -> *) resp.
(MonadBase IO m, Decode resp) =>
Mailbox -> m (Either OperationError resp)
mailboxReadDecoded Mailbox
m
                 case Either OperationError Written
outcome of
                   Left OperationError
e
                     -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw OperationError
e
                   Right Written
resp
                     -> let r :: FieldType (Required 2 (Enumeration OpResult))
r = forall a. HasField a => a -> FieldType a
getField forall a b. (a -> b) -> a -> b
$ Written -> Required 2 (Enumeration OpResult)
_wwResult Written
resp in
                        case FieldType (Required 2 (Enumeration OpResult))
r of
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_PREPARE_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_FORWARD_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_COMMIT_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_WRONG_EXPECTED_VERSION -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ Text -> ExpectedVersion -> OperationError
WrongExpectedVersion Text
stream ExpectedVersion
exp_v
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_STREAM_DELETED -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ StreamName -> OperationError
StreamDeleted forall a b. (a -> b) -> a -> b
$ Text -> StreamName
StreamName Text
stream
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_INVALID_TRANSACTION -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw OperationError
InvalidTransaction
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_ACCESS_DENIED -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ forall t. StreamId t -> OperationError
AccessDenied forall a b. (a -> b) -> a -> b
$ Text -> StreamName
StreamName Text
stream
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_SUCCESS -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a b. (a -> b) -> a -> b
$ forall a. a -> Loop a
Break ()

--------------------------------------------------------------------------------
-- | Transactional commit operation.
transactionCommit
  :: Settings
  -> Exec
  -> Text
  -> ExpectedVersion
  -> Int64
  -> Maybe Credentials
  -> IO (Async WriteResult)
transactionCommit :: Settings
-> Exec
-> Text
-> ExpectedVersion
-> Int64
-> Maybe Credentials
-> IO (Async WriteResult)
transactionCommit Settings{Bool
Maybe Text
Maybe TLSSettings
Maybe Credentials
LogType
NominalDiffTime
LoggerFilter
MonitoringBackend
Retry
s_defaultUserCredentials :: Maybe Credentials
s_defaultConnectionName :: Maybe Text
s_monitoring :: MonitoringBackend
s_operationRetry :: Retry
s_operationTimeout :: NominalDiffTime
s_loggerDetailed :: Bool
s_loggerFilter :: LoggerFilter
s_loggerType :: LogType
s_ssl :: Maybe TLSSettings
s_reconnect_delay :: NominalDiffTime
s_retry :: Retry
s_requireMaster :: Bool
s_heartbeatTimeout :: NominalDiffTime
s_heartbeatInterval :: NominalDiffTime
s_defaultUserCredentials :: Settings -> Maybe Credentials
s_defaultConnectionName :: Settings -> Maybe Text
s_monitoring :: Settings -> MonitoringBackend
s_operationRetry :: Settings -> Retry
s_operationTimeout :: Settings -> NominalDiffTime
s_loggerDetailed :: Settings -> Bool
s_loggerFilter :: Settings -> LoggerFilter
s_loggerType :: Settings -> LogType
s_ssl :: Settings -> Maybe TLSSettings
s_reconnect_delay :: Settings -> NominalDiffTime
s_retry :: Settings -> Retry
s_requireMaster :: Settings -> Bool
s_heartbeatTimeout :: Settings -> NominalDiffTime
s_heartbeatInterval :: Settings -> NominalDiffTime
..} Exec
exec Text
stream ExpectedVersion
exp_v Int64
trans_id Maybe Credentials
cred
  = do Mailbox
m <- forall (m :: * -> *). MonadBase IO m => m Mailbox
mailboxNew
       forall (m :: * -> *) a.
MonadBaseControl IO m =>
m a -> m (Async (StM m a))
async forall a b. (a -> b) -> a -> b
$
         do let req :: Commit
req = Int64 -> Bool -> Commit
newCommit Int64
trans_id Bool
s_requireMaster
            Package
pkg <- forall msg (m :: * -> *).
(Encode msg, MonadIO m) =>
Command -> Maybe Credentials -> msg -> m Package
createPkg Command
transactionCommitCmd Maybe Credentials
cred Commit
req
            forall (m :: * -> *) a. Monad m => m (Loop a) -> m a
keepLooping forall a b. (a -> b) -> a -> b
$
              do forall p a (m :: * -> *).
(Pub p, Typeable a, MonadIO m) =>
p -> a -> m ()
publishWith Exec
exec (Mailbox -> Lifetime -> Package -> Transmit
Transmit Mailbox
m Lifetime
OneTime Package
pkg)
                 Either OperationError Committed
outcome <- forall (m :: * -> *) resp.
(MonadBase IO m, Decode resp) =>
Mailbox -> m (Either OperationError resp)
mailboxReadDecoded Mailbox
m
                 case Either OperationError Committed
outcome of
                   Left OperationError
e
                     -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw OperationError
e
                   Right Committed
resp
                     -> let r :: FieldType (Required 2 (Enumeration OpResult))
r = forall a. HasField a => a -> FieldType a
getField forall a b. (a -> b) -> a -> b
$ Committed -> Required 2 (Enumeration OpResult)
_ccResult Committed
resp
                            com_pos :: FieldType (Optional 7 (Value Int64))
com_pos = forall a. HasField a => a -> FieldType a
getField forall a b. (a -> b) -> a -> b
$ Committed -> Optional 7 (Value Int64)
_commitPosition Committed
resp
                            pre_pos :: FieldType (Optional 6 (Value Int64))
pre_pos = forall a. HasField a => a -> FieldType a
getField forall a b. (a -> b) -> a -> b
$ Committed -> Optional 6 (Value Int64)
_preparePosition Committed
resp
                            lst_num :: FieldType (Required 5 (Value Int64))
lst_num = forall a. HasField a => a -> FieldType a
getField forall a b. (a -> b) -> a -> b
$ Committed -> Required 5 (Value Int64)
_lastNumber Committed
resp
                            p_int :: Int64
p_int = forall a. a -> Maybe a -> a
fromMaybe (-Int64
1) FieldType (Optional 6 (Value Int64))
pre_pos
                            c_int :: Int64
c_int = forall a. a -> Maybe a -> a
fromMaybe (-Int64
1) FieldType (Optional 7 (Value Int64))
com_pos
                            pos :: Position
pos = Int64 -> Int64 -> Position
Position Int64
c_int Int64
p_int
                            res :: WriteResult
res = Int64 -> Position -> WriteResult
WriteResult FieldType (Required 5 (Value Int64))
lst_num Position
pos in
                        case FieldType (Required 2 (Enumeration OpResult))
r of
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_PREPARE_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_FORWARD_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_COMMIT_TIMEOUT -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a. Loop a
Loop
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_WRONG_EXPECTED_VERSION -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ Text -> ExpectedVersion -> OperationError
WrongExpectedVersion Text
stream ExpectedVersion
exp_v
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_STREAM_DELETED -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ StreamName -> OperationError
StreamDeleted forall a b. (a -> b) -> a -> b
$ Text -> StreamName
StreamName Text
stream
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_INVALID_TRANSACTION -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw OperationError
InvalidTransaction
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_ACCESS_DENIED -> forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throw forall a b. (a -> b) -> a -> b
$ forall t. StreamId t -> OperationError
AccessDenied forall a b. (a -> b) -> a -> b
$ Text -> StreamName
StreamName Text
stream
                          FieldType (Required 2 (Enumeration OpResult))
OpResult
OP_SUCCESS -> forall (f :: * -> *) a. Applicative f => a -> f a
pure forall a b. (a -> b) -> a -> b
$ forall a. a -> Loop a
Break WriteResult
res