Service Provider Promotions again (#877)

* Remove unused files

* Update service provider promotions types in file_store

* Rework service provider promotion rewards

We now get incentive fund bps and active promotions from mobile-config
list incentive promotions from mobile-config client

* Start to fill out listing incentive programs for service providers from mobile-config

* Redo carrier client listing incentive promotions

* Raise dc_sessions and sp_promotions

sp_promotions needs to be written to the reward manifest. If we pull it up a level, we can uncouple service provider rewards from carrier_client and the database.

* Remove funds container, replace with promotions that has bps and promotions

* Use ServiceProviderPromotions wrapper

* clean up test structure

* clean up exported naming from service provider

* update to use published proto types

* Add metadata db to carrier service

Fill out query to return all active service provider promotions

* Keep importing RadioReward until we stop reading them

* Use request timestamp to filter active promotions returned

* update to rebased proto branch

* go back to proto master

file_store/src/cli/dump_mobile_rewards.rs

@@ -24,7 +24,7 @@ impl Cmd {
         let mut subscriber_reward = vec![];
         let mut service_provider_reward = vec![];
         let mut unallocated_reward = vec![];
-        // let mut promotion_reward = vec![];
+        let mut promotion_reward = vec![];
 
         while let Some(result) = file_stream.next().await {
             let msg = result?;
@@ -62,21 +62,11 @@ impl Cmd {
                         "unallocated_reward_type": reward.reward_type,
                         "amount": reward.amount,
                     })),
-                    // PromotionReward(reward) => {
-                    //     let entity = reward.entity.unwrap();
-                    //     match entity {
-                    //         Entity::SubscriberId(id) => promotion_reward.push(json!({
-                    //             "subscriber_id": uuid::Uuid::from_slice(&id).unwrap(),
-                    //             "service_provider_amount": reward.service_provider_amount,
-                    //             "matched_amount": reward.matched_amount,
-                    //         })),
-                    //         Entity::GatewayKey(key) => promotion_reward.push(json!({
-                    //             "gateway_key": PublicKey::try_from(key)?.to_string(),
-                    //             "service_provider_amount": reward.service_provider_amount,
-                    //             "matched_amount": reward.matched_amount,
-                    //         })),
-                    //     }
-                    // }
+                    PromotionReward(reward) => promotion_reward.push(json!({
+                        "entity": reward.entity,
+                        "service_provider_amount": reward.service_provider_amount,
+                        "matched_amount": reward.matched_amount,
+                    })),
                 },
                 None => todo!(),
             }
@@ -88,7 +78,7 @@ impl Cmd {
             "gateway_reward": gateway_reward,
             "subscriber_reward": subscriber_reward,
             "service_provider_reward": service_provider_reward,
-            // "promotion_reward": promotion_reward,
+            "promotion_reward": promotion_reward,
             "unallocated_reward": unallocated_reward,
         }))?;
 

file_store/src/lib.rs

@@ -20,7 +20,6 @@ pub mod mobile_radio_threshold;
 pub mod mobile_session;
 pub mod mobile_subscriber;
 pub mod mobile_transfer;
-pub mod promotion_reward;
 pub mod reward_manifest;
 mod settings;
 pub mod speedtest;

file_store/src/traits/file_sink_write.rs

@@ -273,18 +273,3 @@ impl_file_sink!(
     FileType::RewardManifest.to_str(),
     "reward_manifest"
 );
-impl_file_sink!(
-    proto::ServiceProviderPromotionFundV1,
-    FileType::ServiceProviderPromotionFund.to_str(),
-    "service_provider_promotion_fund"
-);
-impl_file_sink!(
-    poc_mobile::PromotionRewardIngestReportV1,
-    FileType::PromotionRewardIngestReport.to_str(),
-    "promotion_reward_ingest_report"
-);
-impl_file_sink!(
-    poc_mobile::VerifiedPromotionRewardV1,
-    FileType::VerifiedPromotionReward.to_str(),
-    "verified_promotion_reward"
-);

file_store/src/traits/msg_verify.rs

@@ -84,6 +84,8 @@ impl_msg_verify!(mobile_config::AuthorizationListReqV1, signature);
 impl_msg_verify!(mobile_config::AuthorizationListResV1, signature);
 impl_msg_verify!(mobile_config::EntityVerifyReqV1, signature);
 impl_msg_verify!(mobile_config::EntityVerifyResV1, signature);
+impl_msg_verify!(mobile_config::CarrierIncentivePromotionListReqV1, signature);
+impl_msg_verify!(mobile_config::CarrierIncentivePromotionListResV1, signature);
 impl_msg_verify!(mobile_config::CarrierKeyToEntityReqV1, signature);
 impl_msg_verify!(mobile_config::CarrierKeyToEntityResV1, signature);
 impl_msg_verify!(mobile_config::GatewayInfoReqV1, signature);
@@ -95,7 +97,6 @@ impl_msg_verify!(mobile_config::BoostedHexInfoStreamReqV1, signature);
 impl_msg_verify!(mobile_config::BoostedHexModifiedInfoStreamReqV1, signature);
 impl_msg_verify!(mobile_config::BoostedHexInfoStreamResV1, signature);
 impl_msg_verify!(poc_mobile::SubscriberVerifiedMappingEventReqV1, signature);
-impl_msg_verify!(poc_mobile::PromotionRewardReqV1, signature);
 
 #[cfg(test)]
 mod test {

ingest/src/server_mobile.rs

@@ -14,8 +14,7 @@ use helium_proto::services::poc_mobile::{
     CoverageObjectIngestReportV1, CoverageObjectReqV1, CoverageObjectRespV1,
     DataTransferSessionIngestReportV1, DataTransferSessionReqV1, DataTransferSessionRespV1,
     InvalidatedRadioThresholdIngestReportV1, InvalidatedRadioThresholdReportReqV1,
-    InvalidatedRadioThresholdReportRespV1, PromotionRewardIngestReportV1, PromotionRewardReqV1,
-    PromotionRewardRespV1, RadioThresholdIngestReportV1, RadioThresholdReportReqV1,
+    InvalidatedRadioThresholdReportRespV1, RadioThresholdIngestReportV1, RadioThresholdReportReqV1,
     RadioThresholdReportRespV1, ServiceProviderBoostedRewardsBannedRadioIngestReportV1,
     ServiceProviderBoostedRewardsBannedRadioReqV1, ServiceProviderBoostedRewardsBannedRadioRespV1,
     SpeedtestIngestReportV1, SpeedtestReqV1, SpeedtestRespV1, SubscriberLocationIngestReportV1,
@@ -47,7 +46,6 @@ pub struct GrpcServer {
     sp_boosted_rewards_ban_sink:
         FileSinkClient<ServiceProviderBoostedRewardsBannedRadioIngestReportV1>,
     subscriber_mapping_event_sink: FileSinkClient<SubscriberVerifiedMappingEventIngestReportV1>,
-    promotion_reward_sink: FileSinkClient<PromotionRewardIngestReportV1>,
     required_network: Network,
     address: SocketAddr,
     api_token: MetadataValue<Ascii>,
@@ -87,7 +85,6 @@ impl GrpcServer {
             ServiceProviderBoostedRewardsBannedRadioIngestReportV1,
         >,
         subscriber_mapping_event_sink: FileSinkClient<SubscriberVerifiedMappingEventIngestReportV1>,
-        promotion_reward_sink: FileSinkClient<PromotionRewardIngestReportV1>,
         required_network: Network,
         address: SocketAddr,
         api_token: MetadataValue<Ascii>,
@@ -103,7 +100,6 @@ impl GrpcServer {
             coverage_object_report_sink,
             sp_boosted_rewards_ban_sink,
             subscriber_mapping_event_sink,
-            promotion_reward_sink,
             required_network,
             address,
             api_token,
@@ -441,30 +437,6 @@ impl poc_mobile::PocMobile for GrpcServer {
         let id = timestamp.to_string();
         Ok(Response::new(SubscriberVerifiedMappingEventResV1 { id }))
     }
-
-    async fn submit_promotion_reward(
-        &self,
-        request: Request<PromotionRewardReqV1>,
-    ) -> GrpcResult<PromotionRewardRespV1> {
-        let received_timestamp: u64 = Utc::now().timestamp_millis() as u64;
-        let event = request.into_inner();
-
-        custom_tracing::record_b58("pub_key", &event.carrier_pub_key);
-
-        let report = self
-            .verify_public_key(event.carrier_pub_key.as_ref())
-            .and_then(|public_key| self.verify_network(public_key))
-            .and_then(|public_key| self.verify_signature(public_key, event))
-            .map(|(_, event)| PromotionRewardIngestReportV1 {
-                received_timestamp,
-                report: Some(event),
-            })?;
-
-        let _ = self.promotion_reward_sink.write(report, []).await;
-
-        let id = received_timestamp.to_string();
-        Ok(Response::new(PromotionRewardRespV1 { id }))
-    }
 }
 
 pub async fn grpc_server(settings: &Settings) -> Result<()> {
@@ -574,16 +546,6 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         )
         .await?;
 
-    let (subscriber_referral_eligibility_sink, subscriber_referral_eligibility_server) =
-        PromotionRewardIngestReportV1::file_sink(
-            store_base_path,
-            file_upload.clone(),
-            FileSinkCommitStrategy::Automatic,
-            FileSinkRollTime::Duration(settings.roll_time),
-            env!("CARGO_PKG_NAME"),
-        )
-        .await?;
-
     let Some(api_token) = settings
         .token
         .as_ref()
@@ -603,7 +565,6 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         coverage_object_report_sink,
         sp_boosted_rewards_ban_sink,
         subscriber_mapping_event_sink,
-        subscriber_referral_eligibility_sink,
         settings.network,
         settings.listen_addr,
         api_token,
@@ -627,7 +588,6 @@ pub async fn grpc_server(settings: &Settings) -> Result<()> {
         .add_task(coverage_object_report_sink_server)
         .add_task(sp_boosted_rewards_ban_sink_server)
         .add_task(subscriber_mapping_event_server)
-        .add_task(subscriber_referral_eligibility_server)
         .add_task(grpc_server)
         .build()
         .start()

ingest/tests/common/mod.rs

@@ -44,7 +44,6 @@ pub async fn setup_mobile() -> anyhow::Result<(TestClient, Trigger)> {
     let (coverage_obj_tx, _rx) = tokio::sync::mpsc::channel(10);
     let (sp_boosted_tx, _rx) = tokio::sync::mpsc::channel(10);
     let (subscriber_mapping_tx, subscriber_mapping_rx) = tokio::sync::mpsc::channel(10);
-    let (promotion_rewards_tx, _rx) = tokio::sync::mpsc::channel(10);
 
     tokio::spawn(async move {
         let grpc_server = GrpcServer::new(
@@ -58,7 +57,6 @@ pub async fn setup_mobile() -> anyhow::Result<(TestClient, Trigger)> {
             FileSinkClient::new(coverage_obj_tx, "noop"),
             FileSinkClient::new(sp_boosted_tx, "noop"),
             FileSinkClient::new(subscriber_mapping_tx, "test_file_sink"),
-            FileSinkClient::new(promotion_rewards_tx, "noop"),
             Network::MainNet,
             socket_addr,
             api_token,