Service Provider Rewards and Promotions (#867)

* Add promotion funds workspace

This workspace is all about dealing with Service Provider Promotion Fund
allocation.

HIP-114
https://github.com/helium/HIP/blob/main/0114-incentive-escrow-fund-for-subscriber-referrals.md

Service Provider Promotions are stored in CarrierV0 on Solana.
To keep the mobile-verifier from talking to a chain, this service will
periodically check Solana and compare Service Providers allocations to
what is stored in S3.

If the values have changed, a new file will be output to a bucket for
the mobile-verifier rewarder to read from.

NOTE: Allocation Values are stored in Bps (Basis Points)
https://www.investopedia.com/terms/b/basispoint.asp

** Commands

*** ./promotion_fund write-solana

Fetch Allocation values from Solana and write them to S3.
This command _always_ writes an S3 file.

*** ./promotion_fund print-s3

Using the lookback time in the provided settings file, show the
Allocation values this service would start up with.

*** ./promotion_fund server

Start a server that reads from S3, then checks with Solana periodically
for updated Allocatino values. Writing new files when needed.

* Supporting material for promotion_fund workspace

- ingest promotion rewards, nothing will be done with them until the
  processor is added into mobile-verifier.
- dump reward files
- add sp_allocations dummy field to rewarder output
- reward indexer mobile promotion type added

* use existing reward_types for reward_indexer

Otherwise inserting a new reward would match on the address and
continually change the reward_type column for no reason.

* update proto

* Add Continuous struct helper for file_source

Default trait impls cannot be added to functions, but they can be added
to structs. This struct keeps us from needing to do the gross blank
generic filling in when we want a continuous file source with a decode
other than MsgDecodeFileInfoPollerParser.

* Add the Service Provider Promotions Daemon

* Swap over Service Provider rewards to use percents

Rather than always dealing with conrete numbers, now that Service
Providers can allocate a percentage of their rewards to promotions for
subscribers or gateways, it becomes easier to speak of everything in
percentages.

This is a quick overview of what a more detailed overview you will find
in the PR.

For every Service Provider (SP) we figure out how much of the total
rewards allocation they are being awarded for data transfer. We get the
percent they have allocated for promotions (essentially from solana, but
really s3), and we determine which percentage of the _total rewards
allocation_ that is.

If a SP is getting 50% of the total rewards, and they allocate 20% of
those rewards to promotions, then the SP will receive 40% of the total,
and promotions from them will represent 10% of the total.

We do that for all SPs. The unallocated percentage is then distributed
to the promotions of each SP. If there is more than enough unallocated
left over, each SP get's a matched percentage of the whole to what they
set aside. When there is not enough, they get a percentage equal to
their initial rewards percentage.

This is to keep a service provider from getting the bulk of extra
rewards by settings aside a large amount for promotions, and receiving
little in rewards for data transfer, but getting more for matching.

A SP may never receive more in matched rewards than they have allocated
themselves.

* Wrap everything in a service provider type

This makes invocations of rewarding look very consistent

* Update tests to use new service_provider code

* remove unused code

* use wrapped types for promotions

* add back promotion_fund cli

* port over existing tests

* service provider reward info collection can hold onto the reward date

* add test with seeded promotions

* the nature of the math has changed

because we're dealing with percentages, and not doing bankers rounding
or nearest even, the likelihood that a calculation comes out extremely
close to a hole number then gets rounded down is increased. Especially
when dealing with percentages to the 7th decimal point and a base number
in the trillions.

* make sure we're cleaning up promotions

* stub for reporting sp allocations

I'm hoping there's an easy way to do this someone can point me to

* rename collection class for consistency

it get a bit verbose with all the service_provider flying around, but I
think it will be rather helpful when you only have the type and it
hasn't been aliased.

* fix up uncompiled tests

* add tests from good examples in original PR

#856

* if for some reason a key is duplicated, absorb both

* add proptests for service provider rewards

* Found a rounding error

truncating to 5 decimal places for a % can result in the summed
allocation exceeding 100%.

Thanks proptest

* Discussion has been had

It has been noted that in tests a single bone is okay, as long as we're
not going over the allocated bones

* commit commit

let's not have the worst of both worlds where we comment out unused
code.
Remove the printlns, they're not hard to add back in if you need them.

* fill out service provider reward allocations in reward_manifest

* Service Providers can use multiple keys for dc_transfer

For accounting reasons, Service Providers can have multiple payer keys
listed. We need to ensuure all their dc transfer are accumulated into
their service provider entry when we are generating rewards.

* fixup cargo deps after rebase

* Change promotion_fund to promotion-fund

* Read Service Provider Promotion values from unique bucket

- Add settings `promotion_ingest`

---------

Co-authored-by: Brian Balser <[email protected]>

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/pkg/settings-template.toml

@@ -47,6 +47,14 @@ block = 0
 #
 bucket = "mainnet-mobile-ingest"
 
+[promotion_ingest]
+
+# Input bucket details for Service Provider Promotion Funds
+
+# Name of bucket to access ingest data. Required
+#
+bucket = "price"
+
 # Region for bucket. Defaults to below
 #
 # region = "us-west-2"

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(())
+    }
+}