module Haskakafka.Consumer
( runConsumerConf
, runConsumer
, newKafkaConsumerConf
, newKafkaConsumer
, setRebalanceCallback
, assign
, subscribe
, pollMessage
, closeConsumer
, setOffsetStore

-- Types
, CIT.ConsumerGroupId (..)
, CIT.TopicName (..)
, CIT.BrokersString (..)
, CIT.KafkaTopicPartition (..)
)
where

import           Control.Exception
import           Foreign
import           Haskakafka
import           Haskakafka.Consumer.Internal.Convert
import           Haskakafka.Consumer.Internal.Types
import           Haskakafka.InternalRdKafka
import           Haskakafka.InternalRdKafkaEnum
import           Haskakafka.InternalSetup
import           Haskakafka.InternalShared
import           Haskakafka.InternalTypes

import qualified Haskakafka.Consumer.Internal.Types   as CIT

-- | Runs high-level kafka consumer.
--
-- A callback provided is expected to call 'pollMessage' when convenient.
runConsumerConf :: KafkaConf                            -- ^ Consumer config (see 'newKafkaConsumerConf')
                -> BrokersString                        -- ^ Comma separated list of brokers with ports (e.g. @localhost:9092@)
                -> [TopicName]                          -- ^ List of topics to be consumed
                -> (Kafka -> IO (Either KafkaError ())) -- ^ A callback function to poll and handle messages
                -> IO (Either KafkaError ())
runConsumerConf c bs ts f =
    bracket mkConsumer clConsumer runHandler
    where
        mkConsumer = do
            kafka <- newKafkaConsumer bs c
            _ <- setHlConsumer kafka
            sErr  <- subscribe kafka ts
            return $ if hasError sErr
                         then Left (sErr, kafka)
                         else Right kafka

        clConsumer (Left (_, kafka)) = kafkaErrorToEither <$> closeConsumer kafka
        clConsumer (Right kafka) = kafkaErrorToEither <$> closeConsumer kafka

        runHandler (Left (err, _)) = return $ Left err
        runHandler (Right kafka) = f kafka

-- | Runs high-level kafka consumer.
--
-- A callback provided is expected to call 'pollMessage' when convenient.
runConsumer :: ConsumerGroupId                       -- ^ Consumer group id (a @group.id@ property of a kafka consumer)
             -> ConfigOverrides                      -- ^ Extra kafka consumer parameters (see kafka documentation)
             -> BrokersString                        -- ^ Comma separated list of brokers with ports (e.g. @localhost:9092@)
             -> [TopicName]                          -- ^ List of topics to be consumed
             -> (Kafka -> IO (Either KafkaError ())) -- ^ A callback function to poll and handle messages
             -> IO (Either KafkaError ())
runConsumer g c bs ts f = do
    conf <- newKafkaConsumerConf g c
    runConsumerConf conf bs ts f

-- | Creates a new kafka configuration for a consumer with a specified 'ConsumerGroupId'.
newKafkaConsumerConf :: ConsumerGroupId  -- ^ Consumer group id (a @group.id@ property of a kafka consumer)
                     -> ConfigOverrides  -- ^ Extra kafka consumer parameters (see kafka documentation)
                     -> IO KafkaConf     -- ^ Kafka configuration which can be altered before it is used in 'newKafkaConsumer'
newKafkaConsumerConf (ConsumerGroupId gid) conf = do
    kc <- kafkaConf conf
    setKafkaConfValue kc "group.id" gid
    return kc

-- | Creates a new kafka consumer
newKafkaConsumer :: BrokersString -- ^ Comma separated list of brokers with ports (e.g. @localhost:9092@)
                 -> KafkaConf     -- ^ Kafka configuration for a consumer (see 'newKafkaConsumerConf')
                 -> IO Kafka      -- ^ Kafka instance
newKafkaConsumer (BrokersString bs) conf = do
    kafka <- newKafkaPtr RdKafkaConsumer conf
    addBrokers kafka bs
    return kafka

-- | Sets a callback that is called when rebalance is needed.
--
-- Callback implementations suppose to watch for 'KafkaResponseError' 'RdKafkaRespErrAssignPartitions' and
-- for 'KafkaResponseError' 'RdKafkaRespErrRevokePartitions'. Other error codes are not expected and would indicate
-- something really bad happening in a system, or bugs in @librdkafka@ itself.
--
-- A callback is expected to call 'assign' according to the error code it receives.
--
--     * When 'RdKafkaRespErrAssignPartitions' happens 'assign' should be called with all the partitions it was called with.
--       It is OK to alter partitions offsets before calling 'assign'.
--
--     * When 'RdKafkaRespErrRevokePartitions' happens 'assign' should be called with an empty list of partitions.
setRebalanceCallback :: KafkaConf
                     -> (Kafka -> KafkaError -> [KafkaTopicPartition] -> IO ())
                     -> IO ()
setRebalanceCallback (KafkaConf conf) callback = rdKafkaConfSetRebalanceCb conf realCb
  where
    realCb :: Ptr RdKafkaT -> RdKafkaRespErrT -> Ptr RdKafkaTopicPartitionListT -> Ptr Word8 -> IO ()
    realCb rk err pl _ = do
        rk' <- newForeignPtr_ rk
        pl' <- peek pl
        ps  <- fromNativeTopicPartitionList pl'
        callback (Kafka rk' (KafkaConf conf)) (KafkaResponseError err) ps

-- | Assigns specified partitions to a current consumer.
-- Assigning an empty list means unassigning from all partitions that are currently assigned.
-- See 'setRebalanceCallback' for more details.
assign :: Kafka -> [KafkaTopicPartition] -> IO KafkaError
assign (Kafka k _) ps =
    let pl = if null ps
                then newForeignPtr_ nullPtr
                else toNativeTopicPartitionList ps
    in  KafkaResponseError <$> (pl >>= rdKafkaAssign k)

-- | Subscribes to a given list of topics.
--
-- Wildcard (regex) topics are supported by the librdkafka assignor:
-- any topic name in the topics list that is prefixed with @^@ will
-- be regex-matched to the full list of topics in the cluster and matching
-- topics will be added to the subscription list.
subscribe :: Kafka -> [TopicName] -> IO KafkaError
subscribe (Kafka k _) ts = do
    pl <- newRdKafkaTopicPartitionListT (length ts)
    mapM_ (\(TopicName t) -> rdKafkaTopicPartitionListAdd pl t (-1)) ts
    KafkaResponseError <$> rdKafkaSubscribe k pl

-- | Closes the consumer and destroys it.
closeConsumer :: Kafka -> IO KafkaError
closeConsumer (Kafka k _) = KafkaResponseError <$> rdKafkaConsumerClose k

-----------------------------------------------------------------------------
setTopicValue :: KafkaTopic -> String -> String -> IO ()
setTopicValue (KafkaTopic _ _ conf) = setKafkaTopicConfValue conf

pollMessage :: Kafka
               -> Int -- ^ the timeout, in milliseconds (@10^3@ per second)
               -> IO (Either KafkaError KafkaMessage) -- ^ Left on error or timeout, right for success
pollMessage (Kafka k _) timeout =
    rdKafkaConsumerPoll k (fromIntegral timeout) >>= fromMessagePtr

-- | Redirects 'consumeMessage' to poll. Implementation details.
setHlConsumer :: Kafka -> IO KafkaError
setHlConsumer (Kafka k _) = KafkaResponseError <$> rdKafkaPollSetConsumer k

-- | Sets the offset store for a specified topic.
-- @librdkafka@ supports both @broker@ and @file@ but it seems that consumers with groups
-- can only support @broker@. Which is good and enough.
setOffsetStore :: KafkaTopic -> OffsetStoreMethod -> IO ()
setOffsetStore t o =
    let setValue = setTopicValue t
    in  case o of
          OffsetStoreBroker ->
              setValue "offset.store.method" "broker"

          OffsetStoreFile path sync -> do
              setValue "offset.store.method" "file"
              setValue "offset.store.file" path
              setValue "offset.store.sync.interval.ms" (show $ offsetSyncToInt sync)