franz

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

-export([start_client/2, produce_sync_offset/5, produce_sync/5, produce/5, create_topic/4, start_topic_subscriber/7, ack_return/1, start_consumer/3, stop_client/1, produce_cb/6, fetch/4]).
-export_type([franz_error/0, franz_client/0, ack/1, kafka_message/0, time_stamp_type/0, consumer_config/0, offset_time/0, isolation_level/0, offset_reset_policy/0, client_config/0, producer_config/0, compression/0, producer_partition/0, partitioner/0, value/0, consumer_partition/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 ack(FOA) :: any() | {gleam_phantom, FOA}.

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

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

-type consumer_config() :: {begin_offset, offset_time()} |
    {min_bytes, integer()} |
    {max_bytes, integer()} |
    {max_wait_time, integer()} |
    {sleep_timeout, integer()} |
    {prefetch_count, integer()} |
    {prefetch_bytes, integer()} |
    {offset_reset_policy, offset_reset_policy()} |
    {size_stat_window, integer()} |
    {isolation_level, isolation_level()} |
    {share_leader_conn, boolean()}.

-type offset_time() :: earliest | latest | {message_timestamp, integer()}.

-type isolation_level() :: read_committed | read_uncommitted.

-type offset_reset_policy() :: reset_by_subscriber |
    reset_to_earliest |
    reset_to_latest.

-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(producer_config())} |
    {unknown_topic_cache_ttl, integer()}.

-type producer_config() :: {required_acks, integer()} |
    {ack_timeout, integer()} |
    {partition_buffer_limit, integer()} |
    {partition_onwire_limit, integer()} |
    {max_batch_size, integer()} |
    {max_retries, integer()} |
    {retry_backoff_ms, integer()} |
    {compression, compression()} |
    {max_linger_ms, integer()} |
    {max_linger_count, integer()}.

-type compression() :: no_compression | gzip | snappy.

-type producer_partition() :: {partition, integer()} |
    {partitioner, partitioner()}.

-type partitioner() :: {partition_fun,
        fun((integer(), integer(), bitstring(), bitstring()) -> {ok, integer()} |
            {error, nil})} |
    random |
    hash.

-type value() :: {value, bitstring(), list({binary(), binary()})} |
    {value_with_timestamp, bitstring(), integer(), list({binary(), binary()})}.

-type consumer_partition() :: {consumer_partitions, list(integer())} | all.

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 120).
-spec start_client(list({binary(), integer()}), list(client_config())) -> {ok,
        franz_client()} |
    {error, franz_error()}.
start_client(Bootstrap_endpoints, Client_config) ->
    franz_ffi:start_client(Bootstrap_endpoints, Client_config).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 126).
-spec produce_sync_offset(
    franz_client(),
    binary(),
    producer_partition(),
    bitstring(),
    value()
) -> {ok, integer()} | {error, franz_error()}.
produce_sync_offset(Client, Topic, Partition, Key, Value) ->
    franz_ffi:produce_sync_offset(Client, Topic, Partition, Key, Value).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 135).
-spec produce_sync(
    franz_client(),
    binary(),
    producer_partition(),
    bitstring(),
    value()
) -> {ok, nil} | {error, franz_error()}.
produce_sync(Client, Topic, Partition, Key, Value) ->
    franz_ffi:produce_sync(Client, Topic, Partition, Key, Value).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 144).
-spec produce(
    franz_client(),
    binary(),
    producer_partition(),
    bitstring(),
    value()
) -> {ok, nil} | {error, franz_error()}.
produce(Client, Topic, Partition, Key, Value) ->
    franz_ffi:produce(Client, Topic, Partition, Key, Value).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 153).
-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("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 161).
-spec start_topic_subscriber(
    franz_client(),
    binary(),
    consumer_partition(),
    list(consumer_config()),
    list({integer(), integer()}),
    fun((integer(), any(), FOR) -> ack(FOR)),
    FOR
) -> {ok, gleam@erlang@process:pid_()} | {error, franz_error()}.
start_topic_subscriber(
    Client,
    Topic,
    Partitions,
    Consumer_config,
    Commited_offsets,
    Callback,
    Init_callback_state
) ->
    franz_ffi:start_topic_subscriber(
        Client,
        Topic,
        Partitions,
        Consumer_config,
        Commited_offsets,
        Callback,
        Init_callback_state
    ).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 172).
-spec ack_return(FOV) -> ack(FOV).
ack_return(Cb_state) ->
    franz_ffi:ack_return(Cb_state).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 175).
-spec start_consumer(franz_client(), binary(), list(consumer_config())) -> {ok,
        nil} |
    {error, franz_error()}.
start_consumer(Client, Topic, Options) ->
    franz_ffi:start_consumer(Client, Topic, Options).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 182).
-spec stop_client(franz_client()) -> nil.
stop_client(Client) ->
    franz_ffi:stop_client(Client).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 185).
-spec produce_cb(
    franz_client(),
    binary(),
    producer_partition(),
    bitstring(),
    bitstring(),
    fun((integer(), integer()) -> any())
) -> {ok, integer()} | {error, franz_error()}.
produce_cb(Client, Topic, Partition, Key, Value, Callback) ->
    franz_ffi:produce_cb(Client, Topic, Partition, Key, Value, Callback).

-file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 195).
-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).