Unsettled transfer sessions

.github/workflows/CI.yml

@@ -92,7 +92,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        package: [boost-manager,file-store,iot-config,iot-packet-verifier,iot-verifier,mobile-config,mobile-verifier]
+        package: [boost-manager,file-store,iot-config,iot-packet-verifier,iot-verifier,mobile-config,mobile-packet-verifier,mobile-verifier]
     concurrency: 
       group: ${{ github.workflow }}-${{ github.ref }}-tests-postgres-${{ matrix.package }}
       cancel-in-progress: true
@@ -142,7 +142,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        package: [coverage-map,coverage-point-calculator,ingest,mobile-packet-verifier,reward-scheduler,task-manager]
+        package: [coverage-map,coverage-point-calculator,ingest,reward-scheduler,task-manager]
     concurrency: 
       group: ${{ github.workflow }}-${{ github.ref }}-tests-${{ matrix.package }}
       cancel-in-progress: true

Cargo.toml

@@ -70,10 +70,10 @@ helium-lib = { git = "https://github.com/helium/helium-wallet-rs.git", branch =
 hextree = { git = "https://github.com/jaykickliter/HexTree", branch = "main", features = [
   "disktree",
 ] }
-helium-proto = { git = "https://github.com/helium/proto", branch = "master", features = [
+helium-proto = { git = "https://github.com/helium/proto", branch = "mj/unsettled-transfer-session", features = [
   "services",
 ] }
-beacon = { git = "https://github.com/helium/proto", branch = "master" }
+beacon = { git = "https://github.com/helium/proto", branch = "mj/unsettled-transfer-session" }
 solana-client = "1.18"
 solana-sdk = "1.18"
 solana-program = "1.18"

file_store/src/file_info.rs

@@ -146,6 +146,7 @@ pub const DATA_TRANSFER_SESSION_INGEST_REPORT: &str = "data_transfer_session_ing
 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 PENDING_DATA_TRANSFER_SESION: &str = "pending_data_transfer_sesion";
 pub const PRICE_REPORT: &str = "price_report";
 pub const MOBILE_REWARD_SHARE: &str = "mobile_reward_share";
 pub const MAPPER_MSG: &str = "mapper_msg";
@@ -193,6 +194,7 @@ pub enum FileType {
     DataTransferSessionIngestReport,
     InvalidDataTransferSessionIngestReport,
     ValidDataTransferSession,
+    PendingDataTransferSession,
     PriceReport,
     MobileRewardShare,
     SubscriberLocationReq,
@@ -270,6 +272,7 @@ impl fmt::Display for FileType {
                 INVALID_DATA_TRANSFER_SESSION_INGEST_REPORT
             }
             Self::ValidDataTransferSession => VALID_DATA_TRANSFER_SESSION,
+            Self::PendingDataTransferSession => PENDING_DATA_TRANSFER_SESION,
             Self::PriceReport => PRICE_REPORT,
             Self::MobileRewardShare => MOBILE_REWARD_SHARE,
             Self::MapperMsg => MAPPER_MSG,
@@ -344,6 +347,7 @@ impl FileType {
                 INVALID_DATA_TRANSFER_SESSION_INGEST_REPORT
             }
             Self::ValidDataTransferSession => VALID_DATA_TRANSFER_SESSION,
+            Self::PendingDataTransferSession => PENDING_DATA_TRANSFER_SESION,
             Self::PriceReport => PRICE_REPORT,
             Self::MobileRewardShare => MOBILE_REWARD_SHARE,
             Self::MapperMsg => MAPPER_MSG,
@@ -418,6 +422,7 @@ impl FromStr for FileType {
                 Self::InvalidDataTransferSessionIngestReport
             }
             VALID_DATA_TRANSFER_SESSION => Self::ValidDataTransferSession,
+            PENDING_DATA_TRANSFER_SESION => Self::PendingDataTransferSession,
             PRICE_REPORT => Self::PriceReport,
             MOBILE_REWARD_SHARE => Self::MobileRewardShare,
             MAPPER_MSG => Self::MapperMsg,

file_store/src/mobile_session.rs

@@ -8,7 +8,9 @@ use helium_crypto::PublicKeyBinary;
 use helium_proto::services::poc_mobile::{
     invalid_data_transfer_ingest_report_v1::DataTransferIngestReportStatus,
     DataTransferEvent as DataTransferEventProto, DataTransferRadioAccessTechnology,
-    DataTransferSessionIngestReportV1, DataTransferSessionReqV1, InvalidDataTransferIngestReportV1,
+    DataTransferSessionIngestReportV1, DataTransferSessionReqV1,
+    DataTransferSessionSettlementStatus, InvalidDataTransferIngestReportV1,
+    PendingDataTransferSessionV1,
 };
 
 use serde::Serialize;
@@ -181,6 +183,7 @@ pub struct DataTransferSessionReq {
     pub rewardable_bytes: u64,
     pub pub_key: PublicKeyBinary,
     pub signature: Vec<u8>,
+    pub status: DataTransferSessionSettlementStatus,
 }
 
 impl MsgDecode for DataTransferSessionReq {
@@ -191,6 +194,7 @@ impl TryFrom<DataTransferSessionReqV1> for DataTransferSessionReq {
     type Error = Error;
 
     fn try_from(v: DataTransferSessionReqV1) -> Result<Self> {
+        let status = v.status();
         Ok(Self {
             rewardable_bytes: v.rewardable_bytes,
             signature: v.signature,
@@ -199,6 +203,7 @@ impl TryFrom<DataTransferSessionReqV1> for DataTransferSessionReq {
                 .ok_or_else(|| Error::not_found("data transfer usage"))?
                 .try_into()?,
             pub_key: v.pub_key.into(),
+            status,
         })
     }
 }
@@ -212,7 +217,28 @@ impl From<DataTransferSessionReq> for DataTransferSessionReqV1 {
             rewardable_bytes: v.rewardable_bytes,
             pub_key: v.pub_key.into(),
             signature: v.signature,
-            ..Default::default()
+            reward_cancelled: false,
+            status: v.status as i32,
+        }
+    }
+}
+
+impl DataTransferSessionReq {
+    pub fn to_pending_proto(
+        self,
+        received_timestamp: DateTime<Utc>,
+    ) -> PendingDataTransferSessionV1 {
+        let event_timestamp = self.data_transfer_usage.timestamp.encode_timestamp_millis();
+        let received_timestamp = received_timestamp.encode_timestamp_millis();
+
+        PendingDataTransferSessionV1 {
+            pub_key: self.pub_key.into(),
+            payer: self.data_transfer_usage.payer.into(),
+            upload_bytes: self.data_transfer_usage.upload_bytes,
+            download_bytes: self.data_transfer_usage.download_bytes,
+            rewardable_bytes: self.rewardable_bytes,
+            event_timestamp,
+            received_timestamp,
         }
     }
 }

file_store/src/traits/file_sink_write.rs

@@ -157,6 +157,11 @@ impl_file_sink!(
     FileType::OracleBoostingReport.to_str(),
     "oracle_boosting_report"
 );
+impl_file_sink!(
+    poc_mobile::PendingDataTransferSessionV1,
+    FileType::PendingDataTransferSession.to_str(),
+    "pending_data_transfer_session"
+);
 impl_file_sink!(
     poc_mobile::RadioThresholdIngestReportV1,
     FileType::RadioThresholdIngestReport.to_str(),

mobile_packet_verifier/Cargo.toml

@@ -41,3 +41,6 @@ http-serde = { workspace = true }
 sha2 = { workspace = true }
 humantime-serde = { workspace = true }
 custom-tracing = { path = "../custom_tracing" }
+
+[dev-dependencies]
+rand = { workspace = true }

mobile_packet_verifier/migrations/7_pending_data_transfer_sessions.sql

@@ -0,0 +1,11 @@
+CREATE TABLE pending_data_transfer_sessions (
+       pub_key              TEXT NOT NULL,
+       payer                TEXT NOT NULL,
+       event_id             TEXT NOT NULL,
+       uploaded_bytes       BIGINT NOT NULL,
+       downloaded_bytes     BIGINT NOT NULL,
+       rewardable_bytes     BIGINT NOT NULL,
+       recv_timestamp       TIMESTAMPTZ NOT NULL,
+       inserted_at          TIMESTAMPTZ NOT NULL,
+       PRIMARY KEY(pub_key, payer, recv_timestamp)
+);