franz

-module(franz).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/franz.gleam").
-export([new/2, with_config/2, named_client/1, start/1, stop_client/1, create_topic/6, fetch/5, delete_topics/3, supervised/1]).
-export_type([client/0, franz_error/0, kafka_message/0, time_stamp_type/0, endpoint/0, fetch_option/0, client_config/0, consumer_group/0, builder/0, message/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 client() :: {client, gleam@erlang@process:name(message())}.

-type franz_error() :: unknown_error |
    client_down |
    unknown_topic_or_partition |
    producer_down |
    topic_already_exists |
    {consumer_not_found, binary(), integer()} |
    {producer_not_found, binary(), integer()} |
    offset_out_of_range |
    corrupt_message |
    invalid_fetch_size |
    leader_not_available |
    not_leader_or_follower |
    request_timed_out |
    broker_not_available |
    replica_not_available |
    message_too_large |
    network_exception |
    coordinator_load_in_progress |
    coordinator_not_available |
    not_coordinator |
    illegal_generation |
    inconsistent_group_protocol |
    invalid_group_id |
    unknown_member_id |
    invalid_session_timeout |
    rebalance_in_progress |
    invalid_commit_offset_size |
    topic_authorization_failed |
    group_authorization_failed |
    cluster_authorization_failed |
    invalid_topic |
    record_list_too_large |
    not_enough_replicas |
    not_enough_replicas_after_append |
    invalid_required_acks |
    invalid_timestamp |
    invalid_partitions |
    invalid_replication_factor |
    invalid_replica_assignment |
    invalid_config |
    unsupported_sasl_mechanism |
    illegal_sasl_state |
    unsupported_version |
    stale_controller_epoch |
    offset_metadata_too_large |
    not_controller.

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

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

-opaque builder() :: {builder,
        list(endpoint()),
        list(client_config()),
        gleam@erlang@process:name(message())}.

-type message() :: any().

-file("src/franz.gleam", 170).
?DOC(
    " Creates a new client builder with the given bootstrap endpoints.\n"
    " The bootstrap endpoints are the initial Kafka brokers to connect to.\n"
    " The name parameter is used to identify the client process.\n"
).
-spec new(list(endpoint()), gleam@erlang@process:name(message())) -> builder().
new(Bootstrap_endpoints, Name) ->
    {builder, Bootstrap_endpoints, [], Name}.

-file("src/franz.gleam", 179).
?DOC(
    " Adds a client configuration option to the client builder.\n"
    " Multiple configurations can be chained together.\n"
).
-spec with_config(builder(), client_config()) -> builder().
with_config(Client_builder, Client_config) ->
    {builder,
        erlang:element(2, Client_builder),
        [Client_config | erlang:element(3, Client_builder)],
        erlang:element(4, Client_builder)}.

-file("src/franz.gleam", 203).
?DOC(
    " Gets a client reference from a process name.\n"
    " Useful when you need to reference an existing named client.\n"
).
-spec named_client(gleam@erlang@process:name(message())) -> client().
named_client(Name) ->
    {client, Name}.

-file("src/franz.gleam", 188).
?DOC(
    " Starts a new Franz client with the configured settings.\n"
    " Returns an actor.StartResult that contains the client on success.\n"
).
-spec start(builder()) -> {ok, gleam@otp@actor:started(client())} |
    {error, gleam@otp@actor:start_error()}.
start(Client_builder) ->
    case franz_ffi:start_client(
        erlang:element(2, Client_builder),
        erlang:element(3, Client_builder),
        erlang:element(4, Client_builder)
    ) of
        {ok, Pid} ->
            {ok,
                {started, Pid, named_client(erlang:element(4, Client_builder))}};

        {error, Error} ->
            {error, {init_exited, {abnormal, Error}}}
    end.

-file("src/franz.gleam", 209).
?DOC(" Stops a client.\n").
-spec stop_client(client()) -> nil.
stop_client(Client) ->
    franz_ffi:stop_client(Client).

-file("src/franz.gleam", 213).
?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", 226).
?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(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", 235).
-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).

-file("src/franz.gleam", 243).
?DOC(
    " Creates a supervised worker for the Franz client.\n"
    " This can be used with Gleam's OTP supervision trees to ensure the client is restarted on failure.\n"
).
-spec supervised(builder()) -> gleam@otp@supervision:child_specification(client()).
supervised(Builder) ->
    gleam@otp@supervision:worker(fun() -> start(Builder) end).