Protocol.hs

{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE GeneralisedNewtypeDeriving #-}
module Database.Franz.Internal.Protocol
  ( apiVersion
  , defaultPort
  , IndexName
  , StreamName(..)
  , streamNameToPath
  , FranzException(..)
  , RequestType(..)
  , ItemRef(..)
  , Query(..)
  , RawRequest(..)
  , ResponseId
  , ResponseHeader(..)
  , PayloadHeader(..)) where

import Control.Exception (Exception)
import qualified Data.ByteString.Char8 as B
import Data.Serialize hiding (getInt64le)
import Database.Franz.Internal.IO (getInt64le)
import Data.Hashable (Hashable)
import Data.String
import qualified Data.Text as T
import qualified Data.Text.Encoding as T
import Network.Socket (PortNumber)
import GHC.Generics (Generic)
import qualified Data.ByteString.FastBuilder as BB
import qualified Data.Vector as V

apiVersion :: B.ByteString
apiVersion = B.pack "0"

defaultPort :: PortNumber
defaultPort = 1886

type IndexName = B.ByteString

newtype StreamName = StreamName { unStreamName :: B.ByteString }
  deriving newtype (Show, Eq, Ord, Hashable, Serialize)

streamNameToPath :: StreamName -> FilePath
streamNameToPath = T.unpack . T.decodeUtf8 . unStreamName

-- | UTF-8 encoded
instance IsString StreamName where
  fromString = StreamName . BB.toStrictByteString . BB.stringUtf8

data FranzException = MalformedRequest !String
  | StreamNotFound !FilePath
  | IndexNotFound !IndexName ![IndexName]
  | InternalError !String
  | ClientError !String
  deriving (Show, Generic)
instance Serialize FranzException
instance Exception FranzException

data RequestType = AllItems
  | LastItem
  | FirstItem
  deriving (Show, Eq, Generic)
instance Serialize RequestType

data ItemRef = BySeqNum !Int -- ^ sequential number
  | ByIndex !IndexName !Int -- ^ index name and value
  deriving (Show, Eq, Generic)
instance Serialize ItemRef

data Query = Query
  { reqStream :: !StreamName
  , reqFrom :: !ItemRef -- ^ name of the index to search
  , reqTo :: !ItemRef -- ^ name of the index to search
  , reqType :: !RequestType
  } deriving (Show, Eq, Generic)
instance Serialize Query

data RawRequest
  = RawRequest !ResponseId !Query
  | RawClean !ResponseId deriving Generic
instance Serialize RawRequest

type ResponseId = Int

data ResponseHeader = Response !ResponseId
    -- ^ response ID, number of streams; there are items satisfying the query
    | ResponseWait !ResponseId -- ^ response ID; requested elements are not available right now
    | ResponseError !ResponseId !FranzException -- ^ something went wrong
    deriving (Show, Generic)
instance Serialize ResponseHeader

-- | Initial seqno, final seqno, base offset, index names
data PayloadHeader = PayloadHeader !Int !Int !Int !(V.Vector IndexName)

instance Serialize PayloadHeader where
  put (PayloadHeader s t u xs) = f s *> f t *> f u
    *> put (V.length xs) *> mapM_ put xs where
    f = putInt64le . fromIntegral
  get = PayloadHeader <$> getInt64le <*> getInt64le <*> getInt64le
    <*> do
      len <- get
      V.replicateM len get