franz

-module(franz).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/franz.gleam").
-export([new/1, with_config/2, start/1, stop_client/1, create_topic/6, fetch/5, delete_topics/3]).
-export_type([franz_client/0, franz_error/0, kafka_message/0, time_stamp_type/0, endpoint/0, fetch_option/0, client_config/0, consumer_group/0, client_builder/0]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

-type franz_client() :: any().

-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()} |
    offset_out_of_range.

-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 endpoint() :: {endpoint, binary(), integer()}.

-type fetch_option() :: {max_wait_time, integer()} |
    {min_bytes, integer()} |
    {max_bytes, integer()} |
    {isolation_level, franz@isolation_level:isolation_level()}.

-type client_config() :: {restart_delay_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()}.

-type consumer_group() :: {consumer_group, binary(), binary()}.

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

-file("src/franz.gleam", 103).
?DOC(" Create a new client builder with the given bootstrap endpoints.\n").
-spec new(list(endpoint())) -> client_builder().
new(Bootstrap_endpoints) ->
    {client_builder, Bootstrap_endpoints, []}.

-file("src/franz.gleam", 108).
?DOC(" Add a client configuration to the client builder.\n").
-spec with_config(client_builder(), client_config()) -> client_builder().
with_config(Client_builder, Client_config) ->
    {client_builder,
        erlang:element(2, Client_builder),
        [Client_config | erlang:element(3, Client_builder)]}.

-file("src/franz.gleam", 119).
?DOC(" Start a new client with the given configuration.\n").
-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", 125).
?DOC(" Stops a client.\n").
-spec stop_client(franz_client()) -> nil.
stop_client(Client) ->
    franz_ffi:stop_client(Client).

-file("src/franz.gleam", 129).
?DOC(" Create a new topic with the given number of partitions and replication factor.\n").
-spec create_topic(
    list(endpoint()),
    binary(),
    integer(),
    integer(),
    list({binary(), binary()}),
    integer()
) -> {ok, nil} | {error, franz_error()}.
create_topic(Endpoints, Name, Partitions, Replication_factor, Configs, Timeout) ->
    franz_ffi:create_topic(
        Endpoints,
        Name,
        Partitions,
        Replication_factor,
        Configs,
        Timeout
    ).

-file("src/franz.gleam", 142).
?DOC(
    " Fetch a single message set from the given topic-partition.\n"
    " On success, the function returns the messages along with the last stable offset (when using ReadCommited mode, the last committed offset) or the high watermark offset (offset of the last message that was successfully copied to all replicas, incremented by 1), whichever is lower. \n"
    " In essence, this is the offset up to which it was possible to read the messages at the time of fetching\n"
).
-spec fetch(
    franz_client(),
    binary(),
    integer(),
    integer(),
    list(fetch_option())
) -> {ok, {integer(), kafka_message()}} | {error, franz_error()}.
fetch(Client, Topic, Partition, Offset, Fetch_options) ->
    franz_ffi:fetch(Client, Topic, Partition, Offset, Fetch_options).

-file("src/franz.gleam", 151).
-spec delete_topics(list(endpoint()), list(binary()), integer()) -> {ok, nil} |
    {error, franz_error()}.
delete_topics(Endpoints, Names, Timeout) ->
    franz_ffi:delete_topics(Endpoints, Names, Timeout).