Current section
Files
Jump to
Current section
Files
src/telega@polling.erl
-module(telega@polling).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/telega/polling.gleam").
-export([calculate_new_offset/2, stop/1, get_status/1, get_config_info/1, is_running/1, wait_finish/1, supervised/7, start_polling/6, start_polling_default/2, start_polling_with_offset/7, start_polling_with_notify/7]).
-export_type([polling_config/0, poller/0, poller_status/0, polling_message/0, polling_state/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.
?MODULEDOC(
" Long polling implementation for Telegram Bot API.\n"
"\n"
" This module provides long polling as an alternative to webhooks for receiving updates.\n"
" A polling worker actor continuously fetches updates from Telegram and dispatches them\n"
" to the bot's message handlers.\n"
"\n"
" ## Supervised mode (recommended)\n"
"\n"
" When using `telega.init_for_polling()`, the polling worker is automatically started\n"
" inside the supervision tree as a `Permanent` child. No manual setup is needed:\n"
"\n"
" ```gleam\n"
" let assert Ok(_bot) =\n"
" telega.new_for_polling(token: \"BOT_TOKEN\")\n"
" |> telega.with_router(router)\n"
" |> telega.init_for_polling_nil_session()\n"
"\n"
" process.sleep_forever()\n"
" ```\n"
"\n"
" Use `telega.with_polling_config()` on the builder to customize timeout, limit,\n"
" and poll interval before calling `init_for_polling()`.\n"
"\n"
" ## Manual mode\n"
"\n"
" For advanced use cases (custom offsets, separate lifecycle management),\n"
" start polling manually:\n"
"\n"
" ```gleam\n"
" let assert Ok(poller) =\n"
" polling.start_polling_default(\n"
" client: telega.get_api_config(bot),\n"
" bot: telega.get_bot_subject_internal(bot),\n"
" )\n"
" ```\n"
"\n"
" ## Error handling\n"
"\n"
" The polling worker uses exponential backoff for transient errors (network issues,\n"
" rate limits, server errors). Critical errors (invalid token, bot deleted, or\n"
" 10+ consecutive failures) stop polling and invoke the optional `on_stop` callback.\n"
).
-type polling_config() :: {polling_config,
telega@client:telegram_client(),
gleam@erlang@process:subject(telega@bot:bot_message()),
integer(),
integer(),
list(binary()),
integer(),
gleam@option:option(fun((telega@error:telega_error()) -> nil))}.
-opaque poller() :: {poller,
gleam@erlang@process:subject(polling_message()),
polling_config(),
poller_status()}.
-type poller_status() :: starting | running | stopped | {failed, binary()}.
-type polling_message() :: {start_polling, gleam@option:option(integer())} |
stop_polling |
{poll_next, integer()} |
{inject_updates, list(telega@model@types:update()), integer()} |
{set_self, gleam@erlang@process:subject(polling_message())} |
{handle_error, telega@error:telega_error(), integer()}.
-type polling_state() :: {polling_state,
polling_config(),
gleam@option:option(integer()),
boolean(),
gleam@option:option(gleam@erlang@process:subject(polling_message())),
integer()}.
-file("src/telega/polling.gleam", 93).
?DOC(" Internal function to create default polling configuration\n").
-spec create_config(
telega@client:telegram_client(),
gleam@erlang@process:subject(telega@bot:bot_message()),
integer(),
integer(),
list(binary()),
integer(),
gleam@option:option(fun((telega@error:telega_error()) -> nil))
) -> polling_config().
create_config(
Client,
Bot,
Timeout,
Limit,
Allowed_updates,
Poll_interval,
On_stop
) ->
{polling_config,
Client,
Bot,
Timeout,
Limit,
Allowed_updates,
Poll_interval,
On_stop}.
-file("src/telega/polling.gleam", 404).
?DOC(" Calculate the next offset based on received updates\n").
-spec calculate_new_offset(list(telega@model@types:update()), integer()) -> integer().
calculate_new_offset(Updates, Current_offset) ->
case Updates of
[] ->
Current_offset;
_ ->
case gleam@list:last(Updates) of
{ok, Update} ->
erlang:element(2, Update) + 1;
{error, _} ->
Current_offset
end
end.
-file("src/telega/polling.gleam", 418).
?DOC(
" Process a single update by converting and sending to bot\n"
" Handles errors gracefully to prevent polling from stopping\n"
).
-spec process_update(
gleam@erlang@process:subject(telega@bot:bot_message()),
telega@model@types:update()
) -> nil.
process_update(Bot_subject, Update) ->
Decoded_update = telega@update:raw_to_update(Update),
telega@bot:handle_update(Bot_subject, Decoded_update),
nil.
-file("src/telega/polling.gleam", 381).
?DOC(" Poll for updates and process them\n").
-spec poll_updates(polling_state(), integer()) -> {ok, integer()} |
{error, telega@error:telega_error()}.
poll_updates(State, Offset) ->
Parameters = {get_updates_parameters,
{some, Offset},
{some, erlang:element(5, erlang:element(2, State))},
{some, erlang:element(4, erlang:element(2, State))},
case erlang:element(6, erlang:element(2, State)) of
[] ->
none;
Updates ->
{some, Updates}
end},
gleam@result:'try'(
telega@api:get_updates(
erlang:element(2, erlang:element(2, State)),
{some, Parameters}
),
fun(Updates@1) ->
gleam@list:each(
Updates@1,
fun(Update) ->
process_update(
erlang:element(3, erlang:element(2, State)),
Update
)
end
),
{ok, calculate_new_offset(Updates@1, Offset)}
end
).
-file("src/telega/polling.gleam", 425).
?DOC(" Schedule the next poll\n").
-spec schedule_next_poll(polling_state(), integer()) -> nil.
schedule_next_poll(State, Offset) ->
Delay = case erlang:element(4, erlang:element(2, State)) of
0 ->
erlang:element(7, erlang:element(2, State));
_ ->
10
end,
case erlang:element(5, State) of
{some, Self} ->
gleam@erlang@process:send_after(Self, Delay, {poll_next, Offset}),
nil;
none ->
nil
end.
-file("src/telega/polling.gleam", 549).
?DOC(" Stop polling\n").
-spec stop(poller()) -> nil.
stop(Poller) ->
gleam@erlang@process:send(erlang:element(2, Poller), stop_polling).
-file("src/telega/polling.gleam", 554).
?DOC(" Get the current status of the poller\n").
-spec get_status(poller()) -> poller_status().
get_status(Poller) ->
erlang:element(4, Poller).
-file("src/telega/polling.gleam", 559).
?DOC(" Get the polling configuration metadata\n").
-spec get_config_info(poller()) -> {integer(),
integer(),
list(binary()),
integer()}.
get_config_info(Poller) ->
{erlang:element(4, erlang:element(3, Poller)),
erlang:element(5, erlang:element(3, Poller)),
erlang:element(6, erlang:element(3, Poller)),
erlang:element(7, erlang:element(3, Poller))}.
-file("src/telega/polling.gleam", 569).
?DOC(" Check if poller is running\n").
-spec is_running(poller()) -> boolean().
is_running(Poller) ->
case erlang:element(4, Poller) of
running ->
true;
_ ->
false
end.
-file("src/telega/polling.gleam", 578).
?DOC(
" Wait for the poller to finish\n"
" This function blocks indefinitely until the polling worker stops\n"
).
-spec wait_finish(poller()) -> nil.
wait_finish(Poller) ->
case gleam@erlang@process:subject_owner(erlang:element(2, Poller)) of
{ok, Pid} ->
Monitor = gleam@erlang@process:monitor(Pid),
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
gleam@erlang@process:select_specific_monitor(
_pipe,
Monitor,
fun(_) -> nil end
)
end,
gleam_erlang_ffi:select(Selector);
{error, _} ->
gleam_erlang_ffi:sleep_forever()
end.
-file("src/telega/polling.gleam", 208).
?DOC(" Handle messages in the polling worker\n").
-spec handle_polling_message(polling_state(), polling_message()) -> gleam@otp@actor:next(polling_state(), polling_message()).
handle_polling_message(State, Message) ->
case Message of
{set_self, Subject} ->
gleam@otp@actor:continue(
{polling_state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
{some, Subject},
erlang:element(6, State)}
);
{start_polling, Offset} ->
New_state = {polling_state,
erlang:element(2, State),
Offset,
true,
erlang:element(5, State),
erlang:element(6, State)},
case erlang:element(5, State) of
{some, Self} ->
gleam@erlang@process:send(
Self,
{poll_next, gleam@option:unwrap(Offset, 0)}
);
none ->
telega@internal@log:error(
<<"No self reference available for polling"/utf8>>
)
end,
gleam@otp@actor:continue(New_state);
stop_polling ->
gleam@otp@actor:continue(
{polling_state,
erlang:element(2, State),
erlang:element(3, State),
false,
erlang:element(5, State),
erlang:element(6, State)}
);
{poll_next, Offset@1} ->
case erlang:element(4, State) of
false ->
gleam@otp@actor:continue(State);
true ->
case poll_updates(State, Offset@1) of
{ok, New_offset} ->
schedule_next_poll(State, New_offset),
gleam@otp@actor:continue(
{polling_state,
erlang:element(2, State),
{some, New_offset},
erlang:element(4, State),
erlang:element(5, State),
0}
);
{error, Error} ->
case erlang:element(5, State) of
{some, Self@1} ->
gleam@erlang@process:send(
Self@1,
{handle_error, Error, Offset@1}
);
none ->
nil
end,
gleam@otp@actor:continue(State)
end
end;
{inject_updates, Updates, Offset@2} ->
gleam@list:each(
Updates,
fun(Update) ->
process_update(
erlang:element(3, erlang:element(2, State)),
Update
)
end
),
New_offset@1 = calculate_new_offset(Updates, Offset@2),
gleam@otp@actor:continue(
{polling_state,
erlang:element(2, State),
{some, New_offset@1},
erlang:element(4, State),
erlang:element(5, State),
erlang:element(6, State)}
);
{handle_error, Error@1, Offset@3} ->
New_consecutive_errors = erlang:element(6, State) + 1,
Should_stop = case Error@1 of
{telegram_api_error, 401, _} ->
true;
{telegram_api_error, 404, _} ->
true;
_ when New_consecutive_errors >= 10 ->
true;
{fetch_error, _} ->
false;
{telegram_api_error, 429, _} ->
false;
{telegram_api_error, 500, _} ->
false;
{telegram_api_error, 502, _} ->
false;
{telegram_api_error, 503, _} ->
false;
{json_decode_error, _} ->
false;
_ ->
true
end,
case Should_stop of
true ->
telega@internal@log:error(
<<<<<<<<"Critical polling error (consecutive: "/utf8,
(gleam@string:inspect(
New_consecutive_errors
))/binary>>/binary,
"): "/utf8>>/binary,
(telega@error:to_string(Error@1))/binary>>/binary,
" - stopping polling"/utf8>>
),
case erlang:element(8, erlang:element(2, State)) of
{some, Callback} ->
Callback(Error@1);
none ->
nil
end,
gleam@otp@actor:stop();
false ->
Is_timeout = case Error@1 of
{fetch_error, Msg} ->
gleam_stdlib:contains_string(
Msg,
<<"ResponseTimeout"/utf8>>
)
orelse gleam_stdlib:contains_string(
Msg,
<<"Timeout"/utf8>>
);
_ ->
false
end,
case Is_timeout of
true ->
case New_consecutive_errors > 10 of
true ->
telega@internal@log:warning(
<<<<"Persistent timeout issues detected (count: "/utf8,
(gleam@string:inspect(
New_consecutive_errors
))/binary>>/binary,
")"/utf8>>
);
false ->
nil
end;
false ->
telega@internal@log:error(
<<<<<<<<"Polling error (consecutive: "/utf8,
(gleam@string:inspect(
New_consecutive_errors
))/binary>>/binary,
"): "/utf8>>/binary,
(telega@error:to_string(Error@1))/binary>>/binary,
" - retrying"/utf8>>
)
end,
Delay = case New_consecutive_errors of
N when N =< 3 ->
1000;
N@1 when N@1 =< 6 ->
5000;
_ ->
10000
end,
case erlang:element(5, State) of
{some, Self@2} ->
gleam@erlang@process:send_after(
Self@2,
Delay,
{poll_next, Offset@3}
),
nil;
none ->
nil
end,
gleam@otp@actor:continue(
{polling_state,
erlang:element(2, State),
erlang:element(3, State),
erlang:element(4, State),
erlang:element(5, State),
New_consecutive_errors}
)
end
end.
-file("src/telega/polling.gleam", 135).
?DOC(" Start the polling worker actor\n").
-spec start_polling_worker(polling_config()) -> {ok,
gleam@erlang@process:subject(polling_message())} |
{error, gleam@otp@actor:start_error()}.
start_polling_worker(Config) ->
Initial_state = {polling_state, Config, none, false, none, 0},
gleam@result:'try'(
begin
_pipe = gleam@otp@actor:new(Initial_state),
_pipe@1 = gleam@otp@actor:on_message(
_pipe,
fun handle_polling_message/2
),
gleam@otp@actor:start(_pipe@1)
end,
fun(Started) ->
gleam@erlang@process:send(
erlang:element(3, Started),
{set_self, erlang:element(3, Started)}
),
{ok, erlang:element(3, Started)}
end
).
-file("src/telega/polling.gleam", 160).
?DOC(
" Create a `ChildSpecification` for running polling inside a supervision tree.\n"
" The polling worker will automatically delete the webhook and start polling.\n"
).
-spec supervised(
telega@client:telegram_client(),
gleam@erlang@process:subject(telega@bot:bot_message()),
integer(),
integer(),
list(binary()),
integer(),
gleam@option:option(fun((telega@error:telega_error()) -> nil))
) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(polling_message())).
supervised(Client, Bot, Timeout, Limit, Allowed_updates, Poll_interval, On_stop) ->
_pipe@3 = gleam@otp@supervision:worker(
fun() ->
Config = create_config(
Client,
Bot,
Timeout,
Limit,
Allowed_updates,
Poll_interval,
On_stop
),
gleam@result:'try'(
begin
_pipe = telega@api:delete_webhook(erlang:element(2, Config)),
gleam@result:map_error(
_pipe,
fun(Err) ->
{init_failed, telega@error:to_string(Err)}
end
)
end,
fun(_) ->
Initial_state = {polling_state,
Config,
none,
false,
none,
0},
gleam@result:'try'(
begin
_pipe@1 = gleam@otp@actor:new(Initial_state),
_pipe@2 = gleam@otp@actor:on_message(
_pipe@1,
fun handle_polling_message/2
),
gleam@otp@actor:start(_pipe@2)
end,
fun(Started) ->
gleam@erlang@process:send(
erlang:element(3, Started),
{set_self, erlang:element(3, Started)}
),
gleam@erlang@process:send(
erlang:element(3, Started),
{start_polling, none}
),
{ok, Started}
end
)
end
)
end
),
gleam@otp@supervision:restart(_pipe@3, permanent).
-file("src/telega/polling.gleam", 441).
?DOC(" Internal function to start polling with given configuration and offset\n").
-spec start_polling_internal(polling_config(), gleam@option:option(integer())) -> {ok,
poller()} |
{error, telega@error:telega_error()}.
start_polling_internal(Config, Offset) ->
gleam@result:'try'(
telega@api:delete_webhook(erlang:element(2, Config)),
fun(_) ->
gleam@result:'try'(
begin
_pipe = start_polling_worker(Config),
gleam@result:map_error(
_pipe,
fun(Err) ->
{actor_error,
<<"Failed to start polling worker: "/utf8,
(gleam@string:inspect(Err))/binary>>}
end
)
end,
fun(Worker) ->
gleam@erlang@process:send(Worker, {start_polling, Offset}),
{ok, {poller, Worker, Config, starting}}
end
)
end
).
-file("src/telega/polling.gleam", 462).
?DOC(" Start polling with the given client and bot subject.\n").
-spec start_polling(
telega@client:telegram_client(),
gleam@erlang@process:subject(telega@bot:bot_message()),
integer(),
integer(),
list(binary()),
integer()
) -> {ok, poller()} | {error, telega@error:telega_error()}.
start_polling(Client, Bot, Timeout, Limit, Allowed_updates, Poll_interval) ->
Config = create_config(
Client,
Bot,
Timeout,
Limit,
Allowed_updates,
Poll_interval,
none
),
start_polling_internal(Config, none).
-file("src/telega/polling.gleam", 485).
?DOC(" Start polling with default configuration.\n").
-spec start_polling_default(
telega@client:telegram_client(),
gleam@erlang@process:subject(telega@bot:bot_message())
) -> {ok, poller()} | {error, telega@error:telega_error()}.
start_polling_default(Client, Bot) ->
start_polling(Client, Bot, 30, 100, [], 1000).
-file("src/telega/polling.gleam", 500).
?DOC(" Start polling with a custom offset.\n").
-spec start_polling_with_offset(
telega@client:telegram_client(),
gleam@erlang@process:subject(telega@bot:bot_message()),
integer(),
integer(),
integer(),
list(binary()),
integer()
) -> {ok, poller()} | {error, telega@error:telega_error()}.
start_polling_with_offset(
Client,
Bot,
Offset,
Timeout,
Limit,
Allowed_updates,
Poll_interval
) ->
Config = create_config(
Client,
Bot,
Timeout,
Limit,
Allowed_updates,
Poll_interval,
none
),
start_polling_internal(Config, {some, Offset}).
-file("src/telega/polling.gleam", 525).
?DOC(
" Start polling with a notification callback for when polling stops due to errors.\n"
" The callback will be invoked with the error that caused polling to stop.\n"
).
-spec start_polling_with_notify(
telega@client:telegram_client(),
gleam@erlang@process:subject(telega@bot:bot_message()),
integer(),
integer(),
list(binary()),
integer(),
fun((telega@error:telega_error()) -> nil)
) -> {ok, poller()} | {error, telega@error:telega_error()}.
start_polling_with_notify(
Client,
Bot,
Timeout,
Limit,
Allowed_updates,
Poll_interval,
On_stop
) ->
Config = create_config(
Client,
Bot,
Timeout,
Limit,
Allowed_updates,
Poll_interval,
{some, On_stop}
),
start_polling_internal(Config, none).