simplexmq/src/Simplex/Messaging/Server/StoreLog.hs

{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# OPTIONS_GHC -fno-warn-orphans #-}

module Simplex.Messaging.Server.StoreLog
  ( StoreLog, -- constructors are not exported
    StoreLogRecord (..), -- used in tests
    openWriteStoreLog,
    openReadStoreLog,
    storeLogFilePath,
    closeStoreLog,
    writeStoreLogRecord,
    logCreateQueue,
    logCreateLink,
    logDeleteLink,
    logSecureQueue,
    logUpdateKeys,
    logAddNotifier,
    logSuspendQueue,
    logBlockQueue,
    logUnblockQueue,
    logDeleteQueue,
    logDeleteNotifier,
    logUpdateQueueTime,
    logNewService,
    logQueueService,
    readWriteStoreLog,
    readLogLines,
    foldLogLines,
  )
where

import Control.Applicative (optional, (<|>))
import qualified Control.Exception as E
import Control.Logger.Simple
import Control.Monad
import qualified Data.Attoparsec.ByteString.Char8 as A
import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import Data.Functor (($>))
import Data.List (sort, stripPrefix)
import Data.List.NonEmpty (NonEmpty)
import Data.Maybe (mapMaybe)
import qualified Data.Text as T
import Data.Time.Clock (UTCTime, addUTCTime, getCurrentTime, nominalDay)
import Data.Time.Format.ISO8601 (iso8601Show, iso8601ParseM)
import GHC.IO (catchAny)
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.StoreLog.Types
import Simplex.Messaging.SystemTime
import Simplex.Messaging.Util (ifM, tshow, unlessM, whenM)
import System.Directory (doesFileExist, listDirectory, removeFile, renameFile)
import System.IO
import System.FilePath (takeDirectory, takeFileName)

data StoreLogRecord
  = CreateQueue RecipientId QueueRec
  | CreateLink RecipientId LinkId QueueLinkData
  | DeleteLink RecipientId
  | SecureQueue QueueId SndPublicAuthKey
  | UpdateKeys RecipientId (NonEmpty RcvPublicAuthKey)
  | AddNotifier QueueId NtfCreds
  | SuspendQueue QueueId
  | BlockQueue QueueId BlockingInfo
  | UnblockQueue QueueId
  | DeleteQueue QueueId
  | DeleteNotifier QueueId
  | UpdateTime QueueId SystemDate
  | NewService ServiceRec
  | QueueService RecipientId ASubscriberParty (Maybe ServiceId)
  deriving (Show)

data SLRTag
  = CreateQueue_
  | CreateLink_
  | DeleteLink_
  | SecureQueue_
  | UpdateKeys_
  | AddNotifier_
  | SuspendQueue_
  | BlockQueue_
  | UnblockQueue_
  | DeleteQueue_
  | DeleteNotifier_
  | UpdateTime_
  | NewService_
  | QueueService_

instance StrEncoding QueueRec where
  strEncode QueueRec {recipientKeys, rcvDhSecret, rcvServiceId, senderId, senderKey, queueMode, queueData, notifier, status, updatedAt} =
    B.concat
      [ p "rk=" recipientKeys,
        p " rdh=" rcvDhSecret,
        p " sid=" senderId,
        p " sk=" senderKey,
        maybe "" ((" queue_mode=" <>) . smpEncode) queueMode,
        opt " link_id=" (fst <$> queueData),
        opt " queue_data=" (snd <$> queueData),
        opt " notifier=" notifier,
        opt " updated_at=" updatedAt,
        statusStr,
        opt " rsrv=" rcvServiceId
      ]
    where
      p :: StrEncoding a => ByteString -> a -> ByteString
      p param = (param <>) . strEncode
      opt :: StrEncoding a => ByteString -> Maybe a -> ByteString
      opt = maybe "" . p
      statusStr = case status of
        EntityActive -> ""
        _ -> " status=" <> strEncode status

  strP = do
    recipientKeys <- "rk=" *> strP_
    rcvDhSecret <- "rdh=" *> strP_
    senderId <- "sid=" *> strP_
    senderKey <- "sk=" *> strP
    queueMode <-
      toQueueMode <$> (" sndSecure=" *> strP)
        <|> Just <$> (" queue_mode=" *> smpP)
        <|> pure Nothing -- unknown queue mode, we cannot imply that it is contact address
    queueData <- optional $ (,) <$> (" link_id=" *> strP) <*> (" queue_data=" *> strP)
    notifier <- optional $ " notifier=" *> strP
    updatedAt <- optional $ " updated_at=" *> strP
    status <- (" status=" *> strP) <|> pure EntityActive
    rcvServiceId <- optional $ " rsrv=" *> strP
    pure
      QueueRec
        { recipientKeys,
          rcvDhSecret,
          senderId,
          senderKey,
          queueMode,
          queueData,
          notifier,
          status,
          updatedAt,
          rcvServiceId
        }
    where
      toQueueMode sndSecure = Just $ if sndSecure then QMMessaging else QMContact

instance StrEncoding SLRTag where
  strEncode = \case
    CreateQueue_ -> "CREATE"
    CreateLink_ -> "LINK"
    DeleteLink_ -> "LDELETE"
    SecureQueue_ -> "SECURE"
    UpdateKeys_ -> "KEYS"
    AddNotifier_ -> "NOTIFIER"
    SuspendQueue_ -> "SUSPEND"
    BlockQueue_ -> "BLOCK"
    UnblockQueue_ -> "UNBLOCK"
    DeleteQueue_ -> "DELETE"
    DeleteNotifier_ -> "NDELETE"
    UpdateTime_ -> "TIME"
    NewService_ -> "NEW_SERVICE"
    QueueService_ -> "QUEUE_SERVICE"

  strP =
    A.choice
      [ "CREATE" $> CreateQueue_,
        "LINK" $> CreateLink_,
        "LDELETE" $> DeleteLink_,
        "SECURE" $> SecureQueue_,
        "KEYS" $> UpdateKeys_,
        "NOTIFIER" $> AddNotifier_,
        "SUSPEND" $> SuspendQueue_,
        "BLOCK" $> BlockQueue_,
        "UNBLOCK" $> UnblockQueue_,
        "DELETE" $> DeleteQueue_,
        "NDELETE" $> DeleteNotifier_,
        "TIME" $> UpdateTime_,
        "NEW_SERVICE" $> NewService_,
        "QUEUE_SERVICE" $> QueueService_
      ]

instance StrEncoding StoreLogRecord where
  strEncode = \case
    CreateQueue rId q -> B.unwords [strEncode CreateQueue_, "rid=" <> strEncode rId, strEncode q]
    CreateLink rId lnkId d -> strEncode (CreateLink_, rId, lnkId, d)
    DeleteLink rId -> strEncode (DeleteLink_, rId)
    SecureQueue rId sKey -> strEncode (SecureQueue_, rId, sKey)
    UpdateKeys rId rKeys -> strEncode (UpdateKeys_, rId, rKeys)
    AddNotifier rId ntfCreds -> strEncode (AddNotifier_, rId, ntfCreds)
    SuspendQueue rId -> strEncode (SuspendQueue_, rId)
    BlockQueue rId info -> strEncode (BlockQueue_, rId, info)
    UnblockQueue rId -> strEncode (UnblockQueue_, rId)
    DeleteQueue rId -> strEncode (DeleteQueue_, rId)
    DeleteNotifier rId -> strEncode (DeleteNotifier_, rId)
    UpdateTime rId t ->  strEncode (UpdateTime_, rId, t)
    NewService sr -> strEncode (NewService_, sr)
    QueueService rId party serviceId -> strEncode (QueueService_, rId, party, serviceId)

  strP =
    strP_ >>= \case
      CreateQueue_ -> CreateQueue <$> ("rid=" *> strP_) <*> strP
      CreateLink_ -> CreateLink <$> strP_ <*> strP_ <*> strP
      DeleteLink_ -> DeleteLink <$> strP
      SecureQueue_ -> SecureQueue <$> strP_ <*> strP
      UpdateKeys_ -> UpdateKeys <$> strP_ <*> strP
      AddNotifier_ -> AddNotifier <$> strP_ <*> strP
      SuspendQueue_ -> SuspendQueue <$> strP
      BlockQueue_ -> BlockQueue <$> strP_ <*> strP
      UnblockQueue_ -> UnblockQueue <$> strP
      DeleteQueue_ -> DeleteQueue <$> strP
      DeleteNotifier_ -> DeleteNotifier <$> strP
      UpdateTime_ -> UpdateTime <$> strP_ <*> strP
      NewService_ -> NewService <$> strP
      QueueService_ -> QueueService <$> strP_ <*> strP_ <*> strP

openWriteStoreLog :: Bool -> FilePath -> IO (StoreLog 'WriteMode)
openWriteStoreLog append f = do
  h <- openFile f $ if append then AppendMode else WriteMode
  hSetBuffering h LineBuffering
  pure $ WriteStoreLog f h

openReadStoreLog :: FilePath -> IO (StoreLog 'ReadMode)
openReadStoreLog f = do
  unlessM (doesFileExist f) (writeFile f "")
  ReadStoreLog f <$> openFile f ReadMode

storeLogFilePath :: StoreLog a -> FilePath
storeLogFilePath = \case
  WriteStoreLog f _ -> f
  ReadStoreLog f _ -> f

closeStoreLog :: StoreLog a -> IO ()
closeStoreLog = \case
  WriteStoreLog _ h -> close_ h
  ReadStoreLog _ h -> close_ h
  where
    close_ h = hClose h `catchAny` \e -> logError ("STORE: closeStoreLog, error closing, " <> tshow e)

writeStoreLogRecord :: StrEncoding r => StoreLog 'WriteMode -> r -> IO ()
writeStoreLogRecord (WriteStoreLog _ h) r = E.uninterruptibleMask_ $ do
  B.hPut h $ strEncode r `B.snoc` '\n' -- hPutStrLn makes write non-atomic for length > 1024
  hFlush h

logCreateQueue :: StoreLog 'WriteMode -> RecipientId -> QueueRec -> IO ()
logCreateQueue s rId q = writeStoreLogRecord s $ CreateQueue rId q

logCreateLink :: StoreLog 'WriteMode -> RecipientId -> LinkId -> QueueLinkData -> IO ()
logCreateLink s rId lnkId d = writeStoreLogRecord s $ CreateLink rId lnkId d

logDeleteLink :: StoreLog 'WriteMode -> RecipientId -> IO ()
logDeleteLink s = writeStoreLogRecord s . DeleteLink

logSecureQueue :: StoreLog 'WriteMode -> QueueId -> SndPublicAuthKey -> IO ()
logSecureQueue s qId sKey = writeStoreLogRecord s $ SecureQueue qId sKey

logUpdateKeys :: StoreLog 'WriteMode -> QueueId -> NonEmpty RcvPublicAuthKey -> IO ()
logUpdateKeys s rId rKeys = writeStoreLogRecord s $ UpdateKeys rId rKeys

logAddNotifier :: StoreLog 'WriteMode -> QueueId -> NtfCreds -> IO ()
logAddNotifier s qId ntfCreds = writeStoreLogRecord s $ AddNotifier qId ntfCreds

logSuspendQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
logSuspendQueue s = writeStoreLogRecord s . SuspendQueue

logBlockQueue :: StoreLog 'WriteMode -> QueueId -> BlockingInfo -> IO ()
logBlockQueue s qId info = writeStoreLogRecord s $ BlockQueue qId info

logUnblockQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
logUnblockQueue s = writeStoreLogRecord s . UnblockQueue

logDeleteQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
logDeleteQueue s = writeStoreLogRecord s . DeleteQueue

logDeleteNotifier :: StoreLog 'WriteMode -> QueueId -> IO ()
logDeleteNotifier s = writeStoreLogRecord s . DeleteNotifier

logUpdateQueueTime :: StoreLog 'WriteMode -> QueueId -> SystemDate -> IO ()
logUpdateQueueTime s qId t = writeStoreLogRecord s $ UpdateTime qId t

logNewService :: StoreLog 'WriteMode -> ServiceRec -> IO ()
logNewService s  = writeStoreLogRecord s . NewService

logQueueService :: (PartyI p, ServiceParty p) => StoreLog 'WriteMode -> RecipientId -> SParty p -> Maybe ServiceId -> IO ()
logQueueService s rId party = writeStoreLogRecord s . QueueService rId (ASP party)

readWriteStoreLog :: (FilePath -> s -> IO ()) -> (StoreLog 'WriteMode -> s -> IO ()) -> FilePath -> s -> IO (StoreLog 'WriteMode)
readWriteStoreLog readStore writeStore f st =
  ifM
    (doesFileExist tempBackup)
    (useTempBackup >> readWriteLog)
    (ifM (doesFileExist f) readWriteLog (writeLog "creating store log..."))
  where
    f' = T.pack f
    tempBackup = f <> ".start"
    useTempBackup = do
      -- preserve current file, use temp backup
      logWarn $ "Server terminated abnormally on last start, restoring state from " <> T.pack tempBackup
      whenM (doesFileExist f) $ do
        renameFile f (f <> ".bak")
        logNote $ "preserved incomplete state " <> f' <> " as " <> (f' <> ".bak")
      renameFile tempBackup f
    readWriteLog = do
      -- log backup is made in two steps to mitigate the crash during the compacting.
      -- Temporary backup file .start will be used when it is present.
      readStore f st
      renameFile f tempBackup -- 1) make temp backup
      s <- writeLog "compacting store log (do not terminate)..." -- 2) save state
      renameBackup -- 3) timed backup
      removeStoreLogBackups f
      pure s
    writeLog msg = do
      s <- openWriteStoreLog False f
      logNote msg
      writeStore s st
      pure s
    renameBackup = do
      ts <- getCurrentTime
      let timedBackup = f <> "." <> iso8601Show ts
      renameFile tempBackup timedBackup
      logNote $ "original state preserved as " <> T.pack timedBackup

removeStoreLogBackups :: FilePath -> IO ()
removeStoreLogBackups f = do
  ts <- getCurrentTime
  times <- sort . mapMaybe backupPathTime <$> listDirectory (takeDirectory f)
  let new = addUTCTime (- nominalDay) ts
      old = addUTCTime (- oldBackupTTL) ts
      times1 = filter (< new) times -- exclude backups newer than 24 hours
      times2 = take (length times1 - minOldBackups) times1 -- keep 3 backups older than 24 hours
      toDelete = filter (< old) times2 -- remove all backups older than 21 day
  mapM_ (removeFile . backupPath) toDelete
  when (length toDelete > 0) $ do
    putStrLn $ "Removed " <> show (length toDelete) <> " backups:"
    mapM_ (putStrLn . backupPath) toDelete
  where
    backupPathTime :: FilePath -> Maybe UTCTime
    backupPathTime = iso8601ParseM <=< stripPrefix backupPathPfx
    backupPath :: UTCTime -> FilePath
    backupPath ts = f <> "." <> iso8601Show ts
    backupPathPfx = takeFileName f <> "."
    minOldBackups = 3
    oldBackupTTL = 21 * nominalDay

readLogLines :: Bool -> FilePath -> (Bool -> B.ByteString -> IO ()) -> IO ()
readLogLines tty f action = foldLogLines tty f (const action) ()

foldLogLines :: Bool -> FilePath -> (a -> Bool -> B.ByteString -> IO a) -> a -> IO a
foldLogLines tty f action initValue = do
  (count :: Int, acc) <- withFile f ReadMode $ \h -> ifM (hIsEOF h) (pure (0, initValue)) (loop h 0 initValue)
  putStrLn $ progress count
  pure acc
  where
    loop h !i !acc = do
      s <- B.hGetLine h
      eof <- hIsEOF h
      acc' <- action acc eof s
      let i' = i + 1
      when (tty && i' `mod` 100000 == 0) $ putStr (progress i' <> "\r") >> hFlush stdout
      if eof then pure (i', acc') else loop h i' acc'
    progress i = "Processed: " <> show i <> " log lines"