franz

-module(franz).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([new_client/1, with_config/2, start/1, stop_client/1, create_topic/4, fetch/4]).
-export_type([franz_error/0, franz_client/0, kafka_message/0, time_stamp_type/0, client_config/0, client_builder/0]).

-type franz_error() :: unknown_error |
    client_down |
    unknown_topic_or_partition |
    producer_down |
    topic_already_exists |
    {consumer_not_found, binary()} |
    {producer_not_found, binary(), integer()}.

-type franz_client() :: any().

-type kafka_message() :: {kafka_message,
        integer(),
        bitstring(),
        bitstring(),
        time_stamp_type(),
        integer(),
        list({binary(), binary()})} |
    {kafka_message_set, binary(), integer(), integer(), list(kafka_message())}.

-type time_stamp_type() :: undefined | create | append.

-type client_config() :: {restart_delay_seconds, integer()} |
    {get_metadata_timeout_seconds, integer()} |
    {reconnect_cool_down_seconds, integer()} |
    {allow_topic_auto_creation, boolean()} |
    {auto_start_producers, boolean()} |
    {default_producer_config, list(franz@producer_config:producer_config())} |
    {unknown_topic_cache_ttl, integer()}.

-opaque client_builder() :: {client_builder,
        list({binary(), integer()}),
        list(client_config())}.

-file("src/franz.gleam", 71).
-spec new_client(list({binary(), integer()})) -> client_builder().
new_client(Bootstrap_endpoints) ->
    {client_builder, Bootstrap_endpoints, []}.

-file("src/franz.gleam", 75).
-spec with_config(client_builder(), client_config()) -> client_builder().
with_config(Client_builder, Client_config) ->
    _record = Client_builder,
    {client_builder,
        erlang:element(2, _record),
        [Client_config | erlang:element(3, Client_builder)]}.

-file("src/franz.gleam", 85).
-spec start(client_builder()) -> {ok, franz_client()} | {error, franz_error()}.
start(Client_builder) ->
    franz_ffi:start_client(
        erlang:element(2, Client_builder),
        erlang:element(3, Client_builder)
    ).

-file("src/franz.gleam", 90).
-spec stop_client(franz_client()) -> nil.
stop_client(Client) ->
    franz_ffi:stop_client(Client).

-file("src/franz.gleam", 93).
-spec create_topic(list({binary(), integer()}), binary(), integer(), integer()) -> {ok,
        nil} |
    {error, franz_error()}.
create_topic(Bootstrap_endpoints, Topic, Partitions, Replication_factor) ->
    franz_ffi:create_topic(
        Bootstrap_endpoints,
        Topic,
        Partitions,
        Replication_factor
    ).

-file("src/franz.gleam", 101).
-spec fetch(franz_client(), binary(), integer(), integer()) -> {ok,
        {integer(), kafka_message()}} |
    {error, franz_error()}.
fetch(Client, Topic, Partition, Offset) ->
    franz_ffi:fetch(Client, Topic, Partition, Offset).