Service Provider Rewards and Promotions

file_store/src/file_source.rs

@@ -1,23 +1,51 @@
 use crate::{
-    file_info_poller::{FileInfoPollerConfigBuilder, MsgDecodeFileInfoPollerParser},
+    file_info_poller::{
+        FileInfoPollerConfigBuilder, MsgDecodeFileInfoPollerParser, ProstFileInfoPollerParser,
+    },
     file_sink, BytesMutStream, Error, FileStore,
 };
 use async_compression::tokio::bufread::GzipDecoder;
 use futures::{
     stream::{self},
     StreamExt, TryFutureExt, TryStreamExt,
 };
-use std::path::{Path, PathBuf};
+use std::{
+    marker::PhantomData,
+    path::{Path, PathBuf},
+};
 use tokio::{fs::File, io::BufReader};
 use tokio_util::codec::{length_delimited::LengthDelimitedCodec, FramedRead};
 
+pub struct Continuous<Store, Parser>(PhantomData<(Store, Parser)>);
+
+impl Continuous<FileStore, MsgDecodeFileInfoPollerParser> {
+    pub fn msg_source<Msg, State>(
+    ) -> FileInfoPollerConfigBuilder<Msg, State, FileStore, MsgDecodeFileInfoPollerParser>
+    where
+        Msg: Clone,
+    {
+        FileInfoPollerConfigBuilder::<Msg, State, FileStore, MsgDecodeFileInfoPollerParser>::default()
+            .parser(MsgDecodeFileInfoPollerParser)
+    }
+}
+
+impl Continuous<FileStore, ProstFileInfoPollerParser> {
+    pub fn prost_source<Msg, State>(
+    ) -> FileInfoPollerConfigBuilder<Msg, State, FileStore, ProstFileInfoPollerParser>
+    where
+        Msg: Clone,
+    {
+        FileInfoPollerConfigBuilder::<Msg, State, FileStore, ProstFileInfoPollerParser>::default()
+            .parser(ProstFileInfoPollerParser)
+    }
+}
+
 pub fn continuous_source<T, S>(
 ) -> FileInfoPollerConfigBuilder<T, S, FileStore, MsgDecodeFileInfoPollerParser>
 where
     T: Clone,
 {
-    FileInfoPollerConfigBuilder::<T, S, FileStore, MsgDecodeFileInfoPollerParser>::default()
-        .parser(MsgDecodeFileInfoPollerParser)
+    Continuous::msg_source::<T, S>()
 }
 
 pub fn source<I, P>(paths: I) -> BytesMutStream

mobile_verifier/Cargo.toml

@@ -63,3 +63,4 @@ coverage-map = { path = "../coverage_map" }
 
 [dev-dependencies]
 backon = "0"
+proptest = "1.5.0"

mobile_verifier/migrations/37_sp_promotions.sql

@@ -0,0 +1,21 @@
+CREATE TABLE IF NOT EXISTS subscriber_promotion_rewards (
+       time_of_reward TIMESTAMPTZ NOT NULL,
+       subscriber_id BYTEA NOT NULL,
+       carrier_key TEXT NOT NULL,
+       shares BIGINT NOT NULL,
+       PRIMARY KEY (time_of_reward, subscriber_id, carrier_key)
+);
+
+CREATE TABLE IF NOT EXISTS gateway_promotion_rewards (
+       time_of_reward TIMESTAMPTZ NOT NULL,
+       gateway_key TEXT NOT NULL,
+       carrier_key TEXT NOT NULL,
+       shares BIGINT NOT NULL,
+       PRIMARY KEY (time_of_reward, gateway_key, carrier_key) 
+);
+
+CREATE TABLE IF NOT EXISTS service_provider_promotion_funds (
+       service_provider BIGINT NOT NULL PRIMARY KEY,
+       basis_points BIGINT NOT NULL,
+       inserted_at TIMESTAMPTZ NOT NULL
+);

mobile_verifier/src/cli/mod.rs

@@ -1,3 +1,4 @@
+pub mod promotion_funds;
 pub mod reward_from_db;
 pub mod server;
 pub mod verify_disktree;

mobile_verifier/src/cli/promotion_funds.rs

@@ -0,0 +1,59 @@
+use crate::{
+    service_provider::promotions::funds::{
+        delete_promotion_fund, fetch_promotion_funds, save_promotion_fund,
+    },
+    Settings,
+};
+
+#[derive(Debug, clap::Args)]
+pub struct Cmd {
+    #[clap(subcommand)]
+    sub_command: SubCommand,
+}
+
+#[derive(Debug, clap::Subcommand)]
+enum SubCommand {
+    /// Print Service Provider promotions in mobile-verifier db
+    List,
+    /// Set Service Provider promotion in mobile-verifier db
+    Set {
+        service_provider_id: i32,
+        basis_points: u16,
+    },
+    /// Remove Service Provider promotion allocation from mobile-verifier db
+    Unset { service_provider_id: i32 },
+}
+
+impl Cmd {
+    pub async fn run(&self, settings: &Settings) -> anyhow::Result<()> {
+        let pool = settings.database.connect(env!("CARGO_PKG_NAME")).await?;
+
+        match self.sub_command {
+            SubCommand::List => {
+                let funds = fetch_promotion_funds(&pool).await?;
+                println!("{funds:?}");
+            }
+            SubCommand::Set {
+                service_provider_id,
+                basis_points,
+            } => {
+                let mut txn = pool.begin().await?;
+                save_promotion_fund(&mut txn, service_provider_id, basis_points).await?;
+                txn.commit().await?;
+
+                let funds = fetch_promotion_funds(&pool).await?;
+                println!("{funds:?}");
+            }
+            SubCommand::Unset {
+                service_provider_id,
+            } => {
+                delete_promotion_fund(&pool, service_provider_id).await?;
+
+                let funds = fetch_promotion_funds(&pool).await?;
+                println!("{funds:?}");
+            }
+        }
+
+        Ok(())
+    }
+}

mobile_verifier/src/cli/server.rs

@@ -8,6 +8,7 @@ use crate::{
     heartbeats::{cbrs::CbrsHeartbeatDaemon, wifi::WifiHeartbeatDaemon},
     radio_threshold::RadioThresholdIngestor,
     rewarder::Rewarder,
+    service_provider,
     sp_boosted_rewards_bans::ServiceProviderBoostedRewardsBanIngestor,
     speedtests::SpeedtestDaemon,
     subscriber_location::SubscriberLocationIngestor,
@@ -137,7 +138,7 @@ impl Cmd {
                     file_upload.clone(),
                     report_ingest.clone(),
                     speedtests_avg.clone(),
-                    gateway_client,
+                    gateway_client.clone(),
                 )
                 .await?,
             )
@@ -193,6 +194,18 @@ impl Cmd {
                 )
                 .await?,
             )
+            .add_task(
+                service_provider::PromotionDaemon::create_managed_task(
+                    pool.clone(),
+                    settings,
+                    file_upload.clone(),
+                    report_ingest.clone(),
+                    gateway_client.clone(),
+                    auth_client.clone(),
+                    entity_client.clone(),
+                )
+                .await?,
+            )
             .add_task(DataSessionIngestor::create_managed_task(pool.clone(), settings).await?)
             .add_task(
                 ServiceProviderBoostedRewardsBanIngestor::create_managed_task(

mobile_verifier/src/data_session.rs

@@ -10,8 +10,6 @@ use futures::{
     TryFutureExt,
 };
 use helium_crypto::PublicKeyBinary;
-use helium_proto::ServiceProvider;
-use rust_decimal::Decimal;
 use sqlx::{PgPool, Pool, Postgres, Row, Transaction};
 use std::{collections::HashMap, ops::Range, time::Instant};
 use task_manager::{ManagedTask, TaskManager};
@@ -30,12 +28,6 @@ pub struct HotspotReward {
     pub rewardable_dc: u64,
 }
 
-#[derive(Clone, Debug)]
-pub struct ServiceProviderDataSession {
-    pub service_provider: ServiceProvider,
-    pub total_dcs: Decimal,
-}
-
 pub type HotspotMap = HashMap<PublicKeyBinary, HotspotReward>;
 
 impl DataSessionIngestor {