<commit_before>{-# LANGUAGE OverloadedStrings   #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE RecordWildCards           #-}
{-# LANGUAGE TypeFamilies               #-}
{-# LANGUAGE RankNTypes               #-}

module DynamoDbEventStore.AmazonkaInterpreter where

import           Data.Aeson
import           Data.Time.Clock
import           Control.Exception.Lens
import           Data.Monoid
import           Control.Monad.Free
import           Control.Monad.Catch
import           Control.Monad.IO.Class
import           Data.Int
import           Data.Map                (Map)
import qualified Data.HashMap.Strict     as HM
import           Control.Lens
import           Data.Maybe              (fromJust)
import qualified Data.Map                as M
import           Data.List.NonEmpty
import qualified Data.ByteString         as BS
import qualified Data.ByteString.Lazy    as BL
import qualified Data.Text               as T
import qualified Data.Vector             as V
import           TextShow
import           System.Random
import           EventStoreCommands
import           System.IO (stdout)
import qualified Safe
--import           Aws
--import           Aws.Core
--import           Aws.DynamoDb.Commands
--import           Aws.DynamoDb.Core

import Network.AWS
import Network.AWS.DynamoDB
import Control.Monad.Trans.Resource
import Control.Monad.Trans.AWS (newLogger)

fieldStreamId :: T.Text
fieldStreamId = "streamId"
fieldEventNumber :: T.Text
fieldEventNumber = "eventNumber"
fieldEventType :: T.Text
fieldEventType = "eventType"
fieldPageStatus :: T.Text
fieldPageStatus = "pageStatus"
fieldBody :: T.Text
fieldBody = "body"
fieldPageKey :: T.Text
fieldPageKey = "pageKey"
fieldPagingRequired :: T.Text
fieldPagingRequired = "pagingRequired"
fieldEventKeys :: T.Text
fieldEventKeys = "eventKeys"
unpagedIndexName :: T.Text
unpagedIndexName = "unpagedIndex"

getDynamoKey :: T.Text -> Int64 -> HM.HashMap T.Text AttributeValue
getDynamoKey hashKey rangeKey =
    HM.fromList [
        (fieldStreamId, set avS (Just hashKey) attributeValue),
        (fieldEventNumber, set avN (Just (showt rangeKey)) attributeValue)
    ]

getDynamoKeyForEvent :: EventKey -> HM.HashMap T.Text AttributeValue
getDynamoKeyForEvent (EventKey (StreamId streamId, eventNumber)) =
    getDynamoKey streamId eventNumber

showText :: Int -> T.Text
showText = T.pack . show

getPagePartitionStreamId :: Int -> Int -> T.Text
getPagePartitionStreamId partition page =
  "$Page-" <> showText partition <> "-" <> showText page

getDynamoKeyForPage :: PageKey -> HM.HashMap T.Text AttributeValue
getDynamoKeyForPage (partition, pageNumber) =
  let
    hashKey = (getPagePartitionStreamId partition pageNumber)
  in
    getDynamoKey hashKey 1

-- from http://haddock.stackage.org/lts-3.2/basic-prelude-0.5.0/src/BasicPrelude.html#readMay
readMay :: Read a => T.Text -> Maybe a
readMay = Safe.readMay . T.unpack
{- 
readItemJson :: Ord k => FromJSON b => k -> Map k AttributeValue -> Maybe b
readItemJson fieldName i =
  getItemField fieldName i >>= decodeStrict

getItemField :: (DynVal b, Ord k) => k -> Map k DValue -> Maybe b
getItemField fieldName i =
  M.lookup fieldName i >>= fromValue

encodeStrictJson :: ToJSON s => s -> BS.ByteString
encodeStrictJson value =
  BL.toStrict . encode $ value

attrJson :: ToJSON s => T.Text -> s -> Attribute
attrJson name value =
  attr name (encodeStrictJson value)

-}

itemAttribute :: T.Text -> Lens' AttributeValue (Maybe v) -> v -> (T.Text, AttributeValue)
itemAttribute key lens value =
  (key, set lens (Just value) attributeValue)

runCmd :: T.Text -> EventStoreCmd (IO a) -> IO a
runCmd _ (Wait' n) = n ()
runCmd tn (GetEvent' eventKey n) = do
  let key = getDynamoKeyForEvent eventKey
  let req = (getItem tn) & (set giKey key)
  resp <- runCommand req
  n $ getResult resp
  where
    getResult :: GetItemResponse -> EventReadResult
    getResult r = do
      let i = view girsItem r
      et <- view (ix fieldEventType . avS) i
      b <- view (ix fieldBody . avB) i
      --let pageKey = readItemJson fieldPageKey i
      return (et, b, Nothing) -- pageKey)
runCmd tn (WriteEvent' (EventKey (StreamId streamId, evtNumber)) t d n) =
  catches writeItem [handler _ConditionalCheckFailedException (\_ -> n EventExists)] 
    where
      writeItem = do
        time <- getCurrentTime
        let item = HM.fromList [
                  itemAttribute fieldStreamId avS streamId,
                  itemAttribute fieldEventNumber avN (showt evtNumber),
                  itemAttribute fieldEventType avS t,
                  itemAttribute fieldPagingRequired avS (T.pack $ show time),
                  itemAttribute fieldBody avB d
                ]
        let conditionExpression = Just "attribute_not_exists(#fieldEventNumber)"
        let expressionAttributeNames = HM.fromList [("#fieldEventNumber", fieldEventNumber)]
        let req0 =
              putItem tn
              & (set piItem item)
              & (set piExpressionAttributeNames expressionAttributeNames)
              & (set piConditionExpression conditionExpression)
        _ <- runCommand req0
        n WriteSuccess
runCmd tn (GetEventsBackward' (StreamId streamId) _ _ n) =
  getBackward
    where
      toRecordedEvent :: HM.HashMap T.Text AttributeValue -> RecordedEvent
      toRecordedEvent i = fromJust $ do
        sId <- view (ix fieldStreamId . avS) i
        eventNumber <- view (ix fieldEventNumber . avN) i >>= readMay
        et <- view (ix fieldEventType . avS) i
        b <- view (ix fieldBody . avB) i
        return $ RecordedEvent sId eventNumber b et
      getBackward = do
        resp <- runCommand $
                query tn
                & (set qScanIndexForward (Just False))
                & (set qExpressionAttributeValues (HM.fromList [(":streamId",set avS (Just streamId) attributeValue)]))
                & (set qExpressionAttributeNames (HM.fromList [("#fieldStreamId", fieldStreamId)]))
                & (set qKeyConditionExpression (Just $ "#fieldStreamId = :streamId"))
        let items :: [HM.HashMap T.Text AttributeValue] = view qrsItems resp
        n $ (fmap toRecordedEvent) items
{-
runCmd tn (SetEventPage' eventKey pk n) =
  catch setEventPage exnHandler
    where
      -- todo: this function is not complete
      exnHandler (DdbError{}) = n SetEventPageError
      setEventPage = do
        let conditions = Conditions CondAnd [ Condition fieldPageKey IsNull ]
        let key = getDynamoKeyForEvent eventKey
        let pageKeyAttribute = attrJson fieldPageKey pk
        let updatePageKey = au pageKeyAttribute
        let pagingReqAttr = attrAs text fieldPagingRequired "unused"<commit_msg>Amazonka: Implemented GetEventsBackward<commit_after>{-# LANGUAGE OverloadedStrings   #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE RecordWildCards           #-}
{-# LANGUAGE TypeFamilies               #-}
{-# LANGUAGE RankNTypes               #-}

module DynamoDbEventStore.AmazonkaInterpreter where

import           Data.Aeson
import           Data.Time.Clock
import           Control.Exception.Lens
import           Data.Monoid
import           Control.Monad.Free
import           Control.Monad.Catch
import           Control.Monad.IO.Class
import           Data.Int
import           Data.Map                (Map)
import qualified Data.HashMap.Strict     as HM
import           Control.Lens
import           Data.Maybe              (fromJust)
import qualified Data.Map                as M
import           Data.List.NonEmpty
import qualified Data.ByteString         as BS
import qualified Data.ByteString.Lazy    as BL
import qualified Data.Text               as T
import qualified Data.Vector             as V
import           TextShow
import           System.Random
import           EventStoreCommands
import           System.IO (stdout)
import qualified Safe
--import           Aws
--import           Aws.Core
--import           Aws.DynamoDb.Commands
--import           Aws.DynamoDb.Core

import Network.AWS
import Network.AWS.DynamoDB
import Control.Monad.Trans.Resource
import Control.Monad.Trans.AWS (newLogger)

fieldStreamId :: T.Text
fieldStreamId = "streamId"
fieldEventNumber :: T.Text
fieldEventNumber = "eventNumber"
fieldEventType :: T.Text
fieldEventType = "eventType"
fieldPageStatus :: T.Text
fieldPageStatus = "pageStatus"
fieldBody :: T.Text
fieldBody = "body"
fieldPageKey :: T.Text
fieldPageKey = "pageKey"
fieldPagingRequired :: T.Text
fieldPagingRequired = "pagingRequired"
fieldEventKeys :: T.Text
fieldEventKeys = "eventKeys"
unpagedIndexName :: T.Text
unpagedIndexName = "unpagedIndex"

getDynamoKey :: T.Text -> Int64 -> HM.HashMap T.Text AttributeValue
getDynamoKey hashKey rangeKey =
    HM.fromList [
        (fieldStreamId, set avS (Just hashKey) attributeValue),
        (fieldEventNumber, set avN (Just (showt rangeKey)) attributeValue)
    ]

getDynamoKeyForEvent :: EventKey -> HM.HashMap T.Text AttributeValue
getDynamoKeyForEvent (EventKey (StreamId streamId, eventNumber)) =
    getDynamoKey streamId eventNumber

showText :: Int -> T.Text
showText = T.pack . show

getPagePartitionStreamId :: Int -> Int -> T.Text
getPagePartitionStreamId partition page =
  "$Page-" <> showText partition <> "-" <> showText page

getDynamoKeyForPage :: PageKey -> HM.HashMap T.Text AttributeValue
getDynamoKeyForPage (partition, pageNumber) =
  let
    hashKey = (getPagePartitionStreamId partition pageNumber)
  in
    getDynamoKey hashKey 1

-- from http://haddock.stackage.org/lts-3.2/basic-prelude-0.5.0/src/BasicPrelude.html#readMay
readMay :: Read a => T.Text -> Maybe a
readMay = Safe.readMay . T.unpack
{- 
readItemJson :: Ord k => FromJSON b => k -> Map k AttributeValue -> Maybe b
readItemJson fieldName i =
  getItemField fieldName i >>= decodeStrict

getItemField :: (DynVal b, Ord k) => k -> Map k DValue -> Maybe b
getItemField fieldName i =
  M.lookup fieldName i >>= fromValue

encodeStrictJson :: ToJSON s => s -> BS.ByteString
encodeStrictJson value =
  BL.toStrict . encode $ value

attrJson :: ToJSON s => T.Text -> s -> Attribute
attrJson name value =
  attr name (encodeStrictJson value)

-}

itemAttribute :: T.Text -> Lens' AttributeValue (Maybe v) -> v -> (T.Text, AttributeValue)
itemAttribute key lens value =
  (key, set lens (Just value) attributeValue)

runCmd :: T.Text -> EventStoreCmd (IO a) -> IO a
runCmd _ (Wait' n) = n ()
runCmd tn (GetEvent' eventKey n) = do
  let key = getDynamoKeyForEvent eventKey
  let req = (getItem tn) & (set giKey key)
  resp <- runCommand req
  n $ getResult resp
  where
    getResult :: GetItemResponse -> EventReadResult
    getResult r = do
      let i = view girsItem r
      et <- view (ix fieldEventType . avS) i
      b <- view (ix fieldBody . avB) i
      --let pageKey = readItemJson fieldPageKey i
      return (et, b, Nothing) -- pageKey)
runCmd tn (WriteEvent' (EventKey (StreamId streamId, evtNumber)) t d n) =
  catches writeItem [handler _ConditionalCheckFailedException (\_ -> n EventExists)] 
    where
      writeItem = do
        time <- getCurrentTime
        let item = HM.fromList [
                  itemAttribute fieldStreamId avS streamId,
                  itemAttribute fieldEventNumber avN (showt evtNumber),
                  itemAttribute fieldEventType avS t,
                  itemAttribute fieldPagingRequired avS (T.pack $ show time),
                  itemAttribute fieldBody avB d
                ]
        let conditionExpression = Just "attribute_not_exists(#fieldEventNumber)"
        let expressionAttributeNames = HM.fromList [("#fieldEventNumber", fieldEventNumber)]
        let req0 =
              putItem tn
              & (set piItem item)
              & (set piExpressionAttributeNames expressionAttributeNames)
              & (set piConditionExpression conditionExpression)
        _ <- runCommand req0
        n WriteSuccess
runCmd tn (GetEventsBackward' (StreamId streamId) _ _ n) =
  getBackward
    where
      toRecordedEvent :: HM.HashMap T.Text AttributeValue -> RecordedEvent
      toRecordedEvent i = fromJust $ do -- todo: remove fromJust
        sId <- view (ix fieldStreamId . avS) i
        eventNumber <- view (ix fieldEventNumber . avN) i >>= readMay
        et <- view (ix fieldEventType . avS) i
        b <- view (ix fieldBody . avB) i
        return $ RecordedEvent sId eventNumber b et
      getBackward = do
        resp <- runCommand $
                query tn
                & (set qScanIndexForward (Just False))
                & (set qExpressionAttributeValues (HM.fromList [(":streamId",set avS (Just streamId) attributeValue)]))
                & (set qExpressionAttributeNames (HM.fromList [("#fieldStreamId", fieldStreamId)]))
                & (set qKeyConditionExpression (Just $ fieldStreamId </> " = :streamId"))
        let items :: [HM.HashMap T.Text AttributeValue] = view qrsItems resp
        n $ (fmap toRecordedEvent) items
{-
runCmd tn (SetEventPage' eventKey pk n) =
  catch setEventPage exnHandler
    where
      -- todo: this function is not complete
      exnHandler (DdbError{}) = n SetEventPageError
      setEventPage = do
        let conditions = Conditions CondAnd [ Condition fieldPageKey IsNull ]
        let key = getDynamoKeyForEvent eventKey
        let pageKeyAttribute = attrJson fieldPageKey pk
        let updatePageKey = au pageKeyAttribute
        let pagingReqAttr = attrAs text fieldPagingRequired "unused"
{'fraction_non_alphanumeric': 0.08663347945612601, 'fraction_numerical': 0.0014859945018203433, 'mean_word_length': 2.6816192560175054, 'pattern_counts': {'":': 2, '<': 30, '<?xml version=': 0, '>': 80, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2485341', 'n_tokens_mistral': 3970, 'n_tokens_neox': 3902, 'n_words': 1152}