-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.