feat: implement follower file download flow

Cargo.toml

@@ -46,6 +46,7 @@ anyhow = "1.0.81"
 array-bytes = "6.1"
 async-channel = "1.8.0"
 async-io = "2.3.2"
+async-stream = "0.3"
 async-trait = "0.1.42"
 axum = "0.8"
 axum-extra = { version = "0.10", features = [
@@ -59,6 +60,7 @@ http = "1.1"
 base64 = "0.21"
 bigdecimal = { version = "0.4.5", features = ["serde"] }
 bincode = "1.3.3"
+bytes = "1"
 cfg-if = { version = "1.0.4" }
 clap = { version = "4.5.3", features = ["derive", "env"] }
 chrono = "0.4"
@@ -95,7 +97,8 @@ prost = "0.12"
 prost-build = "0.12.3"
 rand = "0.8.5"
 reference-trie = "0.29.1"
-rustls = { version = "0.23", default-features = false, features = ["ring"] }
+reqwest = { version = "0.12", features = ["stream"] }
+rustls = { version = "0.23", default-features = false, features = ["ring", "tls12"] }
 rustls-platform-verifier = "0.5"
 rustls-pemfile = "2.2"
 scale-info = { version = "2.11.0", default-features = false, features = [

client/Cargo.toml

@@ -18,9 +18,11 @@ workspace = true
 anyhow = { workspace = true }
 array-bytes = { workspace = true }
 async-channel = { workspace = true }
+async-stream = { workspace = true }
 async-trait = { workspace = true }
 axum = { workspace = true }
 axum-extra = { workspace = true }
+bytes = { workspace = true }
 chrono = { workspace = true }
 hex = { workspace = true, default-features = true }
 kvdb = { workspace = true }
@@ -30,6 +32,7 @@ lru = { workspace = true }
 ordered-float = { workspace = true }
 priority-queue = { workspace = true }
 rand = { workspace = true }
+reqwest = { workspace = true }
 sysinfo = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

client/blockchain-service-db/Cargo.toml

@@ -25,10 +25,10 @@ log = { workspace = true }
 rustls = { workspace = true }
 rustls-platform-verifier = { workspace = true }
 serde = { workspace = true, default-features = true }
-serde_json = { workspace = true }
+serde_json = { workspace = true, features = ["std"] }
 thiserror = { workspace = true }
 tokio = { workspace = true }
-tokio-postgres = { workspace = true }
+tokio-postgres = { workspace = true, features = ["with-serde_json-1"] }
 tokio-postgres-rustls = { workspace = true }
 rustls-pemfile = { workspace = true }
 sc-transaction-pool-api = { workspace = true }

client/blockchain-service-db/migrations/00000000000002_create_leader_info/down.sql

@@ -0,0 +1,4 @@
+-- Drop the leader_info table and related objects
+DROP TRIGGER IF EXISTS trg_leader_info_updated_at ON leader_info;
+DROP FUNCTION IF EXISTS set_leader_info_updated_at();
+DROP TABLE IF EXISTS leader_info;

client/blockchain-service-db/migrations/00000000000002_create_leader_info/up.sql

@@ -0,0 +1,29 @@
+-- Create a singleton table to store leader information
+-- The leader (node holding the advisory lock) writes its metadata here
+-- Metadata is JSON format containing host/port information
+CREATE TABLE IF NOT EXISTS leader_info (
+  id INTEGER PRIMARY KEY CHECK (id = 1), -- Enforce singleton (only one row allowed)
+  metadata JSONB NOT NULL,               -- Leader metadata 
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Pre-insert the singleton row to simplify upserts
+INSERT INTO leader_info (id, metadata, updated_at)
+VALUES (1, '{}'::jsonb, now())
+ON CONFLICT (id) DO NOTHING;
+
+-- Update timestamp trigger for leader_info
+CREATE OR REPLACE FUNCTION set_leader_info_updated_at() RETURNS trigger AS $$
+BEGIN
+  NEW.updated_at = now();
+  RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DO $$ BEGIN
+  CREATE TRIGGER trg_leader_info_updated_at
+  BEFORE UPDATE ON leader_info
+  FOR EACH ROW EXECUTE FUNCTION set_leader_info_updated_at();
+EXCEPTION WHEN duplicate_object THEN
+  NULL;
+END $$;