Add support for token-price conversion

- Temp update Cargo.toml and Cargo.lock
- Update file info and dump cli
- Add docker-compose.yml for local testing

Cargo.toml

@@ -16,7 +16,8 @@ members = [
     "metrics",
     "denylist",
     "iot_packet_verifier",
-    "iot_config"
+    "iot_config",
+    "price",
 ]
 
 [workspace.package]
@@ -53,12 +54,12 @@ sqlx = {version = "0", features = [
 ]}
 
 helium-crypto = {version = "0.6.8", features=["sqlx-postgres", "multisig"]}
-helium-proto = {git = "https://github.com/helium/proto", branch = "master", features = ["services"]}
+helium-proto = {git = "https://github.com/helium/proto", branch = "rg/price-oracle-grpc", features = ["services"]}
 hextree = "*"
 solana-client = "1.14"
 solana-sdk = "1.14"
 reqwest = {version = "0", default-features=false, features = ["gzip", "json", "rustls-tls"]}
-beacon = {git = "https://github.com/helium/gateway-rs.git", branch = "main"}
+beacon = {git = "https://github.com/helium/gateway-rs.git", branch = "rg/proto-test-branch"}
 humantime = "2"
 metrics = "0"
 metrics-exporter-prometheus = "0"

file_store/src/cli/dump.rs

@@ -15,9 +15,10 @@ use helium_proto::{
             CellHeartbeatIngestReportV1, CellHeartbeatReqV1, Heartbeat, RadioRewardShare,
             SpeedtestAvg, SpeedtestIngestReportV1, SpeedtestReqV1,
         },
+        price_oracle::PriceOracleReportV1,
         router::PacketRouterPacketReportV1,
     },
-    BlockchainTxn, Message, RewardManifest, SubnetworkRewards,
+    BlockchainTokenTypeV1, BlockchainTxn, Message, RewardManifest, SubnetworkRewards,
 };
 use serde_json::json;
 use std::io;
@@ -163,8 +164,14 @@ impl Cmd {
                     let packet_report = PacketRouterPacketReportV1::decode(msg)?;
                     print_json(&json!({
                         "oui": packet_report.oui,
-                        "timestamp": packet_report.gateway_timestamp_ms,
-
+                        "timestamp": packet_report.gateway_timestamp_ms}))?;
+                }
+                FileType::PriceReport => {
+                    let manifest = PriceOracleReportV1::decode(msg)?;
+                    print_json(&json!({
+                        "price": manifest.price,
+                        "timestamp": manifest.timestamp,
+                        "token_type": BlockchainTokenTypeV1::from_i32(manifest.token_type),
                     }))?;
                 }
                 _ => (),

file_store/src/cli/info.rs

@@ -7,8 +7,9 @@ use crate::{
 use bytes::BytesMut;
 use chrono::{DateTime, Utc};
 use futures::StreamExt;
-use helium_proto::services::poc_lora::{
-    LoraBeaconIngestReportV1, LoraPocV1, LoraWitnessIngestReportV1,
+use helium_proto::services::{
+    poc_lora::{LoraBeaconIngestReportV1, LoraPocV1, LoraWitnessIngestReportV1},
+    price_oracle::PriceOracleReportV1,
 };
 use helium_proto::{
     services::poc_mobile::{
@@ -72,6 +73,12 @@ impl MsgTimestamp<Result<DateTime<Utc>>> for EntropyReportV1 {
     }
 }
 
+impl MsgTimestamp<Result<DateTime<Utc>>> for PriceOracleReportV1 {
+    fn timestamp(&self) -> Result<DateTime<Utc>> {
+        self.timestamp.to_timestamp()
+    }
+}
+
 fn get_timestamp(file_type: &FileType, buf: &[u8]) -> Result<DateTime<Utc>> {
     let result = match file_type {
         FileType::CellHeartbeat => CellHeartbeatReqV1::decode(buf)
@@ -115,6 +122,9 @@ fn get_timestamp(file_type: &FileType, buf: &[u8]) -> Result<DateTime<Utc>> {
                 })
             })
             .and_then(|beacon_report| beacon_report.timestamp())?,
+        FileType::PriceReport => PriceOracleReportV1::decode(buf)
+            .map_err(Error::from)
+            .and_then(|entry| entry.timestamp())?,
 
         _ => Utc::now(),
     };

file_store/src/file_info.rs

@@ -111,6 +111,7 @@ pub const VALID_PACKET: &str = "valid_packet";
 pub const INVALID_PACKET: &str = "invalid_packet";
 pub const GATEWAY_REWARD_SHARE: &str = "gateway_reward_share";
 pub const DATA_TRANSFER_SESSION_INGEST_REPORT: &str = "data_transfer_session_ingest_report";
+pub const PRICE_REPORT: &str = "price_report";
 
 #[derive(Debug, PartialEq, Eq, Clone, Serialize, Copy, strum::EnumCount)]
 #[serde(rename_all = "snake_case")]
@@ -137,6 +138,7 @@ pub enum FileType {
     InvalidPacket,
     GatewayRewardShare,
     DataTransferSessionIngestReport,
+    PriceReport,
 }
 
 impl fmt::Display for FileType {
@@ -164,6 +166,7 @@ impl fmt::Display for FileType {
             Self::InvalidPacket => INVALID_PACKET,
             Self::GatewayRewardShare => GATEWAY_REWARD_SHARE,
             Self::DataTransferSessionIngestReport => DATA_TRANSFER_SESSION_INGEST_REPORT,
+            Self::PriceReport => PRICE_REPORT,
         };
         f.write_str(s)
     }
@@ -194,6 +197,7 @@ impl FileType {
             Self::InvalidPacket => INVALID_PACKET,
             Self::GatewayRewardShare => GATEWAY_REWARD_SHARE,
             Self::DataTransferSessionIngestReport => DATA_TRANSFER_SESSION_INGEST_REPORT,
+            Self::PriceReport => PRICE_REPORT,
         }
     }
 }
@@ -224,6 +228,7 @@ impl FromStr for FileType {
             INVALID_PACKET => Self::InvalidPacket,
             GATEWAY_REWARD_SHARE => Self::GatewayRewardShare,
             DATA_TRANSFER_SESSION_INGEST_REPORT => Self::DataTransferSessionIngestReport,
+            PRICE_REPORT => Self::PriceReport,
             _ => return Err(Error::from(io::Error::from(io::ErrorKind::InvalidInput))),
         };
         Ok(result)

file_store/src/file_sink.rs

@@ -141,7 +141,7 @@ impl FileSinkBuilder {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Debug, Clone)]
 pub struct FileSinkClient {
     sender: MessageSender,
     metric: &'static str,

price/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "price"
+version = "0.1.0"
+description = "Price Oracle for the Helium Network"
+edition.workspace = true
+authors.workspace = true
+license.workspace = true
+
+[dependencies]
+anyhow = {workspace = true}
+config = {workspace = true}
+clap = {workspace = true}
+thiserror = {workspace = true}
+serde =  {workspace = true}
+serde_json = {workspace = true}
+tonic = {workspace = true}
+futures = {workspace = true}
+futures-util = {workspace = true}
+prost = {workspace = true}
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+metrics = {workspace = true }
+metrics-exporter-prometheus = { workspace = true }
+tokio = { workspace = true }
+chrono = { workspace = true }
+helium-proto = { workspace = true }
+file-store = { path = "../file_store" }
+poc-metrics = { path = "../metrics" }
+triggered = {workspace = true}
+pyth-sdk-solana = "0.7.1"
+solana-client = ">= 1.9, < 1.15"
+solana-program = ">= 1.9, < 1.15"

price/README.md

@@ -0,0 +1,16 @@
+# Price Oracle Server
+
+The price oracle server serves up price data for helium token(s) acquired from
+the [pyth.network](https://pyth.network).
+
+The supported tokens are:
+- HNT
+- HST
+- MOBILE
+- IOT
+
+The price oracle server:
+
+- Requests price for HNT token at a regular interval (60s) from pyth.
+- Stores and uploads [price_report](TODO) to a bucket.
+- TODO: Acquire price for MOBILE and IOT token(s) when available on pyth.

price/docker-compose.yml

@@ -0,0 +1,67 @@
+version: "2.4"
+services:
+  minio:
+    image: minio/minio:latest
+    environment:
+      MINIO_ROOT_USER: oracleadmin
+      MINIO_ROOT_PASSWORD: oracleadmin
+    ports:
+      - "9000:9000"
+      - "9090:9090"
+    volumes:
+      - bucket-data:/data
+    command: server /data --console-address ":9090"
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 30s
+      timeout: 20s
+      retries: 3
+  minio-setup:
+    image: minio/mc:latest
+    depends_on:
+      - minio
+    environment:
+      MINIO_ROOT_USER: oracleadmin
+      MINIO_ROOT_PASSWORD: oracleadmin
+      MINIO_BUCKETS: >
+        price
+      ORACLE_ID: oraclesecretid
+      ORACLE_KEY: oraclesecretkey
+    entrypoint:
+      - /bin/bash
+      - -c
+      - |
+        cat > /bucket-policy.json <<EOF
+        {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": "ListObjectsInBucket",
+                    "Effect": "Allow",
+                    "Action": ["s3:ListBucket"],
+                    "Resource": ["arn:aws:s3:::*"]
+                },
+                {
+                    "Sid": "AllObjectActions",
+                    "Effect": "Allow",
+                    "Action": "s3:*Object",
+                    "Resource": ["arn:aws:s3:::*"]
+                }
+            ]
+        }
+        EOF
+        sleep 2
+        /usr/bin/mc alias set localminio http://minio:9000 $${MINIO_ROOT_USER} $${MINIO_ROOT_PASSWORD}
+        for bucket in $${MINIO_BUCKETS[@]}
+        do
+            if ! /usr/bin/mc ls localminio/$${bucket} > /dev/null 2>&1 ; then
+                echo "creating bucket $${bucket}"
+                /usr/bin/mc mb localminio/$${bucket}
+            fi
+        done
+        /usr/bin/mc admin policy add localminio fullaccess /bucket-policy.json
+        /usr/bin/mc admin user add localminio $${ORACLE_ID} $${ORACLE_KEY}
+        /usr/bin/mc admin policy set localminio fullaccess user=$${ORACLE_ID}
+
+volumes:
+  bucket-data:

price/pkg/settings-template.toml

@@ -0,0 +1,56 @@
+# log settings for the application (RUST_LOG format). Default below
+# 
+# log = "price=debug"
+
+# RPC Endpoint for price oracles. Required.
+rpc_endpoint = "https://api.devnet.solana.com"
+
+# Price sink roll time (mins). Default = 3 mins. Optional.
+sink_roll_mins = 3
+
+# Price tick interval (secs). Default = 60s. Optional.
+tick_interval = 60
+
+# Price age (get price as long as it was updated within `age` seconds of current time) (in secs). Optional.
+age = 60
+
+[cluster]
+name = "devnet"
+hnt_price_key = "6Eg8YdfFJQF2HHonzPUBSCCmyUEhrStg9VBLK957sBe6"
+# mobile_price_key =
+# hst_price_key =
+# iot_price_key =
+
+# Listen addresses for public api. Default below
+#
+# hnt_listen = "0.0.0.0:8080"
+# mobile_listen = "0.0.0.0:8081"
+# iot_listen = "0.0.0.0:8082"
+# hst_listen = "0.0.0.0:8082"
+
+# Cache folder to use. Default blow
+#
+# cache = "/var/data/price"
+
+[output]
+# Output bucket for price
+
+# Name of bucket to write details to. Required
+#
+bucket = "price"
+
+# Region for bucket. Defaults to below
+#
+# region = "us-west-2"
+
+# Optional URL for AWS api endpoint. Inferred from aws config settings or aws
+# IAM context by default
+#
+# endpoint = "https://aws-s3-bucket.aws.com"
+
+
+[metrics]
+
+# Endpoint for metrics. Default below
+#
+# endpoint = "127.0.0.1:19000"

price/src/error.rs

@@ -0,0 +1,9 @@
+#[derive(thiserror::Error, Debug)]
+pub enum PriceError {
+    #[error("unsupported token type: {0}")]
+    UnsupportedTokenType(i32),
+    #[error("unable to fetch price")]
+    UnableToFetch,
+    #[error("unknown price account key, token_type: {0}")]
+    UnknownKey(String),
+}