telega

-module(telega@idempotency).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/telega/idempotency.gleam").
-export([deduplicate/2]).

-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(
    " Webhook/update idempotency: drop updates Telegram re-delivers.\n"
    "\n"
    " Telegram retries an update (same `update_id`) when it does not receive a\n"
    " `200` in time — on a slow response, a redeploy, or a network blip. Without\n"
    " deduplication that means a command runs twice, which is a real problem for\n"
    " non-idempotent actions (sending an invoice, charging Stars, granting a\n"
    " reward).\n"
    "\n"
    " `deduplicate` builds a [pre-router middleware](telega.html#use_pre_handler)\n"
    " that remembers each `update_id` in a [`KeyValueStorage`](telega/storage.html)\n"
    " for a TTL window and stops any update whose id was already seen. Because\n"
    " pre-router middleware runs sequentially inside the single bot actor, the\n"
    " read-then-write below is race-free even under concurrent re-delivery.\n"
    "\n"
    " ```gleam\n"
    " import telega\n"
    " import telega/idempotency\n"
    " import telega/storage/ets\n"
    "\n"
    " let assert Ok(store) = ets.new(name: \"telega_dedup\")\n"
    "\n"
    " telega.new(token:, url:, webhook_path:, secret_token:)\n"
    " |> telega.use_pre_handler(idempotency.deduplicate(\n"
    "   storage: store,\n"
    "   // Keep ids for an hour — comfortably longer than Telegram's retry window.\n"
    "   ttl_ms: 3600_000,\n"
    " ))\n"
    " |> telega.with_router(router)\n"
    " |> telega.init()\n"
    " ```\n"
    "\n"
    " Use a persistent backend (Postgres/SQLite/Redis) when you run more than one\n"
    " bot node or want dedup to survive a restart — the in-memory `ets` store is\n"
    " per-node and cleared on VM restart.\n"
).

-file("src/telega/idempotency.gleam", 50).
?DOC(
    " Build a pre-router middleware that drops updates whose `update_id` was\n"
    " already processed within the last `ttl_ms` milliseconds.\n"
    "\n"
    " On a storage error the update is let through (fail-open): processing an\n"
    " update twice is recoverable, silently dropping a legitimate one is not.\n"
).
-spec deduplicate(telega@storage:key_value_storage(any()), integer()) -> fun((telega@bot:pre_context(any())) -> telega@bot:pre_router_result()).
deduplicate(Storage, Ttl_ms) ->
    fun(Pre_ctx) ->
        Update_id = erlang:element(
            2,
            telega@update:raw(erlang:element(2, Pre_ctx))
        ),
        Key = <<"dedup:"/utf8, (erlang:integer_to_binary(Update_id))/binary>>,
        case (erlang:element(2, Storage))(Key) of
            {ok, {some, _}} ->
                stop;

            {ok, none} ->
                _ = (erlang:element(4, Storage))(Key, <<"1"/utf8>>, Ttl_ms),
                continue;

            {error, _} ->
                continue
        end
    end.