From 573e1233ac52cc695a96bab4b0cc71837d0eb0f6 Mon Sep 17 00:00:00 2001 From: Sasa Prsic Date: Fri, 27 Feb 2026 10:17:08 +0100 Subject: [PATCH 1/6] Add temp tx store. --- ...658cbc0081a377247a1ca8081cbfca33da68.json} | 4 +- ...ba4b067916690de45dee63b6b2eab15a8ec43.json | 24 + ...bae344e7e5e287b01276197a3a97ad3d690de.json | 23 + ...0daa4487a180d819a85980d6af0b09843de20.json | 22 + ...0a715cf79f2e352838105f09c1eaf2d90aac3.json | 97 +++ ...02e8f22ef09551605a7c4f2cc26d315f77aaa.json | 83 +++ ...5408336193be079337587d1b1ff6267c7bc09.json | 40 ++ ...af73a022e7fecafb4a2eb119b089531c21084.json | 12 + Cargo.lock | 1 + Cargo.toml | 1 + .../0001_create_initiated_transactions.sql | 15 + sql/check_initiated_tx_by_block.sql | 1 + sql/check_initiated_tx_by_hash.sql | 1 + sql/delete_indexed_initiated_tx.sql | 21 + sql/insert_initiated_tx.sql | 5 + sql/query_avail_tx.sql | 4 +- sql/query_claimed_tx.sql | 4 + sql/query_initiated_tx.sql | 17 + src/initiate_transaction.rs | 624 ++++++++++++++++++ src/main.rs | 95 ++- src/models.rs | 32 + 21 files changed, 1116 insertions(+), 10 deletions(-) rename .sqlx/{query-b4e72b9ba8a6b9d8947e2fdd6b58ecad9ad0c9bd398c13fdd879dced0dc81c43.json => query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json} (93%) create mode 100644 .sqlx/query-12f0f248f7ccf430681015c0749ba4b067916690de45dee63b6b2eab15a8ec43.json create mode 100644 .sqlx/query-2768e552e48f41ac350ec5b1d15bae344e7e5e287b01276197a3a97ad3d690de.json create mode 100644 .sqlx/query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json create mode 100644 .sqlx/query-38c33350008ff3a7d58278df6fd0a715cf79f2e352838105f09c1eaf2d90aac3.json create mode 100644 .sqlx/query-aeb96222ec314d46fabb7bb143802e8f22ef09551605a7c4f2cc26d315f77aaa.json create mode 100644 .sqlx/query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json create mode 100644 .sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json create mode 100644 migrations/0001_create_initiated_transactions.sql create mode 100644 sql/check_initiated_tx_by_block.sql create mode 100644 sql/check_initiated_tx_by_hash.sql create mode 100644 sql/delete_indexed_initiated_tx.sql create mode 100644 sql/insert_initiated_tx.sql create mode 100644 sql/query_claimed_tx.sql create mode 100644 sql/query_initiated_tx.sql create mode 100644 src/initiate_transaction.rs diff --git a/.sqlx/query-b4e72b9ba8a6b9d8947e2fdd6b58ecad9ad0c9bd398c13fdd879dced0dc81c43.json b/.sqlx/query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json similarity index 93% rename from .sqlx/query-b4e72b9ba8a6b9d8947e2fdd6b58ecad9ad0c9bd398c13fdd879dced0dc81c43.json rename to .sqlx/query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json index cbc8b56..729b5ac 100644 --- a/.sqlx/query-b4e72b9ba8a6b9d8947e2fdd6b58ecad9ad0c9bd398c13fdd879dced0dc81c43.json +++ b/.sqlx/query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT ai.id AS message_id,\n ai.signature_address AS sender,\n es.to AS \"receiver!\",\n COALESCE(es.amount, '0')::text AS \"amount!\",\n ai.block_hash AS source_block_hash,\n ai.ext_hash AS source_transaction_hash,\n ai.block_height AS source_block_height,\n ai.ext_index AS source_tx_index,\n es.block_timestamp,\n be.source_transaction_hash AS \"destination_tx_hash?: String\",\n COALESCE(\n CASE\n WHEN be.message_id IS NOT NULL THEN 'bridged'::status\n END,\n 'in_progress'::status\n ) ::status AS \"final_status!: BridgeStatusEnum\"\nFROM avail_send_message_table es\n INNER JOIN public.avail_indexer AS ai\n ON ai.id = es.id\n LEFT JOIN public.bridge_event AS be\n ON es.id = be.message_id\nWHERE ai.signature_address = $1\n AND es.type = $2\n AND (be.event_type = $3 or be.event_type is null)\n AND ai.ext_success = $4\nORDER BY es.block_timestamp DESC\nLIMIT 1000;\n", + "query": "SELECT ai.id AS message_id,\n ai.signature_address AS sender,\n es.to AS \"receiver!\",\n COALESCE(es.amount, '0')::text AS \"amount!\",\n ai.block_hash AS source_block_hash,\n ai.ext_hash AS source_transaction_hash,\n ai.block_height AS source_block_height,\n ai.ext_index AS source_tx_index,\n ai.block_timestamp,\n be.source_transaction_hash AS \"destination_tx_hash?: String\",\n COALESCE(\n CASE\n WHEN be.message_id IS NOT NULL THEN 'bridged'::status\n END,\n 'in_progress'::status\n ) ::status AS \"final_status!: BridgeStatusEnum\"\nFROM avail_send_message_table es\n INNER JOIN public.avail_indexer AS ai\n ON ai.id = es.id\n LEFT JOIN public.bridge_event AS be\n ON es.id = be.message_id\nWHERE ai.signature_address = $1\n AND es.type = $2\n AND (be.event_type = $3 or be.event_type is null)\n AND ai.ext_success = $4\nORDER BY ai.block_timestamp DESC\nLIMIT 1000;\n", "describe": { "columns": [ { @@ -93,5 +93,5 @@ null ] }, - "hash": "b4e72b9ba8a6b9d8947e2fdd6b58ecad9ad0c9bd398c13fdd879dced0dc81c43" + "hash": "0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68" } diff --git a/.sqlx/query-12f0f248f7ccf430681015c0749ba4b067916690de45dee63b6b2eab15a8ec43.json b/.sqlx/query-12f0f248f7ccf430681015c0749ba4b067916690de45dee63b6b2eab15a8ec43.json new file mode 100644 index 0000000..84f3a9b --- /dev/null +++ b/.sqlx/query-12f0f248f7ccf430681015c0749ba4b067916690de45dee63b6b2eab15a8ec43.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO initiated_transactions\n (source_transaction_hash, direction, message_id, sender, receiver, amount,\n source_block_hash, source_block_number, source_tx_index, timestamp, tx_type)\nVALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)\nON CONFLICT (source_transaction_hash) DO NOTHING\n", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Text", + "Text", + "Text", + "Text", + "Int4", + "Int4", + "Int8", + "Text" + ] + }, + "nullable": [] + }, + "hash": "12f0f248f7ccf430681015c0749ba4b067916690de45dee63b6b2eab15a8ec43" +} diff --git a/.sqlx/query-2768e552e48f41ac350ec5b1d15bae344e7e5e287b01276197a3a97ad3d690de.json b/.sqlx/query-2768e552e48f41ac350ec5b1d15bae344e7e5e287b01276197a3a97ad3d690de.json new file mode 100644 index 0000000..0a969a5 --- /dev/null +++ b/.sqlx/query-2768e552e48f41ac350ec5b1d15bae344e7e5e287b01276197a3a97ad3d690de.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_block_number = $1 AND source_tx_index = $2)\n", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Int4", + "Int4" + ] + }, + "nullable": [ + null + ] + }, + "hash": "2768e552e48f41ac350ec5b1d15bae344e7e5e287b01276197a3a97ad3d690de" +} diff --git a/.sqlx/query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json b/.sqlx/query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json new file mode 100644 index 0000000..461f54d --- /dev/null +++ b/.sqlx/query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_transaction_hash = $1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20" +} diff --git a/.sqlx/query-38c33350008ff3a7d58278df6fd0a715cf79f2e352838105f09c1eaf2d90aac3.json b/.sqlx/query-38c33350008ff3a7d58278df6fd0a715cf79f2e352838105f09c1eaf2d90aac3.json new file mode 100644 index 0000000..e7f233f --- /dev/null +++ b/.sqlx/query-38c33350008ff3a7d58278df6fd0a715cf79f2e352838105f09c1eaf2d90aac3.json @@ -0,0 +1,97 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT ai.id AS message_id,\n ai.signature_address AS sender,\n es.to AS \"receiver!\",\n COALESCE(es.amount, '0')::text AS \"amount!\",\n ai.block_hash AS source_block_hash,\n ai.ext_hash AS source_transaction_hash,\n ai.block_height AS source_block_height,\n ai.ext_index AS source_tx_index,\n EXTRACT(EPOCH FROM ai.block_timestamp)::bigint AS \"block_timestamp!\",\n be.source_transaction_hash AS \"destination_tx_hash?: String\",\n COALESCE(\n CASE\n WHEN be.message_id IS NOT NULL THEN 'bridged'::status\n END,\n 'in_progress'::status\n ) ::status AS \"final_status!: BridgeStatusEnum\"\nFROM avail_send_message_table es\n INNER JOIN public.avail_indexer AS ai\n ON ai.id = es.id\n LEFT JOIN public.bridge_event AS be\n ON es.id = be.message_id\nWHERE ai.signature_address = $1\n AND es.type = $2\n AND (be.event_type = $3 or be.event_type is null)\n AND ai.ext_success = $4\nORDER BY ai.block_timestamp DESC\nLIMIT 1000;\n", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "message_id", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "sender", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "receiver!", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "amount!", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "source_block_hash", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "source_transaction_hash", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "source_block_height", + "type_info": "Int4" + }, + { + "ordinal": 7, + "name": "source_tx_index", + "type_info": "Int4" + }, + { + "ordinal": 8, + "name": "block_timestamp!", + "type_info": "Int8" + }, + { + "ordinal": 9, + "name": "destination_tx_hash?: String", + "type_info": "Text" + }, + { + "ordinal": 10, + "name": "final_status!: BridgeStatusEnum", + "type_info": { + "Custom": { + "name": "status", + "kind": { + "Enum": [ + "initiated", + "in_progress", + "claim_ready", + "bridged" + ] + } + } + } + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Bool" + ] + }, + "nullable": [ + false, + true, + false, + null, + false, + false, + false, + false, + null, + false, + null + ] + }, + "hash": "38c33350008ff3a7d58278df6fd0a715cf79f2e352838105f09c1eaf2d90aac3" +} diff --git a/.sqlx/query-aeb96222ec314d46fabb7bb143802e8f22ef09551605a7c4f2cc26d315f77aaa.json b/.sqlx/query-aeb96222ec314d46fabb7bb143802e8f22ef09551605a7c4f2cc26d315f77aaa.json new file mode 100644 index 0000000..8070c20 --- /dev/null +++ b/.sqlx/query-aeb96222ec314d46fabb7bb143802e8f22ef09551605a7c4f2cc26d315f77aaa.json @@ -0,0 +1,83 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT source_transaction_hash, direction, message_id, sender, receiver,\n amount, source_block_hash, source_block_number, source_tx_index, timestamp, tx_type\nFROM initiated_transactions it\nWHERE tx_type = 'initiate'\n AND (it.sender = $1 OR it.sender = $2)\n AND NOT EXISTS (\n SELECT 1 FROM bridge_event be\n WHERE be.source_transaction_hash = it.source_transaction_hash\n AND it.direction = 'EthAvail'\n )\n AND NOT EXISTS (\n SELECT 1 FROM avail_indexer ai\n WHERE ai.ext_hash = it.source_transaction_hash\n AND it.direction = 'AvailEth'\n )\nORDER BY it.timestamp DESC\nLIMIT 100\n", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "source_transaction_hash", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "direction", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "message_id", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "sender", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "receiver", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "amount", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "source_block_hash", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "source_block_number", + "type_info": "Int4" + }, + { + "ordinal": 8, + "name": "source_tx_index", + "type_info": "Int4" + }, + { + "ordinal": 9, + "name": "timestamp", + "type_info": "Int8" + }, + { + "ordinal": 10, + "name": "tx_type", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + false, + false, + true, + false, + false + ] + }, + "hash": "aeb96222ec314d46fabb7bb143802e8f22ef09551605a7c4f2cc26d315f77aaa" +} diff --git a/.sqlx/query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json b/.sqlx/query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json new file mode 100644 index 0000000..f7b6ba0 --- /dev/null +++ b/.sqlx/query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json @@ -0,0 +1,40 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT message_id, source_transaction_hash, source_block_number, source_tx_index\nFROM initiated_transactions\nWHERE tx_type = 'claim'\n AND message_id = ANY($1)\n", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "message_id", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "source_transaction_hash", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "source_block_number", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "source_tx_index", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "TextArray" + ] + }, + "nullable": [ + false, + false, + false, + true + ] + }, + "hash": "dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09" +} diff --git a/.sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json b/.sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json new file mode 100644 index 0000000..635db59 --- /dev/null +++ b/.sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM initiated_transactions\nWHERE\n-- Clean up initiate records when source tx is indexed\n(tx_type = 'initiate' AND direction = 'EthAvail' AND EXISTS (\n SELECT 1 FROM bridge_event be\n WHERE be.source_transaction_hash = initiated_transactions.source_transaction_hash\n))\nOR (tx_type = 'initiate' AND direction = 'AvailEth' AND EXISTS (\n SELECT 1 FROM avail_indexer ai\n WHERE ai.ext_hash = initiated_transactions.source_transaction_hash\n))\n-- Clean up claim records when claim tx is indexed\nOR (tx_type = 'claim' AND direction = 'AvailEth' AND EXISTS (\n SELECT 1 FROM bridge_event be\n WHERE be.message_id::text = initiated_transactions.message_id\n AND be.event_type = 'MessageReceived'\n))\nOR (tx_type = 'claim' AND direction = 'EthAvail' AND EXISTS (\n SELECT 1 FROM avail_execute_table aet\n WHERE aet.message_id::text = initiated_transactions.message_id\n))\n", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084" +} diff --git a/Cargo.lock b/Cargo.lock index 5a9ade9..d06b32a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1707,6 +1707,7 @@ dependencies = [ "jsonrpsee", "lazy_static", "log", + "parity-scale-codec", "reqwest", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 4700059..311201a 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ sqlx = { version = "0.8.6", features = ["postgres", "runtime-tokio-rustls", "mac bigdecimal = { version = "0.4", features = ["serde"] } uuid = { version = "1.18.0", features = ["v4"] } log = "0.4.25" +parity-scale-codec = "3.6" [target.'cfg(not(target_env = "msvc"))'.dependencies] tikv-jemallocator = "0.6" diff --git a/migrations/0001_create_initiated_transactions.sql b/migrations/0001_create_initiated_transactions.sql new file mode 100644 index 0000000..71f2c4b --- /dev/null +++ b/migrations/0001_create_initiated_transactions.sql @@ -0,0 +1,15 @@ +CREATE TABLE IF NOT EXISTS initiated_transactions ( + source_transaction_hash TEXT PRIMARY KEY, + direction TEXT NOT NULL, + message_id TEXT NOT NULL DEFAULT '0', + sender TEXT NOT NULL, + receiver TEXT NOT NULL, + amount TEXT NOT NULL, + source_block_hash TEXT NOT NULL, + source_block_number INTEGER NOT NULL, + source_tx_index INTEGER, + timestamp BIGINT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_initiated_tx_sender ON initiated_transactions (sender); +ALTER TABLE initiated_transactions ADD COLUMN IF NOT EXISTS tx_type TEXT NOT NULL DEFAULT 'initiate'; diff --git a/sql/check_initiated_tx_by_block.sql b/sql/check_initiated_tx_by_block.sql new file mode 100644 index 0000000..722810d --- /dev/null +++ b/sql/check_initiated_tx_by_block.sql @@ -0,0 +1 @@ +SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_block_number = $1 AND source_tx_index = $2) diff --git a/sql/check_initiated_tx_by_hash.sql b/sql/check_initiated_tx_by_hash.sql new file mode 100644 index 0000000..d8e2e9c --- /dev/null +++ b/sql/check_initiated_tx_by_hash.sql @@ -0,0 +1 @@ +SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_transaction_hash = $1) \ No newline at end of file diff --git a/sql/delete_indexed_initiated_tx.sql b/sql/delete_indexed_initiated_tx.sql new file mode 100644 index 0000000..01f51d8 --- /dev/null +++ b/sql/delete_indexed_initiated_tx.sql @@ -0,0 +1,21 @@ +DELETE FROM initiated_transactions +WHERE +-- Clean up initiate records when source tx is indexed +(tx_type = 'initiate' AND direction = 'EthAvail' AND EXISTS ( + SELECT 1 FROM bridge_event be + WHERE be.source_transaction_hash = initiated_transactions.source_transaction_hash +)) +OR (tx_type = 'initiate' AND direction = 'AvailEth' AND EXISTS ( + SELECT 1 FROM avail_indexer ai + WHERE ai.ext_hash = initiated_transactions.source_transaction_hash +)) +-- Clean up claim records when claim tx is indexed +OR (tx_type = 'claim' AND direction = 'AvailEth' AND EXISTS ( + SELECT 1 FROM bridge_event be + WHERE be.message_id::text = initiated_transactions.message_id + AND be.event_type = 'MessageReceived' +)) +OR (tx_type = 'claim' AND direction = 'EthAvail' AND EXISTS ( + SELECT 1 FROM avail_execute_table aet + WHERE aet.message_id::text = initiated_transactions.message_id +)) diff --git a/sql/insert_initiated_tx.sql b/sql/insert_initiated_tx.sql new file mode 100644 index 0000000..e84001b --- /dev/null +++ b/sql/insert_initiated_tx.sql @@ -0,0 +1,5 @@ +INSERT INTO initiated_transactions + (source_transaction_hash, direction, message_id, sender, receiver, amount, + source_block_hash, source_block_number, source_tx_index, timestamp, tx_type) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) +ON CONFLICT (source_transaction_hash) DO NOTHING diff --git a/sql/query_avail_tx.sql b/sql/query_avail_tx.sql index 66e9d44..1e5da09 100644 --- a/sql/query_avail_tx.sql +++ b/sql/query_avail_tx.sql @@ -6,7 +6,7 @@ SELECT ai.id AS message_id, ai.ext_hash AS source_transaction_hash, ai.block_height AS source_block_height, ai.ext_index AS source_tx_index, - es.block_timestamp, + ai.block_timestamp, be.source_transaction_hash AS "destination_tx_hash?: String", COALESCE( CASE @@ -23,5 +23,5 @@ WHERE ai.signature_address = $1 AND es.type = $2 AND (be.event_type = $3 or be.event_type is null) AND ai.ext_success = $4 -ORDER BY es.block_timestamp DESC +ORDER BY ai.block_timestamp DESC LIMIT 1000; diff --git a/sql/query_claimed_tx.sql b/sql/query_claimed_tx.sql new file mode 100644 index 0000000..b5636ac --- /dev/null +++ b/sql/query_claimed_tx.sql @@ -0,0 +1,4 @@ +SELECT message_id, source_transaction_hash, source_block_number, source_tx_index +FROM initiated_transactions +WHERE tx_type = 'claim' + AND message_id = ANY($1) diff --git a/sql/query_initiated_tx.sql b/sql/query_initiated_tx.sql new file mode 100644 index 0000000..00b6ebb --- /dev/null +++ b/sql/query_initiated_tx.sql @@ -0,0 +1,17 @@ +SELECT source_transaction_hash, direction, message_id, sender, receiver, + amount, source_block_hash, source_block_number, source_tx_index, timestamp, tx_type +FROM initiated_transactions it +WHERE tx_type = 'initiate' + AND (it.sender = $1 OR it.sender = $2) + AND NOT EXISTS ( + SELECT 1 FROM bridge_event be + WHERE be.source_transaction_hash = it.source_transaction_hash + AND it.direction = 'EthAvail' + ) + AND NOT EXISTS ( + SELECT 1 FROM avail_indexer ai + WHERE ai.ext_hash = it.source_transaction_hash + AND it.direction = 'AvailEth' + ) +ORDER BY it.timestamp DESC +LIMIT 100 diff --git a/src/initiate_transaction.rs b/src/initiate_transaction.rs new file mode 100644 index 0000000..6276617 --- /dev/null +++ b/src/initiate_transaction.rs @@ -0,0 +1,624 @@ +use crate::models::{ErrorResponse, InitiateRequest}; +use crate::{ + AppState, +}; +use alloy::primitives::{hex, U256}; +use alloy::sol_types::SolCall; +use anyhow::anyhow; +use axum::{ + extract::{Json, State}, + http::StatusCode, + response::IntoResponse, +}; +use serde_json::{json, Value}; +use std::sync::Arc; +use alloy::signers::k256::sha2::Digest; +use avail_core::data_proof::Message; +use axum::response::Response; +use jsonrpsee::core::client::ClientT; +use jsonrpsee::http_client::HttpClient; +use jsonrpsee::rpc_params; +use parity_scale_codec::{Compact, Decode}; +use sha3::Keccak256; +use sp_core::{blake2_256, twox_128}; +use sp_core::crypto::{AccountId32, Ss58Codec}; +pub(crate) async fn initiate_transaction( + State(state): State>, + Json(request): Json, +) -> Result { + let has_eth = request.eth_tx_hash.is_some(); + let has_avail = request.avail_block_number.is_some() || request.avail_tx_index.is_some(); + + if !has_eth && !has_avail { + return Err(ErrorResponse::with_status( + anyhow!("Either ethTxHash or (availBlockNumber and availTxIndex) must be provided"), + StatusCode::BAD_REQUEST, + )); + } + if has_eth && has_avail { + return Err(ErrorResponse::with_status( + anyhow!("Provide either ethTxHash or (availBlockNumber and availTxIndex), not both"), + StatusCode::BAD_REQUEST, + )); + } + if has_avail && (request.avail_block_number.is_none() || request.avail_tx_index.is_none()) { + return Err(ErrorResponse::with_status( + anyhow!("Both availBlockNumber and availTxIndex must be provided"), + StatusCode::BAD_REQUEST, + )); + } + + if let Some(ref hash) = request.eth_tx_hash { + if hash.len() != 66 || !hash.starts_with("0x") || hex::decode(&hash[2..]).is_err() { + return Err(ErrorResponse::with_status( + anyhow!("Invalid transaction hash format, expected 0x-prefixed 32-byte hex"), + StatusCode::BAD_REQUEST, + )); + } + } + + // Check if already processed to avoid redundant RPC calls (skip for claims) + if request.message_id.is_none() { + if let Some(eth_tx_hash) = &request.eth_tx_hash { + let exists: Option = + sqlx::query_file_scalar!("sql/check_initiated_tx_by_hash.sql", eth_tx_hash) + .fetch_one(&state.db) + .await?; + if exists.unwrap_or(false) { + return Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()); + } + } + if let (Some(bn), Some(ti)) = (request.avail_block_number, request.avail_tx_index) { + let exists: Option = sqlx::query_file_scalar!( + "sql/check_initiated_tx_by_block.sql", + bn as i32, + ti as i32 + ) + .fetch_one(&state.db) + .await?; + if exists.unwrap_or(false) { + return Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()); + } + } + } + + match ( + &request.message_id, + &request.eth_tx_hash, + request.avail_block_number.zip(request.avail_tx_index), + ) { + (Some(message_id), Some(eth_tx_hash), _) => { + claim_eth_transaction(&state, eth_tx_hash, message_id).await + } + (Some(message_id), None, Some((block_number, tx_index))) => { + claim_avail_transaction(&state, block_number, tx_index, message_id).await + } + (None, Some(eth_tx_hash), _) => initiate_eth_transaction(&state, eth_tx_hash).await, + (None, None, Some((block_number, tx_index))) => { + initiate_avail_transaction(&state, block_number, tx_index).await + } + _ => Err(ErrorResponse::with_status( + anyhow!("Wrong input"), + StatusCode::BAD_REQUEST, + )), + } +} + + +async fn initiate_eth_transaction( + state: &Arc, + eth_tx_hash: &str, +) -> anyhow::Result { + let receipt: Value = state + .ethereum_client + .request("eth_getTransactionReceipt", rpc_params![eth_tx_hash]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch ETH receipt: {e:#}"); + ErrorResponse::with_status( + anyhow!("Failed to fetch transaction receipt"), + StatusCode::BAD_GATEWAY, + ) + })?; + + if receipt.is_null() { + return Err(ErrorResponse::with_status( + anyhow!("Transaction not found"), + StatusCode::NOT_FOUND, + )); + } + + let tx: Value = state + .ethereum_client + .request("eth_getTransactionByHash", rpc_params![eth_tx_hash]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch ETH transaction: {e:#}"); + ErrorResponse::with_status( + anyhow!("Failed to fetch transaction"), + StatusCode::BAD_REQUEST, + ) + })?; + + let block_hash = receipt["blockHash"] + .as_str() + .ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing blockHash"), StatusCode::BAD_REQUEST) + })? + .to_string(); + let block_number_hex = receipt["blockNumber"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing blockNumber"), StatusCode::BAD_REQUEST) + })?; + let block_number = + i32::from_str_radix(block_number_hex.trim_start_matches("0x"), 16).map_err(|_| { + ErrorResponse::with_status(anyhow!("Invalid blockNumber"), StatusCode::BAD_REQUEST) + })?; + + // Find MessageSent event in logs + let mut hasher = Keccak256::new(); + hasher.update(b"MessageSent(address,bytes32,uint256)"); + let event_topic = format!("0x{}", hex::encode(hasher.finalize())); + + let logs = receipt["logs"].as_array().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing logs"), StatusCode::BAD_REQUEST) + })?; + + let event_log = logs + .iter() + .find(|log| log["topics"][0].as_str() == Some(&event_topic)) + .ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("MessageSent event not found in transaction"), + StatusCode::BAD_REQUEST, + ) + })?; + + // sender from topic[1] (address left-padded to 32 bytes) + let sender_topic = event_log["topics"][1].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing sender topic"), StatusCode::BAD_REQUEST) + })?; + + let sender = format!("0x{}", &sender_topic[26..]); + + // receiver from topic[2] (bytes32) + let receiver = event_log["topics"][2] + .as_str() + .ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing receiver topic"), StatusCode::BAD_REQUEST) + })? + .to_string(); + + // messageId from data + let msg_data = event_log["data"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing event data"), StatusCode::BAD_REQUEST) + })?; + let message_id = msg_data + .parse::() + .map(|v| v.to_string()) + .map_err(|_| { + ErrorResponse::with_status(anyhow!("Missing messageId"), StatusCode::BAD_REQUEST) + })?; + + // amount from tx input: sendAVAIL(bytes32,uint256) + let input = tx["input"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing input"), StatusCode::BAD_REQUEST) + })?; + let input_bytes = hex::decode(input.trim_start_matches("0x")).map_err(|_| { + ErrorResponse::with_status(anyhow!("Invalid input hex"), StatusCode::BAD_REQUEST) + })?; + let call = crate::AvailBridge::sendAVAILCall::abi_decode(&input_bytes).map_err(|_| { + ErrorResponse::with_status( + anyhow!("Transaction is not a sendAVAIL call"), + StatusCode::BAD_REQUEST, + ) + })?; + let amount = call.amount.to_string(); + + // timestamp from block + let block: Value = state + .ethereum_client + .request("eth_getBlockByNumber", rpc_params![block_number_hex, false]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch block: {e:#}"); + ErrorResponse::with_status(anyhow!("Failed to fetch block"), StatusCode::BAD_REQUEST) + })?; + let timestamp_hex = block["timestamp"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing timestamp"), StatusCode::BAD_REQUEST) + })?; + let timestamp = + i64::from_str_radix(timestamp_hex.trim_start_matches("0x"), 16).map_err(|_| { + ErrorResponse::with_status(anyhow!("Invalid timestamp"), StatusCode::BAD_REQUEST) + })?; + + sqlx::query_file!( + "sql/insert_initiated_tx.sql", + eth_tx_hash, + "EthAvail", + &message_id, + &sender, + &receiver, + &amount, + &block_hash, + block_number, + None::, + timestamp, + "initiate", + ) + .execute(&state.db) + .await?; + + Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) +} + +async fn fetch_avail_block_timestamp( + client: &HttpClient, + block_hash: &str, +) -> anyhow::Result { + let key = format!( + "0x{}{}", + hex::encode(twox_128(b"Timestamp")), + hex::encode(twox_128(b"Now")) + ); + let storage_hex: String = client + .request("state_getStorage", rpc_params![&key, block_hash]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch Avail block timestamp: {e:#}"); + ErrorResponse::with_status( + anyhow!("Failed to fetch block timestamp"), + StatusCode::BAD_GATEWAY, + ) + })?; + let bytes = hex::decode(storage_hex.trim_start_matches("0x")).map_err(|_| { + ErrorResponse::with_status( + anyhow!("Invalid timestamp encoding"), + StatusCode::BAD_GATEWAY, + ) + })?; + let timestamp_bytes: [u8; 8] = + bytes + .get(..8) + .and_then(|s| s.try_into().ok()) + .ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("Invalid timestamp length"), + StatusCode::BAD_GATEWAY, + ) + })?; + let timestamp_ms = u64::from_le_bytes(timestamp_bytes); + Ok((timestamp_ms / 1000) as i64) +} + +async fn initiate_avail_transaction( + state: &Arc, + block_number: u32, + tx_index: u32, +) -> anyhow::Result { + let block_hash: String = state + .avail_client + .request("chain_getBlockHash", rpc_params![block_number]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch Avail block hash: {e:#}"); + ErrorResponse::with_status( + anyhow!("Failed to fetch block hash"), + StatusCode::BAD_GATEWAY, + ) + })?; + + // Fetch block to get the extrinsic and extract signer + let block: Value = state + .avail_client + .request("chain_getBlock", rpc_params![&block_hash]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch Avail block: {e:#}"); + ErrorResponse::with_status(anyhow!("Failed to fetch block"), StatusCode::BAD_GATEWAY) + })?; + + let extrinsics = block["block"]["extrinsics"].as_array().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing extrinsics"), StatusCode::BAD_GATEWAY) + })?; + + let ext_hex = extrinsics + .get(tx_index as usize) + .and_then(|v| v.as_str()) + .ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("Extrinsic not found at index {tx_index}"), + StatusCode::NOT_FOUND, + ) + })?; + + let ext_bytes = hex::decode(ext_hex.trim_start_matches("0x")).map_err(|_| { + ErrorResponse::with_status( + anyhow!("Invalid extrinsic encoding"), + StatusCode::BAD_GATEWAY, + ) + })?; + + // Extrinsic hash matching indexer's ext.metadata.ext_hash + let ext_hash_hex = format!("0x{}", hex::encode(blake2_256(&ext_bytes))); + + // Decode extrinsic using SCALE codec cursor + let mut cursor = &ext_bytes[..]; + let decode_err = |e| { + ErrorResponse::with_status( + anyhow!("Failed to decode extrinsic: {e}"), + StatusCode::BAD_GATEWAY, + ) + }; + + // Skip compact length prefix + >::decode(&mut cursor).map_err(decode_err)?; + + // Version byte (0x84 = V4 signed) + let version = u8::decode(&mut cursor).map_err(decode_err)?; + + // Extract signer from signed extrinsic + let sender = if version & 0x80 != 0 { + // MultiAddress::Id = 0x00 + 32 bytes AccountId + let addr_type = u8::decode(&mut cursor).map_err(decode_err)?; + if addr_type != 0x00 { + return Err(ErrorResponse::with_status( + anyhow!("Unsupported address type"), + StatusCode::BAD_GATEWAY, + )); + } + let account = <[u8; 32]>::decode(&mut cursor).map_err(decode_err)?; + let sender = AccountId32::new(account).to_ss58check(); + + // Skip MultiSignature + let sig_type = u8::decode(&mut cursor).map_err(decode_err)?; + match sig_type { + 0x00 | 0x01 => { + <[u8; 64]>::decode(&mut cursor).map_err(decode_err)?; + } + 0x02 => { + <[u8; 65]>::decode(&mut cursor).map_err(decode_err)?; + } + _ => { + return Err(ErrorResponse::with_status( + anyhow!("Unknown signature type"), + StatusCode::BAD_GATEWAY, + )); + } + }; + + // Skip SignedExtras: Era + Nonce + Tip + AppId + let era = u8::decode(&mut cursor).map_err(decode_err)?; + if era != 0x00 { + u8::decode(&mut cursor).map_err(decode_err)?; // mortal era second byte + } + >::decode(&mut cursor).map_err(decode_err)?; // nonce + >::decode(&mut cursor).map_err(decode_err)?; // tip + >::decode(&mut cursor).map_err(decode_err)?; // app_id + + sender + } else { + String::new() + }; + + // Skip pallet_id + call_id + <[u8; 2]>::decode(&mut cursor).map_err(decode_err)?; + + // Decode SendMessage params: Message + to (H256) + let message = Message::decode(&mut cursor).map_err(decode_err)?; + let to = sp_core::H256::decode(&mut cursor).map_err(decode_err)?; + + let amount = match &message { + Message::FungibleToken { amount, .. } => amount.to_string(), + _ => "0".to_string(), + }; + let receiver = format!("0x{}", hex::encode(to.as_bytes())); + + // ID matching indexer's (block_height << 32) | ext_index + let message_id = ((block_number as u64) << 32 | tx_index as u64).to_string(); + let timestamp = fetch_avail_block_timestamp(&state.avail_client, &block_hash).await?; + + sqlx::query_file!( + "sql/insert_initiated_tx.sql", + &ext_hash_hex, + "AvailEth", + &message_id, + &sender, + &receiver, + &amount, + &block_hash, + block_number as i32, + tx_index as i32, + timestamp, + "initiate", + ) + .execute(&state.db) + .await?; + + Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) +} + +async fn claim_eth_transaction( + state: &Arc, + eth_tx_hash: &str, + message_id: &str, +) -> anyhow::Result { + // Verify tx exists + let receipt: Value = state + .ethereum_client + .request("eth_getTransactionReceipt", rpc_params![eth_tx_hash]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch ETH receipt: {e:#}"); + ErrorResponse::with_status( + anyhow!("Failed to fetch transaction receipt"), + StatusCode::BAD_GATEWAY, + ) + })?; + + if receipt.is_null() { + return Err(ErrorResponse::with_status( + anyhow!("Transaction not found"), + StatusCode::NOT_FOUND, + )); + } + + let block_hash = receipt["blockHash"] + .as_str() + .ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing blockHash"), StatusCode::BAD_REQUEST) + })? + .to_string(); + let block_number_hex = receipt["blockNumber"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing blockNumber"), StatusCode::BAD_REQUEST) + })?; + let block_number = + i32::from_str_radix(block_number_hex.trim_start_matches("0x"), 16).map_err(|_| { + ErrorResponse::with_status(anyhow!("Invalid blockNumber"), StatusCode::BAD_REQUEST) + })?; + + let tx: Value = state + .ethereum_client + .request("eth_getTransactionByHash", rpc_params![eth_tx_hash]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch ETH transaction: {e:#}"); + ErrorResponse::with_status( + anyhow!("Failed to fetch transaction"), + StatusCode::BAD_REQUEST, + ) + })?; + + let sender = tx["from"].as_str().unwrap_or("").to_string(); + + let block: Value = state + .ethereum_client + .request("eth_getBlockByNumber", rpc_params![block_number_hex, false]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch block: {e:#}"); + ErrorResponse::with_status(anyhow!("Failed to fetch block"), StatusCode::BAD_REQUEST) + })?; + let timestamp_hex = block["timestamp"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing timestamp"), StatusCode::BAD_REQUEST) + })?; + let timestamp = + i64::from_str_radix(timestamp_hex.trim_start_matches("0x"), 16).map_err(|_| { + ErrorResponse::with_status(anyhow!("Invalid timestamp"), StatusCode::BAD_REQUEST) + })?; + + sqlx::query_file!( + "sql/insert_initiated_tx.sql", + eth_tx_hash, + "AvailEth", + message_id, + &sender, + "", + "0", + &block_hash, + block_number, + None::, + timestamp, + "claim", + ) + .execute(&state.db) + .await?; + + Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) +} + +async fn claim_avail_transaction( + state: &Arc, + block_number: u32, + tx_index: u32, + message_id: &str, +) -> anyhow::Result { + let block_hash: String = state + .avail_client + .request("chain_getBlockHash", rpc_params![block_number]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch Avail block hash: {e:#}"); + ErrorResponse::with_status( + anyhow!("Failed to fetch block hash"), + StatusCode::BAD_GATEWAY, + ) + })?; + + // Verify extrinsic exists + let block: Value = state + .avail_client + .request("chain_getBlock", rpc_params![&block_hash]) + .await + .map_err(|e| { + tracing::error!("Failed to fetch Avail block: {e:#}"); + ErrorResponse::with_status(anyhow!("Failed to fetch block"), StatusCode::BAD_GATEWAY) + })?; + + let extrinsics = block["block"]["extrinsics"].as_array().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing extrinsics"), StatusCode::BAD_GATEWAY) + })?; + + let ext_hex = extrinsics + .get(tx_index as usize) + .and_then(|v| v.as_str()) + .ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("Extrinsic not found at index {tx_index}"), + StatusCode::NOT_FOUND, + ) + })?; + + let ext_bytes = hex::decode(ext_hex.trim_start_matches("0x")).map_err(|_| { + ErrorResponse::with_status( + anyhow!("Invalid extrinsic encoding"), + StatusCode::BAD_GATEWAY, + ) + })?; + + let ext_hash_hex = format!("0x{}", hex::encode(blake2_256(&ext_bytes))); + + // Decode sender from extrinsic + let mut cursor = &ext_bytes[..]; + let decode_err = |e| { + ErrorResponse::with_status( + anyhow!("Failed to decode extrinsic: {e}"), + StatusCode::BAD_GATEWAY, + ) + }; + + >::decode(&mut cursor).map_err(decode_err)?; + let version = u8::decode(&mut cursor).map_err(decode_err)?; + + let sender = if version & 0x80 != 0 { + let addr_type = u8::decode(&mut cursor).map_err(decode_err)?; + if addr_type != 0x00 { + String::new() + } else { + let account = <[u8; 32]>::decode(&mut cursor).map_err(decode_err)?; + AccountId32::new(account).to_ss58check() + } + } else { + String::new() + }; + + let timestamp = fetch_avail_block_timestamp(&state.avail_client, &block_hash).await?; + + sqlx::query_file!( + "sql/insert_initiated_tx.sql", + &ext_hash_hex, + "EthAvail", + message_id, + &sender, + "", + "0", + &block_hash, + block_number as i32, + tx_index as i32, + timestamp, + "claim", + ) + .execute(&state.db) + .await?; + + Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) +} diff --git a/src/main.rs b/src/main.rs index 69747ef..1726683 100755 --- a/src/main.rs +++ b/src/main.rs @@ -1,8 +1,11 @@ +mod initiate_transaction; mod models; +use crate::initiate_transaction::initiate_transaction; use crate::models::*; use alloy::primitives::{Address, B256, U256, hex}; use alloy::providers::ProviderBuilder; +use alloy::sol_types::SolCall; use anyhow::{Context, Result, anyhow}; use axum::body::{Body, to_bytes}; use axum::response::Response; @@ -11,7 +14,7 @@ use axum::{ extract::{Json, Path, Query, State}, http::StatusCode, response::IntoResponse, - routing::get, + routing::{get, post}, }; use backon::ExponentialBuilder; use backon::Retryable; @@ -19,6 +22,8 @@ use chrono::Utc; use crate::models::TxDirection::{AvailEth, EthAvail}; use alloy::core::sol; +use avail_core::data_proof::Message; +use bigdecimal::BigDecimal; use http::Method; use jsonrpsee::{ core::ClientError, @@ -27,11 +32,12 @@ use jsonrpsee::{ rpc_params, }; use lazy_static::lazy_static; +use parity_scale_codec::Compact; use reqwest::Client; use serde_json::{Value, json}; use sha3::{Digest, Keccak256}; use sp_core::Decode; -use sp_io::hashing::twox_128; +use sp_core::twox_128; use sqlx::PgPool; use std::collections::HashMap; use std::sync::Arc; @@ -45,7 +51,7 @@ use tower_http::{ cors::{Any, CorsLayer}, trace::TraceLayer, }; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; use tracing_subscriber::prelude::*; sol! { @@ -122,6 +128,9 @@ async fn transactions( let mut transaction_data_results: Vec = vec![]; + let eth_addr_for_initiated = address_query.eth_address.map(|a| format!("{:?}", a)); + let avail_addr_for_initiated = address_query.avail_address.clone(); + if let Some(eth_address) = address_query.eth_address { let transactions: Vec = sqlx::query_file_as!( EthTransactionRow, @@ -196,7 +205,9 @@ async fn transactions( .await .context("get header") .unwrap_or(json!({})); - let number_hex = header["number"].as_str().unwrap(); + let number_hex = header["number"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing blockNumber"), StatusCode::BAD_REQUEST) + })?; let latest_block_number = u32::from_str_radix(&number_hex[2..], 16).unwrap_or(0); let claim_estimate = time_until_next_vector_update( @@ -247,6 +258,77 @@ async fn transactions( } } + // Query initiated transactions not yet indexed + let initiated: Vec = sqlx::query_file_as!( + InitiatedTransactionRow, + "sql/query_initiated_tx.sql", + eth_addr_for_initiated.as_deref().unwrap_or(""), + avail_addr_for_initiated.as_deref().unwrap_or("") + ) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + for row in &initiated { + let direction = if row.direction == "EthAvail" { + EthAvail + } else { + AvailEth + }; + let msg_id: BigDecimal = row.message_id.parse().unwrap_or_default(); + transaction_data_results.push(TransactionData::new( + direction, + msg_id, + row.sender.clone(), + row.receiver.clone(), + row.source_block_hash.clone(), + row.source_transaction_hash.clone(), + row.amount.clone(), + BridgeStatusEnum::Initiated, + row.timestamp, + None, + Some(row.source_block_number), + row.source_tx_index, + None, + None, + None, + )); + } + + // Check for claimed transactions and update status to Bridged + let message_ids: Vec = transaction_data_results + .iter() + .map(|t| t.message_id.to_string()) + .collect(); + + if !message_ids.is_empty() { + let claims: Vec = sqlx::query_file_as!( + ClaimedTransactionRow, + "sql/query_claimed_tx.sql", + &message_ids, + ) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + for claim in &claims { + if let Some(tx) = transaction_data_results + .iter_mut() + .find(|t| t.message_id.to_string() == claim.message_id) + { + tx.status = BridgeStatusEnum::Bridged; + tx.destination_tx_hash = Some(claim.source_transaction_hash.clone()); + tx.destination_block_number = Some(claim.source_block_number); + tx.destination_tx_index = claim.source_tx_index; + } + } + } + + // Clean up initiated/claimed transactions that have been indexed + let _ = sqlx::query_file!("sql/delete_indexed_initiated_tx.sql") + .execute(&state.db) + .await; + transaction_data_results.sort_unstable_by(|a, b| b.timestamp.cmp(&a.timestamp)); Ok(( @@ -963,13 +1045,14 @@ async fn main() { ) .route("/v1/transactions", get(transactions)) // fetch all transaction .route("/transactions", get(transactions)) + .route("/v1/initiate", post(initiate_transaction)) .route("/v1/head/{chain_id}", get(get_head)) // get head based on chain .route("/v1/proof/{chain_id}", get(get_proof)) // get proof for avail based on chain .layer(TraceLayer::new_for_http()) .layer(CompressionLayer::new()) .layer( CorsLayer::new() - .allow_methods(vec![Method::GET]) + .allow_methods(vec![Method::GET, Method::POST]) .allow_origin(Any), ) .with_state(shared_state.clone()); @@ -1090,7 +1173,7 @@ async fn track_slot_avail_task(state: Arc) -> Result<()> { let mut slot_block_head = SLOT_BLOCK_HEAD.write().await; tracing::info!("Beacon mapping: {slot}:{bl}"); *slot_block_head = Some((slot, bl, hash, timestamp)); - info!("Timestamp form task: {:?}", timestamp); + debug!("Timestamp form task: {:?}", timestamp); drop(slot_block_head); diff --git a/src/models.rs b/src/models.rs index dccb254..d8cdb5e 100644 --- a/src/models.rs +++ b/src/models.rs @@ -402,3 +402,35 @@ impl TransactionData { } } } + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct InitiateRequest { + pub eth_tx_hash: Option, + pub avail_block_number: Option, + pub avail_tx_index: Option, + pub message_id: Option, +} + +#[derive(Debug, FromRow)] +pub struct InitiatedTransactionRow { + pub source_transaction_hash: String, + pub direction: String, + pub message_id: String, + pub sender: String, + pub receiver: String, + pub amount: String, + pub source_block_hash: String, + pub source_block_number: i32, + pub source_tx_index: Option, + pub timestamp: i64, + pub tx_type: String, +} + +#[derive(Debug, FromRow)] +pub struct ClaimedTransactionRow { + pub message_id: String, + pub source_transaction_hash: String, + pub source_block_number: i32, + pub source_tx_index: Option, +} From 50014f0f9e3bf6c3637796197667353bf959ebd7 Mon Sep 17 00:00:00 2001 From: Sasa Prsic Date: Fri, 27 Feb 2026 15:33:04 +0100 Subject: [PATCH 2/6] simplify delete. --- sql/delete_indexed_initiated_tx.sql | 44 ++++++++++++++++------------- src/initiate_transaction.rs | 35 +++++++++++------------ src/main.rs | 4 ++- 3 files changed, 43 insertions(+), 40 deletions(-) diff --git a/sql/delete_indexed_initiated_tx.sql b/sql/delete_indexed_initiated_tx.sql index 01f51d8..b465241 100644 --- a/sql/delete_indexed_initiated_tx.sql +++ b/sql/delete_indexed_initiated_tx.sql @@ -1,21 +1,25 @@ -DELETE FROM initiated_transactions +DELETE FROM initiated_transactions it WHERE --- Clean up initiate records when source tx is indexed -(tx_type = 'initiate' AND direction = 'EthAvail' AND EXISTS ( - SELECT 1 FROM bridge_event be - WHERE be.source_transaction_hash = initiated_transactions.source_transaction_hash -)) -OR (tx_type = 'initiate' AND direction = 'AvailEth' AND EXISTS ( - SELECT 1 FROM avail_indexer ai - WHERE ai.ext_hash = initiated_transactions.source_transaction_hash -)) --- Clean up claim records when claim tx is indexed -OR (tx_type = 'claim' AND direction = 'AvailEth' AND EXISTS ( - SELECT 1 FROM bridge_event be - WHERE be.message_id::text = initiated_transactions.message_id - AND be.event_type = 'MessageReceived' -)) -OR (tx_type = 'claim' AND direction = 'EthAvail' AND EXISTS ( - SELECT 1 FROM avail_execute_table aet - WHERE aet.message_id::text = initiated_transactions.message_id -)) +-- Scope cleanup to the senders passed into the transactions query. +(it.sender = $1 OR it.sender = $2) +AND ( + -- Clean up initiate records when source tx is indexed. + (it.tx_type = 'initiate' AND it.direction = 'EthAvail' AND EXISTS ( + SELECT 1 FROM bridge_event be + WHERE be.source_transaction_hash = it.source_transaction_hash + )) + OR (it.tx_type = 'initiate' AND it.direction = 'AvailEth' AND EXISTS ( + SELECT 1 FROM avail_indexer ai + WHERE ai.ext_hash = it.source_transaction_hash + )) + -- Clean up claim records when claim tx is indexed. + OR (it.tx_type = 'claim' AND it.direction = 'AvailEth' AND EXISTS ( + SELECT 1 FROM bridge_event be + WHERE be.message_id::text = it.message_id + AND be.event_type = 'MessageReceived' + )) + OR (it.tx_type = 'claim' AND it.direction = 'EthAvail' AND EXISTS ( + SELECT 1 FROM avail_execute_table aet + WHERE aet.message_id::text = it.message_id + )) +) diff --git a/src/initiate_transaction.rs b/src/initiate_transaction.rs index 6276617..1156fd3 100644 --- a/src/initiate_transaction.rs +++ b/src/initiate_transaction.rs @@ -1,27 +1,25 @@ +use crate::AppState; use crate::models::{ErrorResponse, InitiateRequest}; -use crate::{ - AppState, -}; -use alloy::primitives::{hex, U256}; +use alloy::primitives::{U256, hex}; +use alloy::signers::k256::sha2::Digest; use alloy::sol_types::SolCall; use anyhow::anyhow; +use avail_core::data_proof::Message; +use axum::response::Response; use axum::{ extract::{Json, State}, http::StatusCode, response::IntoResponse, }; -use serde_json::{json, Value}; -use std::sync::Arc; -use alloy::signers::k256::sha2::Digest; -use avail_core::data_proof::Message; -use axum::response::Response; use jsonrpsee::core::client::ClientT; use jsonrpsee::http_client::HttpClient; use jsonrpsee::rpc_params; use parity_scale_codec::{Compact, Decode}; +use serde_json::{Value, json}; use sha3::Keccak256; -use sp_core::{blake2_256, twox_128}; use sp_core::crypto::{AccountId32, Ss58Codec}; +use sp_core::{blake2_256, twox_128}; +use std::sync::Arc; pub(crate) async fn initiate_transaction( State(state): State>, Json(request): Json, @@ -104,7 +102,6 @@ pub(crate) async fn initiate_transaction( } } - async fn initiate_eth_transaction( state: &Arc, eth_tx_hash: &str, @@ -245,8 +242,8 @@ async fn initiate_eth_transaction( timestamp, "initiate", ) - .execute(&state.db) - .await?; + .execute(&state.db) + .await?; Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) } @@ -431,8 +428,8 @@ async fn initiate_avail_transaction( timestamp, "initiate", ) - .execute(&state.db) - .await?; + .execute(&state.db) + .await?; Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) } @@ -520,8 +517,8 @@ async fn claim_eth_transaction( timestamp, "claim", ) - .execute(&state.db) - .await?; + .execute(&state.db) + .await?; Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) } @@ -617,8 +614,8 @@ async fn claim_avail_transaction( timestamp, "claim", ) - .execute(&state.db) - .await?; + .execute(&state.db) + .await?; Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) } diff --git a/src/main.rs b/src/main.rs index 1726683..3cc9c6c 100755 --- a/src/main.rs +++ b/src/main.rs @@ -325,7 +325,9 @@ async fn transactions( } // Clean up initiated/claimed transactions that have been indexed - let _ = sqlx::query_file!("sql/delete_indexed_initiated_tx.sql") + let _ = sqlx::query(include_str!("../sql/delete_indexed_initiated_tx.sql")) + .bind(eth_addr_for_initiated.as_deref().unwrap_or("")) + .bind(avail_addr_for_initiated.as_deref().unwrap_or("")) .execute(&state.db) .await; From 1317c68290d9ac957d3a67cc70137468d0473752 Mon Sep 17 00:00:00 2001 From: Sasa Prsic Date: Mon, 2 Mar 2026 14:33:52 +0100 Subject: [PATCH 3/6] tidy tx deleted. --- .env.example | 4 +- ...=> delete_indexed_initiated_tx_global.sql} | 6 +- src/main.rs | 60 ++++++++++++++++--- 3 files changed, 57 insertions(+), 13 deletions(-) rename sql/{delete_indexed_initiated_tx.sql => delete_indexed_initiated_tx_global.sql} (88%) diff --git a/.env.example b/.env.example index e196c61..bd37f3a 100755 --- a/.env.example +++ b/.env.example @@ -26,4 +26,6 @@ POSTGRES_DB= HELIOS_UPDATE_FREQUENCY= -VECTOR_UPDATE_FREQUENCY= \ No newline at end of file +VECTOR_UPDATE_FREQUENCY= + +INDEXED_TX_CLEANUP_INTERVAL_SECONDS= \ No newline at end of file diff --git a/sql/delete_indexed_initiated_tx.sql b/sql/delete_indexed_initiated_tx_global.sql similarity index 88% rename from sql/delete_indexed_initiated_tx.sql rename to sql/delete_indexed_initiated_tx_global.sql index b465241..d647c3b 100644 --- a/sql/delete_indexed_initiated_tx.sql +++ b/sql/delete_indexed_initiated_tx_global.sql @@ -1,8 +1,5 @@ DELETE FROM initiated_transactions it WHERE --- Scope cleanup to the senders passed into the transactions query. -(it.sender = $1 OR it.sender = $2) -AND ( -- Clean up initiate records when source tx is indexed. (it.tx_type = 'initiate' AND it.direction = 'EthAvail' AND EXISTS ( SELECT 1 FROM bridge_event be @@ -21,5 +18,4 @@ AND ( OR (it.tx_type = 'claim' AND it.direction = 'EthAvail' AND EXISTS ( SELECT 1 FROM avail_execute_table aet WHERE aet.message_id::text = it.message_id - )) -) + )); diff --git a/src/main.rs b/src/main.rs index 3cc9c6c..d153766 100755 --- a/src/main.rs +++ b/src/main.rs @@ -324,13 +324,6 @@ async fn transactions( } } - // Clean up initiated/claimed transactions that have been indexed - let _ = sqlx::query(include_str!("../sql/delete_indexed_initiated_tx.sql")) - .bind(eth_addr_for_initiated.as_deref().unwrap_or("")) - .bind(avail_addr_for_initiated.as_deref().unwrap_or("")) - .execute(&state.db) - .await; - transaction_data_results.sort_unstable_by(|a, b| b.timestamp.cmp(&a.timestamp)); Ok(( @@ -907,6 +900,33 @@ async fn fetch_chain_head(state: Arc, chain_id: u64) -> Result { } } +fn cleanup_interval_seconds(raw: Option) -> u64 { + raw.and_then(|v| v.parse::().ok()) + .filter(|seconds| *seconds > 0) + .unwrap_or(60) +} + +async fn cleanup_indexed_initiated_transactions(state: &Arc) -> Result<()> { + sqlx::query(include_str!( + "../sql/delete_indexed_initiated_tx_global.sql" + )) + .execute(&state.db) + .await?; + Ok(()) +} + +async fn cleanup_indexed_initiated_transactions_task( + state: Arc, + cleanup_interval_seconds: u64, +) { + loop { + if let Err(e) = cleanup_indexed_initiated_transactions(&state).await { + warn!("Indexed initiated tx cleanup failed: {e:#}"); + } + tokio::time::sleep(Duration::from_secs(cleanup_interval_seconds)).await; + } +} + #[tokio::main] async fn main() { dotenvy::dotenv().ok(); @@ -923,6 +943,8 @@ async fn main() { .ok() .and_then(|max_request| max_request.parse::().ok()) .unwrap_or(1024); + let indexed_tx_cleanup_interval_seconds = + cleanup_interval_seconds(env::var("INDEXED_TX_CLEANUP_INTERVAL_SECONDS").ok()); let db_url: String = env::var("POSTGRES_URL").unwrap_or("localhost:5432".to_owned()); // Connection pool @@ -1065,6 +1087,7 @@ async fn main() { .await .unwrap(); + let cleanup_state = shared_state.clone(); tokio::spawn(async move { tracing::info!("Starting head tracking task"); if let Err(e) = track_slot_avail_task(shared_state.clone()).await { @@ -1072,6 +1095,10 @@ async fn main() { process::exit(1); } }); + tokio::spawn(cleanup_indexed_initiated_transactions_task( + cleanup_state, + indexed_tx_cleanup_interval_seconds, + )); tracing::info!("🚀 Started server on host {} with port {}", host, port); axum::serve(listener, app).await.unwrap(); } @@ -1230,3 +1257,22 @@ fn test_remaining_time_for_helios_update() { let remaining = time_until_next_helios_update(update as u64, 3600); assert_eq!(2400, remaining.as_secs()); } + +#[test] +fn cleanup_interval_defaults_to_60_seconds() { + assert_eq!(cleanup_interval_seconds(None), 60); +} + +#[test] +fn cleanup_interval_uses_positive_value() { + assert_eq!(cleanup_interval_seconds(Some("15".to_string())), 15); +} + +#[test] +fn cleanup_interval_rejects_invalid_or_zero() { + assert_eq!(cleanup_interval_seconds(Some("0".to_string())), 60); + assert_eq!( + cleanup_interval_seconds(Some("not-a-number".to_string())), + 60 + ); +} From b18cdfeccff3e61664b3d258c765fd83256046b0 Mon Sep 17 00:00:00 2001 From: Sasa Prsic Date: Mon, 2 Mar 2026 14:46:10 +0100 Subject: [PATCH 4/6] Add message id validation. --- src/initiate_transaction.rs | 145 ++++++++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) diff --git a/src/initiate_transaction.rs b/src/initiate_transaction.rs index 1156fd3..9deb035 100644 --- a/src/initiate_transaction.rs +++ b/src/initiate_transaction.rs @@ -54,6 +54,9 @@ pub(crate) async fn initiate_transaction( )); } } + if let Some(message_id) = request.message_id.as_deref() { + validate_message_id(message_id)?; + } // Check if already processed to avoid redundant RPC calls (skip for claims) if request.message_id.is_none() { @@ -439,6 +442,8 @@ async fn claim_eth_transaction( eth_tx_hash: &str, message_id: &str, ) -> anyhow::Result { + let requested_message_id = validate_message_id(message_id)?; + // Verify tx exists let receipt: Value = state .ethereum_client @@ -458,6 +463,9 @@ async fn claim_eth_transaction( StatusCode::NOT_FOUND, )); } + let receipt_message_id = + extract_eth_claim_message_id(&receipt, state.bridge_contract_address.as_str())?; + ensure_message_id_matches(requested_message_id, receipt_message_id)?; let block_hash = receipt["blockHash"] .as_str() @@ -523,12 +531,74 @@ async fn claim_eth_transaction( Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) } +fn validate_message_id(message_id: &str) -> anyhow::Result { + message_id.parse::().map_err(|_| { + ErrorResponse::with_status( + anyhow!("Invalid messageId format, expected unsigned integer"), + StatusCode::BAD_REQUEST, + ) + }) +} + +fn ensure_message_id_matches(requested: U256, derived: U256) -> anyhow::Result<(), ErrorResponse> { + if requested == derived { + Ok(()) + } else { + Err(ErrorResponse::with_status( + anyhow!("Provided messageId does not match claim transaction"), + StatusCode::BAD_REQUEST, + )) + } +} + +fn extract_eth_claim_message_id( + receipt: &Value, + bridge_contract_address: &str, +) -> anyhow::Result { + let logs = receipt["logs"].as_array().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing logs"), StatusCode::BAD_REQUEST) + })?; + + let claim_log = logs + .iter() + .find(|log| { + log["address"] + .as_str() + .map(|address| address.eq_ignore_ascii_case(bridge_contract_address)) + .unwrap_or(false) + }) + .ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("Claim event not found in transaction"), + StatusCode::BAD_REQUEST, + ) + })?; + + let data_hex = claim_log["data"].as_str().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing claim event data"), StatusCode::BAD_REQUEST) + })?; + let data_bytes = hex::decode(data_hex.trim_start_matches("0x")).map_err(|_| { + ErrorResponse::with_status(anyhow!("Invalid claim event data"), StatusCode::BAD_REQUEST) + })?; + + if data_bytes.len() < 32 { + return Err(ErrorResponse::with_status( + anyhow!("Invalid claim event data"), + StatusCode::BAD_REQUEST, + )); + } + + Ok(U256::from_be_slice(&data_bytes[data_bytes.len() - 32..])) +} + async fn claim_avail_transaction( state: &Arc, block_number: u32, tx_index: u32, message_id: &str, ) -> anyhow::Result { + let requested_message_id = validate_message_id(message_id)?; + let block_hash: String = state .avail_client .request("chain_getBlockHash", rpc_params![block_number]) @@ -573,6 +643,8 @@ async fn claim_avail_transaction( })?; let ext_hash_hex = format!("0x{}", hex::encode(blake2_256(&ext_bytes))); + let indexed_message_id = fetch_avail_claim_message_id_by_ext_hash(state, &ext_hash_hex).await?; + ensure_message_id_matches(requested_message_id, indexed_message_id)?; // Decode sender from extrinsic let mut cursor = &ext_bytes[..]; @@ -619,3 +691,76 @@ async fn claim_avail_transaction( Ok((StatusCode::OK, Json(json!({"status": "ok"}))).into_response()) } + +async fn fetch_avail_claim_message_id_by_ext_hash( + state: &Arc, + ext_hash: &str, +) -> anyhow::Result { + let message_id_text: Option = sqlx::query_scalar( + r#" +SELECT aet.message_id::text +FROM avail_execute_table aet +JOIN avail_indexer ai ON ai.id = aet.id +WHERE ai.ext_hash = $1 +LIMIT 1 + "#, + ) + .bind(ext_hash) + .fetch_optional(&state.db) + .await?; + + let message_id_text = message_id_text.ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("Unable to validate claim messageId from fetched transaction"), + StatusCode::BAD_REQUEST, + ) + })?; + + validate_message_id(&message_id_text) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn validate_message_id_accepts_decimal_u256() { + assert!(validate_message_id("12965243546238977").is_ok()); + } + + #[test] + fn validate_message_id_rejects_non_numeric() { + assert!(validate_message_id("abc123").is_err()); + } + + #[test] + fn eth_claim_message_id_must_match_receipt() { + let receipt = json!({ + "logs": [{ + "address": "0x967F7DdC4ec508462231849AE81eeaa68Ad01389", + "data": "0x000000000000000000000000000000000000000000000000002e0fd200000001" + }] + }); + + let expected = + extract_eth_claim_message_id(&receipt, "0x967F7DdC4ec508462231849AE81eeaa68Ad01389") + .map(|v| v.to_string()); + + assert_eq!(expected.ok(), Some("12965243546238977".to_string())); + } + + #[test] + fn message_id_match_accepts_equal_values() { + let requested = U256::from(42); + let derived = U256::from(42); + assert!(ensure_message_id_matches(requested, derived).is_ok()); + } + + #[test] + fn message_id_match_rejects_mismatch() { + let requested = U256::from(42); + let derived = U256::from(43); + assert!(ensure_message_id_matches(requested, derived).is_err()); + } +} From 467ab2aa5921403fbb55ee1b6237893ba72ba00a Mon Sep 17 00:00:00 2001 From: Sasa Prsic Date: Thu, 5 Mar 2026 12:58:39 +0100 Subject: [PATCH 5/6] tidy. --- src/initiate_transaction.rs | 479 +++++++++++++++++++++++------------- src/main.rs | 162 ++++++++---- 2 files changed, 429 insertions(+), 212 deletions(-) diff --git a/src/initiate_transaction.rs b/src/initiate_transaction.rs index 9deb035..df31349 100644 --- a/src/initiate_transaction.rs +++ b/src/initiate_transaction.rs @@ -117,7 +117,7 @@ async fn initiate_eth_transaction( tracing::error!("Failed to fetch ETH receipt: {e:#}"); ErrorResponse::with_status( anyhow!("Failed to fetch transaction receipt"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) })?; @@ -267,13 +267,13 @@ async fn fetch_avail_block_timestamp( tracing::error!("Failed to fetch Avail block timestamp: {e:#}"); ErrorResponse::with_status( anyhow!("Failed to fetch block timestamp"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) })?; let bytes = hex::decode(storage_hex.trim_start_matches("0x")).map_err(|_| { ErrorResponse::with_status( anyhow!("Invalid timestamp encoding"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) })?; let timestamp_bytes: [u8; 8] = @@ -283,18 +283,44 @@ async fn fetch_avail_block_timestamp( .ok_or_else(|| { ErrorResponse::with_status( anyhow!("Invalid timestamp length"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) })?; let timestamp_ms = u64::from_le_bytes(timestamp_bytes); Ok((timestamp_ms / 1000) as i64) } -async fn initiate_avail_transaction( +struct AvailBlockExtrinsic { + block_hash: String, + ext_hex: String, +} + +struct ParsedAvailExtrinsic { + bytes: Vec, + hash_hex: String, + sender: String, + call_offset: usize, +} + +enum AvailSenderMode { + Strict, + Lenient, +} + +const VECTOR_PALLET_ID: u8 = 39; +const VECTOR_SEND_MESSAGE_CALL_ID: u8 = 3; + +struct DecodedSendMessageCall { + message: Message, + to: sp_core::H256, + domain: u32, +} + +async fn fetch_avail_extrinsic_at_index( state: &Arc, block_number: u32, tx_index: u32, -) -> anyhow::Result { +) -> anyhow::Result { let block_hash: String = state .avail_client .request("chain_getBlockHash", rpc_params![block_number]) @@ -303,22 +329,21 @@ async fn initiate_avail_transaction( tracing::error!("Failed to fetch Avail block hash: {e:#}"); ErrorResponse::with_status( anyhow!("Failed to fetch block hash"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) })?; - // Fetch block to get the extrinsic and extract signer let block: Value = state .avail_client .request("chain_getBlock", rpc_params![&block_hash]) .await .map_err(|e| { tracing::error!("Failed to fetch Avail block: {e:#}"); - ErrorResponse::with_status(anyhow!("Failed to fetch block"), StatusCode::BAD_GATEWAY) + ErrorResponse::with_status(anyhow!("Failed to fetch block"), StatusCode::BAD_REQUEST) })?; let extrinsics = block["block"]["extrinsics"].as_array().ok_or_else(|| { - ErrorResponse::with_status(anyhow!("Missing extrinsics"), StatusCode::BAD_GATEWAY) + ErrorResponse::with_status(anyhow!("Missing extrinsics"), StatusCode::BAD_REQUEST) })?; let ext_hex = extrinsics @@ -329,103 +354,154 @@ async fn initiate_avail_transaction( anyhow!("Extrinsic not found at index {tx_index}"), StatusCode::NOT_FOUND, ) - })?; + })? + .to_string(); - let ext_bytes = hex::decode(ext_hex.trim_start_matches("0x")).map_err(|_| { + Ok(AvailBlockExtrinsic { + block_hash, + ext_hex, + }) +} + +fn parse_avail_extrinsic( + ext_hex: &str, + mode: AvailSenderMode, +) -> anyhow::Result { + let bytes = hex::decode(ext_hex.trim_start_matches("0x")).map_err(|_| { ErrorResponse::with_status( anyhow!("Invalid extrinsic encoding"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) })?; - // Extrinsic hash matching indexer's ext.metadata.ext_hash - let ext_hash_hex = format!("0x{}", hex::encode(blake2_256(&ext_bytes))); - - // Decode extrinsic using SCALE codec cursor - let mut cursor = &ext_bytes[..]; + let hash_hex = format!("0x{}", hex::encode(blake2_256(&bytes))); + let mut cursor = &bytes[..]; let decode_err = |e| { ErrorResponse::with_status( anyhow!("Failed to decode extrinsic: {e}"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) }; - // Skip compact length prefix >::decode(&mut cursor).map_err(decode_err)?; - - // Version byte (0x84 = V4 signed) let version = u8::decode(&mut cursor).map_err(decode_err)?; + let mut sender = String::new(); + let mut call_offset = bytes.len() - cursor.len(); - // Extract signer from signed extrinsic - let sender = if version & 0x80 != 0 { - // MultiAddress::Id = 0x00 + 32 bytes AccountId + if version & 0x80 != 0 { let addr_type = u8::decode(&mut cursor).map_err(decode_err)?; if addr_type != 0x00 { - return Err(ErrorResponse::with_status( - anyhow!("Unsupported address type"), - StatusCode::BAD_GATEWAY, - )); + return match mode { + AvailSenderMode::Lenient => Ok(ParsedAvailExtrinsic { + bytes, + hash_hex, + sender, + call_offset, + }), + AvailSenderMode::Strict => Err(ErrorResponse::with_status( + anyhow!("Unsupported address type"), + StatusCode::BAD_REQUEST, + )), + }; } - let account = <[u8; 32]>::decode(&mut cursor).map_err(decode_err)?; - let sender = AccountId32::new(account).to_ss58check(); - // Skip MultiSignature - let sig_type = u8::decode(&mut cursor).map_err(decode_err)?; - match sig_type { - 0x00 | 0x01 => { - <[u8; 64]>::decode(&mut cursor).map_err(decode_err)?; - } - 0x02 => { - <[u8; 65]>::decode(&mut cursor).map_err(decode_err)?; - } - _ => { - return Err(ErrorResponse::with_status( - anyhow!("Unknown signature type"), - StatusCode::BAD_GATEWAY, - )); + let account = <[u8; 32]>::decode(&mut cursor).map_err(decode_err)?; + sender = AccountId32::new(account).to_ss58check(); + + if matches!(mode, AvailSenderMode::Strict) { + let sig_type = u8::decode(&mut cursor).map_err(decode_err)?; + match sig_type { + 0x00 | 0x01 => { + <[u8; 64]>::decode(&mut cursor).map_err(decode_err)?; + } + 0x02 => { + <[u8; 65]>::decode(&mut cursor).map_err(decode_err)?; + } + _ => { + return Err(ErrorResponse::with_status( + anyhow!("Unknown signature type"), + StatusCode::BAD_REQUEST, + )); + } } - }; - // Skip SignedExtras: Era + Nonce + Tip + AppId - let era = u8::decode(&mut cursor).map_err(decode_err)?; - if era != 0x00 { - u8::decode(&mut cursor).map_err(decode_err)?; // mortal era second byte + let era = u8::decode(&mut cursor).map_err(decode_err)?; + if era != 0x00 { + u8::decode(&mut cursor).map_err(decode_err)?; + } + >::decode(&mut cursor).map_err(decode_err)?; + >::decode(&mut cursor).map_err(decode_err)?; + >::decode(&mut cursor).map_err(decode_err)?; } - >::decode(&mut cursor).map_err(decode_err)?; // nonce - >::decode(&mut cursor).map_err(decode_err)?; // tip - >::decode(&mut cursor).map_err(decode_err)?; // app_id - sender - } else { - String::new() + call_offset = bytes.len() - cursor.len(); + } + + Ok(ParsedAvailExtrinsic { + bytes, + hash_hex, + sender, + call_offset, + }) +} + +fn decode_send_message_call( + call_bytes: &[u8], +) -> anyhow::Result { + let mut cursor = call_bytes; + let decode_err = |e| { + ErrorResponse::with_status( + anyhow!("Failed to decode extrinsic: {e}"), + StatusCode::BAD_REQUEST, + ) }; - // Skip pallet_id + call_id - <[u8; 2]>::decode(&mut cursor).map_err(decode_err)?; + let header = <[u8; 2]>::decode(&mut cursor).map_err(decode_err)?; + if header != [VECTOR_PALLET_ID, VECTOR_SEND_MESSAGE_CALL_ID] { + return Err(ErrorResponse::with_status( + anyhow!("Transaction is not a Vector::SendMessage call"), + StatusCode::BAD_REQUEST, + )); + } - // Decode SendMessage params: Message + to (H256) let message = Message::decode(&mut cursor).map_err(decode_err)?; let to = sp_core::H256::decode(&mut cursor).map_err(decode_err)?; + let domain = >::decode(&mut cursor).map_err(decode_err)?.0; + + Ok(DecodedSendMessageCall { + message, + to, + domain, + }) +} - let amount = match &message { +async fn initiate_avail_transaction( + state: &Arc, + block_number: u32, + tx_index: u32, +) -> anyhow::Result { + let avail_ext = fetch_avail_extrinsic_at_index(state, block_number, tx_index).await?; + let parsed_ext = parse_avail_extrinsic(&avail_ext.ext_hex, AvailSenderMode::Strict)?; + let call = decode_send_message_call(&parsed_ext.bytes[parsed_ext.call_offset..])?; + let amount = match &call.message { Message::FungibleToken { amount, .. } => amount.to_string(), _ => "0".to_string(), }; - let receiver = format!("0x{}", hex::encode(to.as_bytes())); + let receiver = format!("0x{}", hex::encode(call.to.as_bytes())); // ID matching indexer's (block_height << 32) | ext_index let message_id = ((block_number as u64) << 32 | tx_index as u64).to_string(); - let timestamp = fetch_avail_block_timestamp(&state.avail_client, &block_hash).await?; + let timestamp = fetch_avail_block_timestamp(&state.avail_client, &avail_ext.block_hash).await?; sqlx::query_file!( "sql/insert_initiated_tx.sql", - &ext_hash_hex, + &parsed_ext.hash_hex, "AvailEth", &message_id, - &sender, + &parsed_ext.sender, &receiver, &amount, - &block_hash, + &avail_ext.block_hash, block_number as i32, tx_index as i32, timestamp, @@ -453,7 +529,7 @@ async fn claim_eth_transaction( tracing::error!("Failed to fetch ETH receipt: {e:#}"); ErrorResponse::with_status( anyhow!("Failed to fetch transaction receipt"), - StatusCode::BAD_GATEWAY, + StatusCode::BAD_REQUEST, ) })?; @@ -481,19 +557,7 @@ async fn claim_eth_transaction( ErrorResponse::with_status(anyhow!("Invalid blockNumber"), StatusCode::BAD_REQUEST) })?; - let tx: Value = state - .ethereum_client - .request("eth_getTransactionByHash", rpc_params![eth_tx_hash]) - .await - .map_err(|e| { - tracing::error!("Failed to fetch ETH transaction: {e:#}"); - ErrorResponse::with_status( - anyhow!("Failed to fetch transaction"), - StatusCode::BAD_REQUEST, - ) - })?; - - let sender = tx["from"].as_str().unwrap_or("").to_string(); + let sender = extract_eth_claim_sender(&receipt, state.bridge_contract_address.as_str())?; let block: Value = state .ethereum_client @@ -555,24 +619,7 @@ fn extract_eth_claim_message_id( receipt: &Value, bridge_contract_address: &str, ) -> anyhow::Result { - let logs = receipt["logs"].as_array().ok_or_else(|| { - ErrorResponse::with_status(anyhow!("Missing logs"), StatusCode::BAD_REQUEST) - })?; - - let claim_log = logs - .iter() - .find(|log| { - log["address"] - .as_str() - .map(|address| address.eq_ignore_ascii_case(bridge_contract_address)) - .unwrap_or(false) - }) - .ok_or_else(|| { - ErrorResponse::with_status( - anyhow!("Claim event not found in transaction"), - StatusCode::BAD_REQUEST, - ) - })?; + let claim_log = find_eth_claim_log(receipt, bridge_contract_address)?; let data_hex = claim_log["data"].as_str().ok_or_else(|| { ErrorResponse::with_status(anyhow!("Missing claim event data"), StatusCode::BAD_REQUEST) @@ -591,96 +638,95 @@ fn extract_eth_claim_message_id( Ok(U256::from_be_slice(&data_bytes[data_bytes.len() - 32..])) } -async fn claim_avail_transaction( - state: &Arc, - block_number: u32, - tx_index: u32, - message_id: &str, -) -> anyhow::Result { - let requested_message_id = validate_message_id(message_id)?; +fn extract_eth_claim_sender( + receipt: &Value, + bridge_contract_address: &str, +) -> anyhow::Result { + let claim_log = find_eth_claim_log(receipt, bridge_contract_address)?; + let sender_topic = claim_log["topics"][1].as_str().ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("Missing claim event sender topic"), + StatusCode::BAD_REQUEST, + ) + })?; - let block_hash: String = state - .avail_client - .request("chain_getBlockHash", rpc_params![block_number]) - .await - .map_err(|e| { - tracing::error!("Failed to fetch Avail block hash: {e:#}"); - ErrorResponse::with_status( - anyhow!("Failed to fetch block hash"), - StatusCode::BAD_GATEWAY, - ) - })?; + let sender_topic_hex = sender_topic.strip_prefix("0x").ok_or_else(|| { + ErrorResponse::with_status( + anyhow!("Invalid claim event sender topic"), + StatusCode::BAD_REQUEST, + ) + })?; + if sender_topic_hex.len() != 64 { + return Err(ErrorResponse::with_status( + anyhow!("Invalid claim event sender topic"), + StatusCode::BAD_REQUEST, + )); + } - // Verify extrinsic exists - let block: Value = state - .avail_client - .request("chain_getBlock", rpc_params![&block_hash]) - .await - .map_err(|e| { - tracing::error!("Failed to fetch Avail block: {e:#}"); - ErrorResponse::with_status(anyhow!("Failed to fetch block"), StatusCode::BAD_GATEWAY) - })?; + let sender_bytes = hex::decode(sender_topic_hex).map_err(|_| { + ErrorResponse::with_status( + anyhow!("Invalid claim event sender topic"), + StatusCode::BAD_REQUEST, + ) + })?; + let sender_array: [u8; 32] = sender_bytes.try_into().map_err(|_| { + ErrorResponse::with_status( + anyhow!("Invalid claim event sender topic"), + StatusCode::BAD_REQUEST, + ) + })?; - let extrinsics = block["block"]["extrinsics"].as_array().ok_or_else(|| { - ErrorResponse::with_status(anyhow!("Missing extrinsics"), StatusCode::BAD_GATEWAY) + Ok(AccountId32::new(sender_array).to_ss58check()) +} + +fn find_eth_claim_log<'a>( + receipt: &'a Value, + bridge_contract_address: &str, +) -> anyhow::Result<&'a Value, ErrorResponse> { + let logs = receipt["logs"].as_array().ok_or_else(|| { + ErrorResponse::with_status(anyhow!("Missing logs"), StatusCode::BAD_REQUEST) })?; - let ext_hex = extrinsics - .get(tx_index as usize) - .and_then(|v| v.as_str()) + logs.iter() + .find(|log| { + log["address"] + .as_str() + .map(|address| address.eq_ignore_ascii_case(bridge_contract_address)) + .unwrap_or(false) + }) .ok_or_else(|| { ErrorResponse::with_status( - anyhow!("Extrinsic not found at index {tx_index}"), - StatusCode::NOT_FOUND, + anyhow!("Claim event not found in transaction"), + StatusCode::BAD_REQUEST, ) - })?; + }) +} - let ext_bytes = hex::decode(ext_hex.trim_start_matches("0x")).map_err(|_| { - ErrorResponse::with_status( - anyhow!("Invalid extrinsic encoding"), - StatusCode::BAD_GATEWAY, - ) - })?; +async fn claim_avail_transaction( + state: &Arc, + block_number: u32, + tx_index: u32, + message_id: &str, +) -> anyhow::Result { + let requested_message_id = validate_message_id(message_id)?; - let ext_hash_hex = format!("0x{}", hex::encode(blake2_256(&ext_bytes))); - let indexed_message_id = fetch_avail_claim_message_id_by_ext_hash(state, &ext_hash_hex).await?; + let avail_ext = fetch_avail_extrinsic_at_index(state, block_number, tx_index).await?; + let parsed_ext = parse_avail_extrinsic(&avail_ext.ext_hex, AvailSenderMode::Lenient)?; + let indexed_message_id = + fetch_avail_claim_message_id_by_ext_hash(state, &parsed_ext.hash_hex).await?; ensure_message_id_matches(requested_message_id, indexed_message_id)?; - // Decode sender from extrinsic - let mut cursor = &ext_bytes[..]; - let decode_err = |e| { - ErrorResponse::with_status( - anyhow!("Failed to decode extrinsic: {e}"), - StatusCode::BAD_GATEWAY, - ) - }; - - >::decode(&mut cursor).map_err(decode_err)?; - let version = u8::decode(&mut cursor).map_err(decode_err)?; - - let sender = if version & 0x80 != 0 { - let addr_type = u8::decode(&mut cursor).map_err(decode_err)?; - if addr_type != 0x00 { - String::new() - } else { - let account = <[u8; 32]>::decode(&mut cursor).map_err(decode_err)?; - AccountId32::new(account).to_ss58check() - } - } else { - String::new() - }; - - let timestamp = fetch_avail_block_timestamp(&state.avail_client, &block_hash).await?; + let timestamp = fetch_avail_block_timestamp(&state.avail_client, &avail_ext.block_hash).await?; sqlx::query_file!( "sql/insert_initiated_tx.sql", - &ext_hash_hex, + &parsed_ext.hash_hex, "EthAvail", message_id, - &sender, + &parsed_ext.sender, "", "0", - &block_hash, + &avail_ext.block_hash, block_number as i32, tx_index as i32, timestamp, @@ -722,6 +768,7 @@ LIMIT 1 #[cfg(test)] mod tests { use super::*; + use parity_scale_codec::Encode; use serde_json::json; #[test] @@ -750,6 +797,29 @@ mod tests { assert_eq!(expected.ok(), Some("12965243546238977".to_string())); } + #[test] + fn eth_claim_sender_is_decoded_from_claim_event_topic() { + let receipt = json!({ + "logs": [{ + "address": "0x967F7DdC4ec508462231849AE81eeaa68Ad01389", + "topics": [ + "0x4ad8286366216a121ffbecdd11163a134fc364cdf7cc99aae4cc3221d8d92269", + "0xcc2fd60dbb2ffedcab868872ea8d8c532759025bb1a9b26c5571dea8da223a3f", + "0x00000000000000000000000048e7e157cf873c15a5a6734ea37c000e1cb2383d" + ], + "data": "0x000000000000000000000000000000000000000000000000002e0fd200000001" + }] + }); + + let sender = + extract_eth_claim_sender(&receipt, "0x967F7DdC4ec508462231849AE81eeaa68Ad01389"); + + assert_eq!( + sender.ok(), + Some("5GgRqSNN1zTsjA6N7cofcdP9yewA6JG83S649HbuBut8MG4o".to_string()) + ); + } + #[test] fn message_id_match_accepts_equal_values() { let requested = U256::from(42); @@ -763,4 +833,77 @@ mod tests { let derived = U256::from(43); assert!(ensure_message_id_matches(requested, derived).is_err()); } + + fn hex_from_extrinsic_body(body: Vec) -> String { + let mut ext = Compact(body.len() as u32).encode(); + ext.extend_from_slice(&body); + format!("0x{}", hex::encode(ext)) + } + + #[test] + fn parse_avail_extrinsic_strict_extracts_sender_and_call_offset() { + let mut body = vec![0x84, 0x00]; + body.extend_from_slice(&[0x11; 32]); + body.push(0x00); + body.extend_from_slice(&[0x22; 64]); + body.push(0x00); + body.extend_from_slice(&Compact(0u32).encode()); + body.extend_from_slice(&Compact(0u128).encode()); + body.extend_from_slice(&Compact(0u32).encode()); + body.extend_from_slice(&[0x09, 0x00, 0xAA, 0xBB]); + + let ext_hex = hex_from_extrinsic_body(body); + let parsed_result = parse_avail_extrinsic(&ext_hex, AvailSenderMode::Strict); + assert!(parsed_result.is_ok()); + let parsed = parsed_result.ok().unwrap(); + + assert_eq!(parsed.sender, AccountId32::new([0x11; 32]).to_ss58check()); + assert_eq!( + &parsed.bytes[parsed.call_offset..parsed.call_offset + 2], + &[0x09, 0x00] + ); + assert!(parsed.hash_hex.starts_with("0x")); + } + + #[test] + fn parse_avail_extrinsic_lenient_allows_non_accountid_address_type() { + let ext_hex = hex_from_extrinsic_body(vec![0x84, 0x01]); + let parsed_result = parse_avail_extrinsic(&ext_hex, AvailSenderMode::Lenient); + assert!(parsed_result.is_ok()); + let parsed = parsed_result.ok().unwrap(); + + assert_eq!(parsed.sender, ""); + assert!(parsed.hash_hex.starts_with("0x")); + } + + #[test] + fn parse_avail_extrinsic_strict_rejects_non_accountid_address_type() { + let ext_hex = hex_from_extrinsic_body(vec![0x84, 0x01]); + assert!(parse_avail_extrinsic(&ext_hex, AvailSenderMode::Strict).is_err()); + } + + #[test] + fn decode_send_message_call_rejects_non_send_message_call() { + let call = vec![0x09, 0x00, 0xAA, 0xBB]; + assert!(decode_send_message_call(&call).is_err()); + } + + #[test] + fn decode_send_message_call_decodes_valid_vector_send_message() { + use parity_scale_codec::Encode; + + let message = Message::FungibleToken { + asset_id: sp_core::H256::from([0xAB; 32]), + amount: 42, + }; + let mut call = vec![VECTOR_PALLET_ID, VECTOR_SEND_MESSAGE_CALL_ID]; + call.extend_from_slice(&message.encode()); + call.extend_from_slice(&sp_core::H256::from([0xCD; 32]).encode()); + call.extend_from_slice(&Compact(7u32).encode()); + + let decoded_result = decode_send_message_call(&call); + assert!(decoded_result.is_ok()); + let decoded = decoded_result.ok().unwrap(); + assert_eq!(decoded.domain, 7); + } } diff --git a/src/main.rs b/src/main.rs index d153766..d7085cd 100755 --- a/src/main.rs +++ b/src/main.rs @@ -197,7 +197,16 @@ async fn transactions( .fetch_all(&state.db) .await?; - let range_blocks = fetch_range_blocks(&state).await?; + let range_blocks = match fetch_range_blocks(&state).await { + Ok(range_blocks) => Some(range_blocks), + Err(err) => { + tracing::warn!( + "Failed to fetch range blocks for /v1/transactions: {:#}", + err.error + ); + None + } + }; let header: Value = state .avail_client @@ -210,32 +219,40 @@ async fn transactions( })?; let latest_block_number = u32::from_str_radix(&number_hex[2..], 16).unwrap_or(0); - let claim_estimate = time_until_next_vector_update( - latest_block_number, - range_blocks.data.end, - state.vector_update_frequency, - 20, - ); - - tracing::info!( - latest_block_number = latest_block_number, - range_end = range_blocks.data.end, - blocks_since = latest_block_number.saturating_sub(range_blocks.data.end), - "time_until_next_vector_update" - ); + let claim_estimate = range_blocks.as_ref().map(|range_blocks| { + time_until_next_vector_update( + latest_block_number, + range_blocks.data.end, + state.vector_update_frequency, + 20, + ) + }); + + if let Some(range_blocks) = range_blocks.as_ref() { + tracing::info!( + latest_block_number = latest_block_number, + range_end = range_blocks.data.end, + blocks_since = latest_block_number.saturating_sub(range_blocks.data.end), + "time_until_next_vector_update" + ); + } for mut tx in transactions { let mut estimate = None; - if tx.final_status == BridgeStatusEnum::InProgress - && tx.source_block_height < range_blocks.data.end as i32 - { - tx.final_status = BridgeStatusEnum::ClaimReady; + if let Some(range_blocks) = range_blocks.as_ref() { + if tx.final_status == BridgeStatusEnum::InProgress + && tx.source_block_height < range_blocks.data.end as i32 + { + tx.final_status = BridgeStatusEnum::ClaimReady; + } } if tx.final_status == BridgeStatusEnum::Initiated || tx.final_status == BridgeStatusEnum::InProgress { - estimate = Some(claim_estimate.as_secs()); + if let Some(claim_estimate) = claim_estimate { + estimate = Some(claim_estimate.as_secs()); + } } transaction_data_results.push(TransactionData::new( @@ -334,6 +351,48 @@ async fn transactions( .into_response()) } +fn summarize_response_body(body: &str) -> String { + let trimmed = body.trim(); + if trimmed.is_empty() { + return "".to_string(); + } + + const MAX_CHARS: usize = 256; + let mut snippet: String = trimmed.chars().take(MAX_CHARS).collect(); + if trimmed.chars().count() > MAX_CHARS { + snippet.push_str("..."); + } + snippet +} + +fn parse_range_blocks_response( + status: StatusCode, + body: &str, +) -> Result { + let body_summary = summarize_response_body(body); + + if !status.is_success() { + let reason = status.canonical_reason().unwrap_or("unknown"); + return Err(ErrorResponse::with_status_and_headers( + anyhow!( + "Range service returned HTTP {} ({reason}) with body: {}", + status.as_u16(), + body_summary + ), + StatusCode::BAD_GATEWAY, + &[("Cache-Control", "public, max-age=60, must-revalidate")], + )); + } + + serde_json::from_str::(body).map_err(|e| { + ErrorResponse::with_status_and_headers( + anyhow!("Cannot decode range blocks response: {e}; body: {body_summary}"), + StatusCode::BAD_GATEWAY, + &[("Cache-Control", "public, max-age=60, must-revalidate")], + ) + }) +} + async fn fetch_range_blocks( state: &Arc, ) -> Result { @@ -355,23 +414,22 @@ async fn fetch_range_blocks( tracing::error!("Cannot parse range blocks: {e:#}"); ErrorResponse::with_status_and_headers( anyhow!("{e:#}"), - StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::BAD_GATEWAY, &[("Cache-Control", "public, max-age=60, must-revalidate")], ) })?; - let range_blocks = response - .json::() - .await - .map_err(|e| { - tracing::error!("Cannot parse range blocks: {e:#}"); - ErrorResponse::with_status_and_headers( - anyhow!("{e:#}"), - StatusCode::INTERNAL_SERVER_ERROR, - &[("Cache-Control", "public, max-age=60, must-revalidate")], - ) - })?; - Ok(range_blocks) + let status = response.status(); + let body = response.text().await.map_err(|e| { + tracing::error!("Cannot read range blocks response body: {e:#}"); + ErrorResponse::with_status_and_headers( + anyhow!("{e:#}"), + StatusCode::BAD_GATEWAY, + &[("Cache-Control", "public, max-age=60, must-revalidate")], + ) + })?; + + parse_range_blocks_response(status, &body) } #[inline(always)] @@ -626,22 +684,22 @@ async fn get_avl_head( tracing::error!("❌ Cannot parse range blocks: {e:#}"); ErrorResponse::with_status_and_headers( anyhow!("{e:#}"), - StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::BAD_GATEWAY, &[("Cache-Control", "public, max-age=60, must-revalidate")], ) })?; - let range_blocks = response - .json::() - .await - .map_err(|e| { - tracing::error!("❌ Cannot parse range blocks: {e:#}"); - ErrorResponse::with_status_and_headers( - anyhow!("{e:#}"), - StatusCode::INTERNAL_SERVER_ERROR, - &[("Cache-Control", "public, max-age=60, must-revalidate")], - ) - })?; + let status = response.status(); + let body = response.text().await.map_err(|e| { + tracing::error!("❌ Cannot read range blocks response body: {e:#}"); + ErrorResponse::with_status_and_headers( + anyhow!("{e:#}"), + StatusCode::BAD_GATEWAY, + &[("Cache-Control", "public, max-age=60, must-revalidate")], + ) + })?; + + let range_blocks = parse_range_blocks_response(status, &body)?; Ok(( StatusCode::OK, @@ -1276,3 +1334,19 @@ fn cleanup_interval_rejects_invalid_or_zero() { 60 ); } + +#[test] +fn parse_range_blocks_response_reports_upstream_status_with_empty_body() { + let err = match parse_range_blocks_response(StatusCode::INTERNAL_SERVER_ERROR, "") { + Ok(_) => panic!("expected parse_range_blocks_response to fail for 500 response"), + Err(err) => err, + }; + + assert_eq!(err.status_code, Some(StatusCode::BAD_GATEWAY)); + assert!( + err.error + .to_string() + .contains("Range service returned HTTP 500") + ); + assert!(err.error.to_string().contains("body: ")); +} From 954503cefa8d9a0a9f0cd4814d4fde15cf673d52 Mon Sep 17 00:00:00 2001 From: Sasa Prsic Date: Thu, 5 Mar 2026 16:04:25 +0100 Subject: [PATCH 6/6] add error response and move inline query to a file. --- .env.example | 2 +- ...d658cbc0081a377247a1ca8081cbfca33da68.json | 97 ------------------- ...3a02b40a360839afb9b27066a285efae0d5a.json} | 4 +- ...34fbc14670da24a3d52835e5e0dd2a2a2868.json} | 8 +- ...747d6cd8b3d61ebb3ec8929c224c89f120718.json | 12 +++ ...af73a022e7fecafb4a2eb119b089531c21084.json | 12 --- ...71b9712152c6473f7f74cb976f12850b25802.json | 22 +++++ .../0001_create_initiated_transactions.sql | 4 +- sql/check_initiated_tx_by_hash.sql | 2 +- sql/query_avail_claim_message_id.sql | 5 + sql/query_avail_tx.sql | 2 +- sql/query_claimed_tx.sql | 1 + src/initiate_transaction.rs | 17 +--- src/main.rs | 23 +++-- 14 files changed, 72 insertions(+), 139 deletions(-) delete mode 100644 .sqlx/query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json rename .sqlx/{query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json => query-4ef579ed62ff47d1f89f8c7ee3af3a02b40a360839afb9b27066a285efae0d5a.json} (74%) rename .sqlx/{query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json => query-56aca9fd618e4be1f56e5bae913934fbc14670da24a3d52835e5e0dd2a2a2868.json} (77%) create mode 100644 .sqlx/query-a754e183e8316aeeee5b0eb373a747d6cd8b3d61ebb3ec8929c224c89f120718.json delete mode 100644 .sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json create mode 100644 .sqlx/query-fcbac1d3922fdf9765c5ffca34d71b9712152c6473f7f74cb976f12850b25802.json create mode 100644 sql/query_avail_claim_message_id.sql diff --git a/.env.example b/.env.example index bd37f3a..1259e99 100755 --- a/.env.example +++ b/.env.example @@ -28,4 +28,4 @@ HELIOS_UPDATE_FREQUENCY= VECTOR_UPDATE_FREQUENCY= -INDEXED_TX_CLEANUP_INTERVAL_SECONDS= \ No newline at end of file +INDEXED_TX_CLEANUP_INTERVAL_SECONDS= diff --git a/.sqlx/query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json b/.sqlx/query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json deleted file mode 100644 index 729b5ac..0000000 --- a/.sqlx/query-0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68.json +++ /dev/null @@ -1,97 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT ai.id AS message_id,\n ai.signature_address AS sender,\n es.to AS \"receiver!\",\n COALESCE(es.amount, '0')::text AS \"amount!\",\n ai.block_hash AS source_block_hash,\n ai.ext_hash AS source_transaction_hash,\n ai.block_height AS source_block_height,\n ai.ext_index AS source_tx_index,\n ai.block_timestamp,\n be.source_transaction_hash AS \"destination_tx_hash?: String\",\n COALESCE(\n CASE\n WHEN be.message_id IS NOT NULL THEN 'bridged'::status\n END,\n 'in_progress'::status\n ) ::status AS \"final_status!: BridgeStatusEnum\"\nFROM avail_send_message_table es\n INNER JOIN public.avail_indexer AS ai\n ON ai.id = es.id\n LEFT JOIN public.bridge_event AS be\n ON es.id = be.message_id\nWHERE ai.signature_address = $1\n AND es.type = $2\n AND (be.event_type = $3 or be.event_type is null)\n AND ai.ext_success = $4\nORDER BY ai.block_timestamp DESC\nLIMIT 1000;\n", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "message_id", - "type_info": "Int8" - }, - { - "ordinal": 1, - "name": "sender", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "receiver!", - "type_info": "Text" - }, - { - "ordinal": 3, - "name": "amount!", - "type_info": "Text" - }, - { - "ordinal": 4, - "name": "source_block_hash", - "type_info": "Text" - }, - { - "ordinal": 5, - "name": "source_transaction_hash", - "type_info": "Text" - }, - { - "ordinal": 6, - "name": "source_block_height", - "type_info": "Int4" - }, - { - "ordinal": 7, - "name": "source_tx_index", - "type_info": "Int4" - }, - { - "ordinal": 8, - "name": "block_timestamp", - "type_info": "Int8" - }, - { - "ordinal": 9, - "name": "destination_tx_hash?: String", - "type_info": "Text" - }, - { - "ordinal": 10, - "name": "final_status!: BridgeStatusEnum", - "type_info": { - "Custom": { - "name": "status", - "kind": { - "Enum": [ - "initiated", - "in_progress", - "claim_ready", - "bridged" - ] - } - } - } - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Text", - "Bool" - ] - }, - "nullable": [ - false, - true, - false, - null, - false, - false, - false, - false, - false, - false, - null - ] - }, - "hash": "0518107277d77bf8b6b8f3d20d5d658cbc0081a377247a1ca8081cbfca33da68" -} diff --git a/.sqlx/query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json b/.sqlx/query-4ef579ed62ff47d1f89f8c7ee3af3a02b40a360839afb9b27066a285efae0d5a.json similarity index 74% rename from .sqlx/query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json rename to .sqlx/query-4ef579ed62ff47d1f89f8c7ee3af3a02b40a360839afb9b27066a285efae0d5a.json index 461f54d..36fb867 100644 --- a/.sqlx/query-2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20.json +++ b/.sqlx/query-4ef579ed62ff47d1f89f8c7ee3af3a02b40a360839afb9b27066a285efae0d5a.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_transaction_hash = $1)", + "query": "SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_transaction_hash = $1)\n", "describe": { "columns": [ { @@ -18,5 +18,5 @@ null ] }, - "hash": "2dca92e8ae9fbf417ee2e9090030daa4487a180d819a85980d6af0b09843de20" + "hash": "4ef579ed62ff47d1f89f8c7ee3af3a02b40a360839afb9b27066a285efae0d5a" } diff --git a/.sqlx/query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json b/.sqlx/query-56aca9fd618e4be1f56e5bae913934fbc14670da24a3d52835e5e0dd2a2a2868.json similarity index 77% rename from .sqlx/query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json rename to .sqlx/query-56aca9fd618e4be1f56e5bae913934fbc14670da24a3d52835e5e0dd2a2a2868.json index f7b6ba0..a0c237d 100644 --- a/.sqlx/query-dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09.json +++ b/.sqlx/query-56aca9fd618e4be1f56e5bae913934fbc14670da24a3d52835e5e0dd2a2a2868.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT message_id, source_transaction_hash, source_block_number, source_tx_index\nFROM initiated_transactions\nWHERE tx_type = 'claim'\n AND message_id = ANY($1)\n", + "query": "SELECT message_id, source_transaction_hash, source_block_number, source_tx_index\nFROM initiated_transactions\nWHERE tx_type = 'claim'\n AND message_id = ANY($1)\n AND (sender = $2 OR sender = $3)\n", "describe": { "columns": [ { @@ -26,7 +26,9 @@ ], "parameters": { "Left": [ - "TextArray" + "TextArray", + "Text", + "Text" ] }, "nullable": [ @@ -36,5 +38,5 @@ true ] }, - "hash": "dee2e7e7a3238412377519a16c05408336193be079337587d1b1ff6267c7bc09" + "hash": "56aca9fd618e4be1f56e5bae913934fbc14670da24a3d52835e5e0dd2a2a2868" } diff --git a/.sqlx/query-a754e183e8316aeeee5b0eb373a747d6cd8b3d61ebb3ec8929c224c89f120718.json b/.sqlx/query-a754e183e8316aeeee5b0eb373a747d6cd8b3d61ebb3ec8929c224c89f120718.json new file mode 100644 index 0000000..88ddbba --- /dev/null +++ b/.sqlx/query-a754e183e8316aeeee5b0eb373a747d6cd8b3d61ebb3ec8929c224c89f120718.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM initiated_transactions it\nWHERE\n -- Clean up initiate records when source tx is indexed.\n (it.tx_type = 'initiate' AND it.direction = 'EthAvail' AND EXISTS (\n SELECT 1 FROM bridge_event be\n WHERE be.source_transaction_hash = it.source_transaction_hash\n ))\n OR (it.tx_type = 'initiate' AND it.direction = 'AvailEth' AND EXISTS (\n SELECT 1 FROM avail_indexer ai\n WHERE ai.ext_hash = it.source_transaction_hash\n ))\n -- Clean up claim records when claim tx is indexed.\n OR (it.tx_type = 'claim' AND it.direction = 'AvailEth' AND EXISTS (\n SELECT 1 FROM bridge_event be\n WHERE be.message_id::text = it.message_id\n AND be.event_type = 'MessageReceived'\n ))\n OR (it.tx_type = 'claim' AND it.direction = 'EthAvail' AND EXISTS (\n SELECT 1 FROM avail_execute_table aet\n WHERE aet.message_id::text = it.message_id\n ));\n", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "a754e183e8316aeeee5b0eb373a747d6cd8b3d61ebb3ec8929c224c89f120718" +} diff --git a/.sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json b/.sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json deleted file mode 100644 index 635db59..0000000 --- a/.sqlx/query-fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM initiated_transactions\nWHERE\n-- Clean up initiate records when source tx is indexed\n(tx_type = 'initiate' AND direction = 'EthAvail' AND EXISTS (\n SELECT 1 FROM bridge_event be\n WHERE be.source_transaction_hash = initiated_transactions.source_transaction_hash\n))\nOR (tx_type = 'initiate' AND direction = 'AvailEth' AND EXISTS (\n SELECT 1 FROM avail_indexer ai\n WHERE ai.ext_hash = initiated_transactions.source_transaction_hash\n))\n-- Clean up claim records when claim tx is indexed\nOR (tx_type = 'claim' AND direction = 'AvailEth' AND EXISTS (\n SELECT 1 FROM bridge_event be\n WHERE be.message_id::text = initiated_transactions.message_id\n AND be.event_type = 'MessageReceived'\n))\nOR (tx_type = 'claim' AND direction = 'EthAvail' AND EXISTS (\n SELECT 1 FROM avail_execute_table aet\n WHERE aet.message_id::text = initiated_transactions.message_id\n))\n", - "describe": { - "columns": [], - "parameters": { - "Left": [] - }, - "nullable": [] - }, - "hash": "fa522e88dec4aa108f3da0de5d3af73a022e7fecafb4a2eb119b089531c21084" -} diff --git a/.sqlx/query-fcbac1d3922fdf9765c5ffca34d71b9712152c6473f7f74cb976f12850b25802.json b/.sqlx/query-fcbac1d3922fdf9765c5ffca34d71b9712152c6473f7f74cb976f12850b25802.json new file mode 100644 index 0000000..c9cf32a --- /dev/null +++ b/.sqlx/query-fcbac1d3922fdf9765c5ffca34d71b9712152c6473f7f74cb976f12850b25802.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT aet.message_id::text\nFROM avail_execute_table aet\nJOIN avail_indexer ai ON ai.id = aet.id\nWHERE ai.ext_hash = $1\nLIMIT 1\n", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "message_id", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "fcbac1d3922fdf9765c5ffca34d71b9712152c6473f7f74cb976f12850b25802" +} diff --git a/migrations/0001_create_initiated_transactions.sql b/migrations/0001_create_initiated_transactions.sql index 71f2c4b..1ecfced 100644 --- a/migrations/0001_create_initiated_transactions.sql +++ b/migrations/0001_create_initiated_transactions.sql @@ -8,8 +8,8 @@ CREATE TABLE IF NOT EXISTS initiated_transactions ( source_block_hash TEXT NOT NULL, source_block_number INTEGER NOT NULL, source_tx_index INTEGER, - timestamp BIGINT NOT NULL + timestamp BIGINT NOT NULL, + tx_type TEXT NOT NULL DEFAULT 'initiate' ); CREATE INDEX IF NOT EXISTS idx_initiated_tx_sender ON initiated_transactions (sender); -ALTER TABLE initiated_transactions ADD COLUMN IF NOT EXISTS tx_type TEXT NOT NULL DEFAULT 'initiate'; diff --git a/sql/check_initiated_tx_by_hash.sql b/sql/check_initiated_tx_by_hash.sql index d8e2e9c..17d8bbf 100644 --- a/sql/check_initiated_tx_by_hash.sql +++ b/sql/check_initiated_tx_by_hash.sql @@ -1 +1 @@ -SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_transaction_hash = $1) \ No newline at end of file +SELECT EXISTS(SELECT 1 FROM initiated_transactions WHERE source_transaction_hash = $1) diff --git a/sql/query_avail_claim_message_id.sql b/sql/query_avail_claim_message_id.sql new file mode 100644 index 0000000..b5b314c --- /dev/null +++ b/sql/query_avail_claim_message_id.sql @@ -0,0 +1,5 @@ +SELECT aet.message_id::text +FROM avail_execute_table aet +JOIN avail_indexer ai ON ai.id = aet.id +WHERE ai.ext_hash = $1 +LIMIT 1 diff --git a/sql/query_avail_tx.sql b/sql/query_avail_tx.sql index 1e5da09..9eea894 100644 --- a/sql/query_avail_tx.sql +++ b/sql/query_avail_tx.sql @@ -6,7 +6,7 @@ SELECT ai.id AS message_id, ai.ext_hash AS source_transaction_hash, ai.block_height AS source_block_height, ai.ext_index AS source_tx_index, - ai.block_timestamp, + EXTRACT(EPOCH FROM ai.block_timestamp)::bigint AS "block_timestamp!", be.source_transaction_hash AS "destination_tx_hash?: String", COALESCE( CASE diff --git a/sql/query_claimed_tx.sql b/sql/query_claimed_tx.sql index b5636ac..baaaf3a 100644 --- a/sql/query_claimed_tx.sql +++ b/sql/query_claimed_tx.sql @@ -2,3 +2,4 @@ SELECT message_id, source_transaction_hash, source_block_number, source_tx_index FROM initiated_transactions WHERE tx_type = 'claim' AND message_id = ANY($1) + AND (sender = $2 OR sender = $3) diff --git a/src/initiate_transaction.rs b/src/initiate_transaction.rs index df31349..e7fe986 100644 --- a/src/initiate_transaction.rs +++ b/src/initiate_transaction.rs @@ -742,18 +742,11 @@ async fn fetch_avail_claim_message_id_by_ext_hash( state: &Arc, ext_hash: &str, ) -> anyhow::Result { - let message_id_text: Option = sqlx::query_scalar( - r#" -SELECT aet.message_id::text -FROM avail_execute_table aet -JOIN avail_indexer ai ON ai.id = aet.id -WHERE ai.ext_hash = $1 -LIMIT 1 - "#, - ) - .bind(ext_hash) - .fetch_optional(&state.db) - .await?; + let message_id_text: Option = + sqlx::query_file_scalar!("sql/query_avail_claim_message_id.sql", ext_hash) + .fetch_optional(&state.db) + .await? + .flatten(); let message_id_text = message_id_text.ok_or_else(|| { ErrorResponse::with_status( diff --git a/src/main.rs b/src/main.rs index d7085cd..cf77958 100755 --- a/src/main.rs +++ b/src/main.rs @@ -284,7 +284,10 @@ async fn transactions( ) .fetch_all(&state.db) .await - .unwrap_or_default(); + .unwrap_or_else(|e| { + tracing::warn!("Failed to fetch initiated transactions: {e:#}"); + vec![] + }); for row in &initiated { let direction = if row.direction == "EthAvail" { @@ -323,15 +326,21 @@ async fn transactions( ClaimedTransactionRow, "sql/query_claimed_tx.sql", &message_ids, + eth_addr_for_initiated.as_deref().unwrap_or(""), + avail_addr_for_initiated.as_deref().unwrap_or("") ) .fetch_all(&state.db) .await - .unwrap_or_default(); + .unwrap_or_else(|e| { + tracing::warn!("Failed to fetch claimed transactions: {e:#}"); + vec![] + }); for claim in &claims { + let claim_id: BigDecimal = claim.message_id.parse().unwrap_or_default(); if let Some(tx) = transaction_data_results .iter_mut() - .find(|t| t.message_id.to_string() == claim.message_id) + .find(|t| t.message_id == claim_id && t.status == BridgeStatusEnum::Initiated) { tx.status = BridgeStatusEnum::Bridged; tx.destination_tx_hash = Some(claim.source_transaction_hash.clone()); @@ -965,11 +974,9 @@ fn cleanup_interval_seconds(raw: Option) -> u64 { } async fn cleanup_indexed_initiated_transactions(state: &Arc) -> Result<()> { - sqlx::query(include_str!( - "../sql/delete_indexed_initiated_tx_global.sql" - )) - .execute(&state.db) - .await?; + sqlx::query_file!("sql/delete_indexed_initiated_tx_global.sql") + .execute(&state.db) + .await?; Ok(()) }