output paper trail of invalid data transfer sessions

Cargo.toml

@@ -57,14 +57,14 @@ 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 = "andymck/add-invalid-mobile-packet-support", features = ["services"]}
 hextree = "*"
 solana-client = "1.14"
 solana-sdk = "1.14"
 solana-program = "1.11"
 spl-token = "3.5.0"
 reqwest = {version = "0", default-features=false, features = ["gzip", "json", "rustls-tls"]}
-beacon = { git = "https://github.com/helium/proto", branch = "master" }
+beacon = { git = "https://github.com/helium/proto", branch = "andymck/add-invalid-mobile-packet-support" }
 humantime = "2"
 metrics = "0"
 metrics-exporter-prometheus = "0"

file_store/src/cli/dump.rs

@@ -3,7 +3,7 @@ use crate::{
     file_source,
     heartbeat::{CellHeartbeat, CellHeartbeatIngestReport},
     iot_packet::IotValidPacket,
-    mobile_session::DataTransferSessionIngestReport,
+    mobile_session::{DataTransferSessionIngestReport, InvalidDataTransferIngestReport},
     speedtest::{CellSpeedtest, CellSpeedtestIngestReport},
     traits::MsgDecode,
     FileType, Result, Settings,
@@ -17,8 +17,8 @@ use helium_proto::{
         poc_lora::{LoraBeaconIngestReportV1, LoraPocV1, LoraWitnessIngestReportV1},
         poc_mobile::{
             mobile_reward_share::Reward, CellHeartbeatIngestReportV1, CellHeartbeatReqV1,
-            Heartbeat, MobileRewardShare, RadioRewardShare, SpeedtestAvg, SpeedtestIngestReportV1,
-            SpeedtestReqV1,
+            Heartbeat, InvalidDataTransferIngestReportV1, MobileRewardShare, RadioRewardShare,
+            SpeedtestAvg, SpeedtestIngestReportV1, SpeedtestReqV1,
         },
         router::PacketRouterPacketReportV1,
     },
@@ -77,6 +77,23 @@ impl Cmd {
                         "timestamp": dtr.report.data_transfer_usage.timestamp,
                     }))?;
                 }
+                FileType::InvalidDataTransferSessionIngestReport => {
+                    let msg: InvalidDataTransferIngestReport =
+                        InvalidDataTransferIngestReportV1::decode(msg)?.try_into()?;
+                    print_json(&json!({
+                        "invalid_reason": msg.reason,
+                        "received_timestamp": msg.report.received_timestamp,
+                        "reward_cancelled": msg.report.report.reward_cancelled,
+                        "hotspot_key": msg.report.report.data_transfer_usage.pub_key,
+                        "routing_key": msg.report.report.pub_key,
+                        "upload_bytes": msg.report.report.data_transfer_usage.upload_bytes,
+                        "download_bytes": msg.report.report.data_transfer_usage.download_bytes,
+                        "radio_access_technology": msg.report.report.data_transfer_usage.radio_access_technology,
+                        "event_id": msg.report.report.data_transfer_usage.event_id,
+                        "payer": msg.report.report.data_transfer_usage.payer,
+                        "timestamp": msg.report.report.data_transfer_usage.timestamp,
+                    }))?;
+                }
                 FileType::IotBeaconIngestReport => {
                     let dec_msg = LoraBeaconIngestReportV1::decode(msg)?;
                     let json = json!({

file_store/src/file_info.rs

@@ -120,6 +120,8 @@ pub const INVALID_PACKET: &str = "invalid_packet";
 pub const NON_REWARDABLE_PACKET: &str = "non_rewardable_packet";
 pub const IOT_REWARD_SHARE: &str = "iot_reward_share";
 pub const DATA_TRANSFER_SESSION_INGEST_REPORT: &str = "data_transfer_session_ingest_report";
+pub const INVALID_DATA_TRANSFER_SESSION_INGEST_REPORT: &str =
+    "invalid_data_transfer_session_ingest_report";
 pub const VALID_DATA_TRANSFER_SESSION: &str = "valid_data_transfer_session";
 pub const PRICE_REPORT: &str = "price_report";
 pub const MOBILE_REWARD_SHARE: &str = "mobile_reward_share";
@@ -150,6 +152,7 @@ pub enum FileType {
     NonRewardablePacket,
     IotRewardShare,
     DataTransferSessionIngestReport,
+    InvalidDataTransferSessionIngestReport,
     ValidDataTransferSession,
     PriceReport,
     MobileRewardShare,
@@ -185,6 +188,9 @@ impl fmt::Display for FileType {
             Self::NonRewardablePacket => NON_REWARDABLE_PACKET,
             Self::IotRewardShare => IOT_REWARD_SHARE,
             Self::DataTransferSessionIngestReport => DATA_TRANSFER_SESSION_INGEST_REPORT,
+            Self::InvalidDataTransferSessionIngestReport => {
+                INVALID_DATA_TRANSFER_SESSION_INGEST_REPORT
+            }
             Self::ValidDataTransferSession => VALID_DATA_TRANSFER_SESSION,
             Self::PriceReport => PRICE_REPORT,
             Self::MobileRewardShare => MOBILE_REWARD_SHARE,
@@ -221,6 +227,9 @@ impl FileType {
             Self::NonRewardablePacket => NON_REWARDABLE_PACKET,
             Self::IotRewardShare => IOT_REWARD_SHARE,
             Self::DataTransferSessionIngestReport => DATA_TRANSFER_SESSION_INGEST_REPORT,
+            Self::InvalidDataTransferSessionIngestReport => {
+                INVALID_DATA_TRANSFER_SESSION_INGEST_REPORT
+            }
             Self::ValidDataTransferSession => VALID_DATA_TRANSFER_SESSION,
             Self::PriceReport => PRICE_REPORT,
             Self::MobileRewardShare => MOBILE_REWARD_SHARE,
@@ -255,6 +264,9 @@ impl FromStr for FileType {
             NON_REWARDABLE_PACKET => Self::NonRewardablePacket,
             IOT_REWARD_SHARE => Self::IotRewardShare,
             DATA_TRANSFER_SESSION_INGEST_REPORT => Self::DataTransferSessionIngestReport,
+            INVALID_DATA_TRANSFER_SESSION_INGEST_REPORT => {
+                Self::InvalidDataTransferSessionIngestReport
+            }
             VALID_DATA_TRANSFER_SESSION => Self::ValidDataTransferSession,
             PRICE_REPORT => Self::PriceReport,
             MOBILE_REWARD_SHARE => Self::MobileRewardShare,

file_store/src/mobile_session.rs

@@ -1,21 +1,17 @@
 use crate::{
-    traits::{MsgDecode, MsgTimestamp, TimestampDecode},
+    error::DecodeError,
+    traits::{MsgDecode, MsgTimestamp, TimestampDecode, TimestampEncode},
     Error, Result,
 };
 use chrono::{DateTime, Utc};
 use helium_crypto::PublicKeyBinary;
 use helium_proto::services::poc_mobile::{
-    DataTransferEvent as DataTransferEventProto, DataTransferRadioAccessTechnology,
-    DataTransferSessionIngestReportV1, DataTransferSessionReqV1,
+    DataTransferEvent as DataTransferEventProto, DataTransferIngestReportStatus,
+    DataTransferRadioAccessTechnology, DataTransferSessionIngestReportV1, DataTransferSessionReqV1,
+    InvalidDataTransferIngestReportV1,
 };
 use serde::Serialize;
 
-impl MsgTimestamp<Result<DateTime<Utc>>> for DataTransferSessionIngestReportV1 {
-    fn timestamp(&self) -> Result<DateTime<Utc>> {
-        self.received_timestamp.to_timestamp_millis()
-    }
-}
-
 #[derive(Serialize, Clone, Debug)]
 pub struct DataTransferSessionIngestReport {
     pub received_timestamp: DateTime<Utc>,
@@ -26,6 +22,18 @@ impl MsgDecode for DataTransferSessionIngestReport {
     type Msg = DataTransferSessionIngestReportV1;
 }
 
+impl MsgTimestamp<Result<DateTime<Utc>>> for DataTransferSessionIngestReportV1 {
+    fn timestamp(&self) -> Result<DateTime<Utc>> {
+        self.received_timestamp.to_timestamp_millis()
+    }
+}
+
+impl MsgTimestamp<u64> for DataTransferSessionIngestReport {
+    fn timestamp(&self) -> u64 {
+        self.received_timestamp.encode_timestamp_millis()
+    }
+}
+
 impl TryFrom<DataTransferSessionIngestReportV1> for DataTransferSessionIngestReport {
     type Error = Error;
 
@@ -40,6 +48,69 @@ impl TryFrom<DataTransferSessionIngestReportV1> for DataTransferSessionIngestRep
     }
 }
 
+impl From<DataTransferSessionIngestReport> for DataTransferSessionIngestReportV1 {
+    fn from(v: DataTransferSessionIngestReport) -> Self {
+        let received_timestamp = v.timestamp();
+        let report: DataTransferSessionReqV1 = v.report.into();
+        Self {
+            report: Some(report),
+            received_timestamp,
+        }
+    }
+}
+
+#[derive(Serialize, Clone, Debug)]
+pub struct InvalidDataTransferIngestReport {
+    pub report: DataTransferSessionIngestReport,
+    pub reason: DataTransferIngestReportStatus,
+    pub timestamp: DateTime<Utc>,
+}
+
+impl MsgDecode for InvalidDataTransferIngestReport {
+    type Msg = InvalidDataTransferIngestReportV1;
+}
+
+impl MsgTimestamp<Result<DateTime<Utc>>> for InvalidDataTransferIngestReportV1 {
+    fn timestamp(&self) -> Result<DateTime<Utc>> {
+        self.timestamp.to_timestamp_millis()
+    }
+}
+
+impl MsgTimestamp<u64> for InvalidDataTransferIngestReport {
+    fn timestamp(&self) -> u64 {
+        self.timestamp.encode_timestamp_millis()
+    }
+}
+
+impl TryFrom<InvalidDataTransferIngestReportV1> for InvalidDataTransferIngestReport {
+    type Error = Error;
+    fn try_from(v: InvalidDataTransferIngestReportV1) -> Result<Self> {
+        let reason = DataTransferIngestReportStatus::from_i32(v.reason).ok_or_else(|| {
+            DecodeError::unsupported_status_reason("invalid_data_transfer_session_reason", v.reason)
+        })?;
+        Ok(Self {
+            report: v
+                .report
+                .ok_or_else(|| Error::not_found("data transfer session ingest report"))?
+                .try_into()?,
+            timestamp: v.timestamp.to_timestamp()?,
+            reason,
+        })
+    }
+}
+
+impl From<InvalidDataTransferIngestReport> for InvalidDataTransferIngestReportV1 {
+    fn from(v: InvalidDataTransferIngestReport) -> Self {
+        let timestamp = v.timestamp();
+        let report: DataTransferSessionIngestReportV1 = v.report.into();
+        Self {
+            report: Some(report),
+            reason: v.reason as i32,
+            timestamp,
+        }
+    }
+}
+
 #[derive(Serialize, Clone, Debug)]
 pub struct DataTransferEvent {
     pub pub_key: PublicKeyBinary,
@@ -58,6 +129,12 @@ impl MsgTimestamp<Result<DateTime<Utc>>> for DataTransferEventProto {
     }
 }
 
+impl MsgTimestamp<u64> for DataTransferEvent {
+    fn timestamp(&self) -> u64 {
+        self.timestamp.encode_timestamp_millis()
+    }
+}
+
 impl MsgDecode for DataTransferEvent {
     type Msg = DataTransferEventProto;
 }
@@ -81,6 +158,22 @@ impl TryFrom<DataTransferEventProto> for DataTransferEvent {
     }
 }
 
+impl From<DataTransferEvent> for DataTransferEventProto {
+    fn from(v: DataTransferEvent) -> Self {
+        let timestamp = v.timestamp();
+        Self {
+            pub_key: v.pub_key.into(),
+            upload_bytes: v.upload_bytes,
+            download_bytes: v.download_bytes,
+            radio_access_technology: v.radio_access_technology as i32,
+            event_id: v.event_id,
+            payer: v.payer.into(),
+            timestamp,
+            signature: v.signature,
+        }
+    }
+}
+
 #[derive(Serialize, Clone, Debug)]
 pub struct DataTransferSessionReq {
     pub data_transfer_usage: DataTransferEvent,
@@ -108,3 +201,15 @@ impl TryFrom<DataTransferSessionReqV1> for DataTransferSessionReq {
         })
     }
 }
+
+impl From<DataTransferSessionReq> for DataTransferSessionReqV1 {
+    fn from(v: DataTransferSessionReq) -> Self {
+        let report: DataTransferEventProto = v.data_transfer_usage.into();
+        Self {
+            data_transfer_usage: Some(report),
+            reward_cancelled: v.reward_cancelled,
+            pub_key: v.pub_key.into(),
+            signature: v.signature,
+        }
+    }
+}