{-# language DataKinds #-}
{-# language DuplicateRecordFields #-}
{-# language NamedFieldPuns #-}
{-# language TypeApplications #-}

module Kafka.Record.Request
  ( Record(..)
  , Header(..)
  , toChunks
  , toChunksOnto
  ) where

import Data.Bytes (Bytes)
import Data.Bytes.Chunks (Chunks(ChunksCons,ChunksNil))
import Data.Int (Int32,Int64)
import Data.Primitive (SmallArray)
import Data.Text (Text)
import Kafka.Builder (Builder)

import qualified Arithmetic.Nat as Nat
import qualified Data.Bytes as Bytes
import qualified Data.Bytes.Chunks as Chunks
import qualified Data.Bytes.Text.Utf8 as Utf8
import qualified Data.Primitive as PM
import qualified Kafka.Builder as Builder
import qualified Kafka.Builder.Bounded as Bounded

-- | Information about @Record@ from Kafka documentation:
--
-- > length: varint
-- > attributes: int8
-- >     bit 0~7: unused
-- > timestampDelta: varlong
-- > offsetDelta: varint
-- > keyLength: varint
-- > key: byte[]
-- > valueLen: varint
-- > value: byte[]
-- > Headers => [Header]
--
data Record = Record
  { Record -> Int64
timestampDelta :: !Int64
  , Record -> Int32
offsetDelta :: !Int32
  , Record -> Bytes
key :: {-# UNPACK #-} !Bytes
    -- ^ Setting the key to the empty byte sequence causes it to be treated
    -- as though it were the null key. Technically, we could do the right
    -- thing by wrapping the Bytes with Maybe, but using the empty string
    -- as the key is a terrible idea anyway.
  , Record -> Bytes
value :: {-# UNPACK #-} !Bytes
    -- ^ In a data-encoding setting, it actually makes more sense for this
    -- to be Chunks, not Bytes. But in a data-decoding setting, Bytes makes
    -- more sense. It might be better to create a separate type for each
    -- setting.
  , Record -> SmallArray Header
headers :: {-# UNPACK #-} !(SmallArray Header)
    -- ^ In a data-encoding setting, it makes more sense for this to be
    -- a chunked (or builder-like) type.
  }

toChunks :: Record -> Chunks
toChunks :: Record -> Chunks
toChunks Record
r = Record -> Chunks -> Chunks
toChunksOnto Record
r Chunks
ChunksNil

-- | Variant of 'toChunks' that gives the caller control over what chunks
-- come after the encoded record. For example, it is possible to improve
-- the performance of
--
-- > foldMap toChunks records
--
-- by rewriting it as
--
-- > foldr toChunksOnto ChunksNil records
toChunksOnto :: Record -> Chunks -> Chunks
toChunksOnto :: Record -> Chunks -> Chunks
toChunksOnto Record
r Chunks
c = Bytes -> Chunks -> Chunks
ChunksCons
  (ByteArray -> Bytes
Bytes.fromByteArray (Nat 10 -> Builder 10 -> ByteArray
forall (n :: Nat). Nat n -> Builder n -> ByteArray
Bounded.run Nat 10
forall (n :: Nat). KnownNat n => Nat n
Nat.constant (Int -> Builder 10
Bounded.varIntNative (Int -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
n))))
  Chunks
recordChunks
  where
  (Int
n,Chunks
recordChunks) = Int -> Builder -> Chunks -> (Int, Chunks)
Builder.runOntoLength Int
128 (Record -> Builder
encodeWithoutLength Record
r) Chunks
c

encodeWithoutLength :: Record -> Builder
encodeWithoutLength :: Record -> Builder
encodeWithoutLength Record{Int64
$sel:timestampDelta:Record :: Record -> Int64
timestampDelta :: Int64
timestampDelta,Int32
$sel:offsetDelta:Record :: Record -> Int32
offsetDelta :: Int32
offsetDelta,Bytes
$sel:key:Record :: Record -> Bytes
key :: Bytes
key,Bytes
$sel:value:Record :: Record -> Bytes
value :: Bytes
value,SmallArray Header
$sel:headers:Record :: Record -> SmallArray Header
headers :: SmallArray Header
headers} =
  Word8 -> Builder
Builder.word8 Word8
0x00
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Int64 -> Builder
Builder.varInt64 Int64
timestampDelta
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Int32 -> Builder
Builder.varInt32 Int32
offsetDelta
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Int -> Builder
Builder.varIntNative
    (case Bytes -> Int
Bytes.length Bytes
key of
      Int
0 -> (-Int
1)
      Int
klen -> Int
klen
    )
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Bytes -> Builder
Builder.copy Bytes
key
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Int -> Builder
Builder.varIntNative (Bytes -> Int
Bytes.length Bytes
value)
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Bytes -> Builder
Builder.bytes Bytes
value
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Int -> Builder
Builder.varIntNative (SmallArray Header -> Int
forall a. SmallArray a -> Int
PM.sizeofSmallArray SmallArray Header
headers)
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  (Header -> Builder) -> SmallArray Header -> Builder
forall m a. Monoid m => (a -> m) -> SmallArray a -> m
forall (t :: * -> *) m a.
(Foldable t, Monoid m) =>
(a -> m) -> t a -> m
foldMap Header -> Builder
encodeHeader SmallArray Header
headers

-- | Information about @Header@ from Kafka documentation:
--
-- > headerKeyLength: varint
-- > headerKey: String
-- > headerValueLength: varint
-- > value: byte[]
data Header = Header
  { Header -> Text
key :: {-# UNPACK #-} !Text
    -- ^ Header key. For records that we are encoding, text (rather than
    -- some kind of builder) is a reasonable choice since header keys are
    -- typically not assembled from smaller pieces.
  , Header -> Bytes
value :: {-# UNPACK #-} !Bytes
    -- ^ Header value. This is currently Bytes, and I'm torn about whether
    -- or not to change it to a builder or chunks type. On one hand, it
    -- makes a more sense for it to be done that way, but on the
    -- other hand, I do not think that values are particularly likely
    -- to be constructed since they are small.
  }

encodeHeader :: Header -> Builder
encodeHeader :: Header -> Builder
encodeHeader Header{Text
$sel:key:Header :: Header -> Text
key :: Text
key,Bytes
$sel:value:Header :: Header -> Bytes
value :: Bytes
value} =
  Int -> Builder
Builder.varIntNative (Bytes -> Int
Bytes.length Bytes
keyBytes)
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Bytes -> Builder
Builder.copy Bytes
keyBytes
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Int -> Builder
Builder.varIntNative (Bytes -> Int
Bytes.length Bytes
value)
  Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<>
  Bytes -> Builder
Builder.copy Bytes
value
  where
  keyBytes :: Bytes
keyBytes = Text -> Bytes
Utf8.fromText Text
key