Merge branch 'main' into registry-query

.github/workflows/tests.yaml

@@ -73,3 +73,18 @@ jobs:
 
       - name: Ensure the project builds
         run: cargo build
+  test:
+    needs: install
+    runs-on: ubuntu-latest
+    steps:
+      - name: Use cashed cargo
+        uses: actions/cache@v3
+        with:
+          path: ~/.cargo
+          key: ${{ runner.os }}-rust-${{ hashFiles('rust-toolchain.toml') }}
+
+      - name: Checkout the source code
+        uses: actions/checkout@v3
+
+      - name: Ensure the project builds
+        run: cargo test -- --test-threads=1

.gitignore

@@ -1,2 +1,3 @@
 /target
 /out
+logs/

Cargo.toml

@@ -6,6 +6,7 @@ license = "GPL-3.0-only"
 
 [workspace]
 members = [
+    "bin/processor",
     "bin/server",
     "bin/tracker",
     "routes",

README.md

@@ -53,12 +53,7 @@ A basic example of registering a parachain:
 
 ```
 curl -X POST http://127.0.0.1:8000/register_para -H "Content-Type: application/json" -d '{
-	"para": {
-		"name": "Acala",
-		"rpc_url": "wss://acala-rpc.dwellir.com",
-		"para_id": 2000,
-		"relay_chain": "Polkadot"
-	}
+    "para": ["Polkadot", 2000]                                                 
 }'
 ```
 

bin/processor/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "processor"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+log = "0.4"
+shared = { path = "../../shared" }
+env_logger = "0.10.1"
+polkadot-core-primitives = { git = "https://github.com/paritytech/polkadot-sdk", branch = "release-polkadot-v1.1.0" }
+types = { path = "../../types" }

bin/processor/src/main.rs

@@ -0,0 +1,84 @@
+// This file is part of RegionX.
+//
+// RegionX is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// RegionX is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with RegionX.  If not, see <https://www.gnu.org/licenses/>.
+
+use shared::{
+	config::config,
+	consumption::{delete_consumption, get_consumption, write_batch_consumption},
+	registry::registered_paras,
+};
+use std::collections::BTreeMap;
+use types::WeightConsumption;
+
+const LOG_TARGET: &str = "processor";
+
+fn main() {
+	env_logger::init();
+
+	let outputs = config().outputs;
+	let paras = registered_paras();
+
+	paras.iter().for_each(|para| {
+		let mut processed = BTreeMap::new();
+
+		log::info!(
+			target: LOG_TARGET,
+			"{}-{} - Processing consumption.",
+			para.relay_chain,
+			para.para_id,
+		);
+
+		(0..outputs).for_each(|output_index| {
+			let consumption = if let Ok(data) = get_consumption(para.clone(), Some(output_index)) {
+				data
+			} else {
+				log::error!(
+					target: LOG_TARGET,
+					"{}-{} - Failed to get consumption.",
+					para.relay_chain,
+					para.para_id,
+				);
+				vec![]
+			};
+
+			consumption.into_iter().for_each(|data| {
+				processed.entry(data.block_number).or_insert(data);
+			});
+		});
+
+		let processed: Vec<WeightConsumption> = processed.values().cloned().collect();
+
+		log::info!(
+			target: LOG_TARGET,
+			"{}-{} - Writing processed consumption. Total blocks tracked: {}",
+			para.relay_chain,
+			para.para_id,
+			processed.len()
+		);
+
+		if let Err(e) = write_batch_consumption(para.clone(), processed) {
+			log::error!(
+				target: LOG_TARGET,
+				"{}-{} - Failed to write batch consumption: {:?}",
+				para.relay_chain,
+				para.para_id,
+				e,
+			);
+
+			return;
+		}
+
+		(0..outputs).for_each(|output_index| delete_consumption(para.clone(), output_index));
+	});
+}

bin/server/src/main.rs

@@ -19,7 +19,10 @@
 /// - `/consumption`: Used to query consumption data associated with a parachain.
 /// - `/register`: Used to register a parachain for consumption tracking.
 use rocket_cors::CorsOptions;
-use routes::{consumption::consumption, register::register_para, registry::registry};
+use routes::{
+	consumption::consumption, extend_subscription::extend_subscription, register::register_para,
+	registry::registry,
+};
 
 #[macro_use]
 extern crate rocket;
@@ -28,5 +31,7 @@ extern crate rocket;
 fn rocket() -> _ {
 	rocket::build()
 		.attach(CorsOptions::default().to_cors().unwrap())
-		.mount("/", routes![consumption, register_para, registry])
+
+		.mount("/", routes![consumption, register_para, registry, extend_subscription])
+
 }

bin/tracker/Cargo.toml

@@ -14,3 +14,4 @@ tokio = { version = "1", features = ["full"] }
 
 types = { path = "../../types" }
 shared = { path = "../../shared" }
+clap = { version = "4.4.18", features = ["derive"] }