-module(gleeth@event_watcher). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/gleeth/event_watcher.gleam"). -export([default_config/0, 'receive'/2, stop/1, start_with_config/4, start/3, start_for_event/4]). -export_type([contract_event/0, event_watcher_config/0, event_watcher/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( " Real-time contract event streaming.\n" "\n" " Combines the block watcher with ABI event decoding to deliver typed\n" " contract events as they happen. Internally spawns a block watcher,\n" " queries logs for each new block, decodes them, and forwards matches\n" " to the caller.\n" "\n" " ## Examples\n" "\n" " ```gleam\n" " let assert Ok(abi) = json.parse_abi(erc20_abi_json)\n" " let assert Ok(ew) = event_watcher.start(provider, usdc_address, abi)\n" "\n" " // Receive decoded events as they arrive\n" " case event_watcher.receive(ew, 30_000) {\n" " Ok(event) -> // event.name, event.params, event.log\n" " Error(Nil) -> // timeout\n" " }\n" "\n" " event_watcher.stop(ew)\n" " ```\n" ). -type contract_event() :: {contract_event, binary(), list({binary(), gleeth@ethereum@abi@types:abi_value()}), gleeth@ethereum@types:log()}. -type event_watcher_config() :: {event_watcher_config, integer(), binary()}. -opaque event_watcher() :: {event_watcher, gleam@erlang@process:subject(contract_event()), gleam@erlang@process:pid_(), gleeth@watcher:watcher()}. -file("src/gleeth/event_watcher.gleam", 55). ?DOC(" Default config: poll every 2s, no event name filter.\n"). -spec default_config() -> event_watcher_config(). default_config() -> {event_watcher_config, 2000, <<""/utf8>>}. -file("src/gleeth/event_watcher.gleam", 168). ?DOC( " Receive the next decoded event, blocking up to `timeout_ms` milliseconds.\n" "\n" " ## Examples\n" "\n" " ```gleam\n" " case event_watcher.receive(ew, 10_000) {\n" " Ok(event) -> // event.name, event.params\n" " Error(Nil) -> // timeout\n" " }\n" " ```\n" ). -spec 'receive'(event_watcher(), integer()) -> {ok, contract_event()} | {error, nil}. 'receive'(Ew, Timeout_ms) -> gleam@erlang@process:'receive'(erlang:element(2, Ew), Timeout_ms). -file("src/gleeth/event_watcher.gleam", 173). ?DOC(" Stop the event watcher and its underlying block watcher.\n"). -spec stop(event_watcher()) -> nil. stop(Ew) -> gleeth@watcher:stop(erlang:element(4, Ew)), gleam@erlang@process:kill(erlang:element(3, Ew)). -file("src/gleeth/event_watcher.gleam", 246). -spec forward_events( list(gleeth@events:event_result()), gleam@erlang@process:subject(contract_event()) ) -> nil. forward_events(Results, Subject) -> case Results of [] -> nil; [Result | Rest] -> case Result of {decoded, Event, Log} -> gleam@erlang@process:send( Subject, {contract_event, erlang:element(2, Event), erlang:element(3, Event), Log} ); {unknown, _} -> nil end, forward_events(Rest, Subject) end. -file("src/gleeth/event_watcher.gleam", 277). -spec compute_event_topic(list(gleeth@ethereum@abi@json:abi_entry()), binary()) -> binary(). compute_event_topic(Abi, Event_name) -> Found = gleam@list:find(Abi, fun(Entry) -> case Entry of {event_entry, Name, _} -> Name =:= Event_name; _ -> false end end), case Found of {ok, {event_entry, Name@1, Inputs}} -> Types = gleam@list:map(Inputs, fun(P) -> erlang:element(3, P) end), gleeth@ethereum@abi@events:event_topic_hex(Name@1, Types); _ -> <<""/utf8>> end. -file("src/gleeth/event_watcher.gleam", 182). -spec event_loop( gleeth@provider:provider(), binary(), list(gleeth@ethereum@abi@json:abi_entry()), binary(), gleam@erlang@process:subject(gleeth@watcher:block_event()), gleam@erlang@process:subject(contract_event()) ) -> nil. event_loop(Provider, Address, Abi, Event_name, Block_events, Event_subject) -> case gleam@erlang@process:'receive'(Block_events, 60000) of {ok, {new_block, Block_number, _}} -> Log_result = case Event_name of <<""/utf8>> -> gleeth@rpc@methods:get_logs( Provider, Block_number, Block_number, Address, [] ); _ -> Topic0 = compute_event_topic(Abi, Event_name), Topics = case Topic0 of <<""/utf8>> -> []; T -> [T] end, gleeth@rpc@methods:get_logs( Provider, Block_number, Block_number, Address, Topics ) end, case Log_result of {ok, Logs} -> Decoded = gleeth@events:decode_logs(Logs, Abi), forward_events(Decoded, Event_subject); {error, _} -> nil end, event_loop( Provider, Address, Abi, Event_name, Block_events, Event_subject ); {error, _} -> event_loop( Provider, Address, Abi, Event_name, Block_events, Event_subject ) end. -file("src/gleeth/event_watcher.gleam", 107). ?DOC(" Start watching with custom configuration.\n"). -spec start_with_config( gleeth@provider:provider(), binary(), list(gleeth@ethereum@abi@json:abi_entry()), event_watcher_config() ) -> {ok, event_watcher()} | {error, binary()}. start_with_config(Provider, Address, Abi, Config) -> Event_subject = gleam@erlang@process:new_subject(), Watcher_config = {watcher_config, erlang:element(2, Config)}, Startup = gleam@erlang@process:new_subject(), _ = proc_lib:spawn( fun() -> Block_events = gleam@erlang@process:new_subject(), case gleeth@watcher:start_with_subject( Provider, Watcher_config, Block_events ) of {ok, Block_watcher} -> Ew = {event_watcher, Event_subject, erlang:self(), Block_watcher}, gleam@erlang@process:send(Startup, {ok, Ew}), event_loop( Provider, Address, Abi, erlang:element(3, Config), Block_events, Event_subject ); {error, Msg} -> gleam@erlang@process:send( Startup, {error, <<"Failed to start block watcher: "/utf8, Msg/binary>>} ) end end ), case gleam@erlang@process:'receive'(Startup, 5000) of {ok, Result} -> Result; {error, _} -> {error, <<"Event watcher startup timed out"/utf8>>} end. -file("src/gleeth/event_watcher.gleam", 75). ?DOC( " Start watching for all contract events with default config.\n" "\n" " ## Examples\n" "\n" " ```gleam\n" " let assert Ok(ew) = event_watcher.start(provider, \"0xA0b8...\", abi)\n" " ```\n" ). -spec start( gleeth@provider:provider(), binary(), list(gleeth@ethereum@abi@json:abi_entry()) ) -> {ok, event_watcher()} | {error, binary()}. start(Provider, Address, Abi) -> start_with_config(Provider, Address, Abi, default_config()). -file("src/gleeth/event_watcher.gleam", 92). ?DOC( " Start watching for a specific event by name.\n" "\n" " ## Examples\n" "\n" " ```gleam\n" " let assert Ok(ew) = event_watcher.start_for_event(\n" " provider, \"0xA0b8...\", abi, \"Transfer\",\n" " )\n" " ```\n" ). -spec start_for_event( gleeth@provider:provider(), binary(), list(gleeth@ethereum@abi@json:abi_entry()), binary() ) -> {ok, event_watcher()} | {error, binary()}. start_for_event(Provider, Address, Abi, Event_name) -> start_with_config( Provider, Address, Abi, {event_watcher_config, 2000, Event_name} ).