feat: adding payment stream creation event to indexer (#213)

* adding payment stream creation event to indexer

* rustfmt

* handle payment stream payments to the provider (increase total_amount_paid)

* update schema; fix up.sql for payment stream

* fix total_amount and total_amount_paid naming incoherency

* fix diesel related errors

* fix handlers for payment stream

* fix clippy

* add tick to the payment stream events

* add TODO comment

* update the typescript files for the PaymentStreamCharged event

* pnpm fmt

* generate typescript types

* fix charged tick and add charged_at_tick value

* some fixes

* generate type for typescript

* fix test

* fix test again

* fix: 🩹 Make it so `last_chargeable_tick` is always set correctly and not left as 0 when user is without funds

* chore: 🏷️ Update `api-augment`

* chore: 🏷️ Update `api-augment`

* genearte types

---------

Co-authored-by: Facundo Farall <37149322+ffarall@users.noreply.github.com>

api-augment/dist/types/interfaces/augment-api-events.d.ts

@@ -1536,15 +1536,23 @@ declare module "@polkadot/api-base/types/events" {
       >;
       /**
        * Event emitted when a payment is charged. Provides information about the user that was charged,
-       * the Provider that received the funds, and the amount that was charged.
+       * the Provider that received the funds, the tick up to which it was charged and the amount that was charged.
        **/
       PaymentStreamCharged: AugmentedEvent<
         ApiType,
-        [userAccount: AccountId32, providerId: H256, amount: u128],
+        [
+          userAccount: AccountId32,
+          providerId: H256,
+          amount: u128,
+          lastTickCharged: u32,
+          chargedAtTick: u32
+        ],
         {
           userAccount: AccountId32;
           providerId: H256;
           amount: u128;
+          lastTickCharged: u32;
+          chargedAtTick: u32;
         }
       >;
       /**

api-augment/dist/types/interfaces/lookup.d.ts

@@ -1718,6 +1718,8 @@ declare const _default: {
         userAccount: string;
         providerId: string;
         amount: string;
+        lastTickCharged: string;
+        chargedAtTick: string;
       };
       LastChargeableInfoUpdated: {
         providerId: string;

api-augment/dist/types/interfaces/types-lookup.d.ts

@@ -2154,6 +2154,8 @@ declare module "@polkadot/types/lookup" {
       readonly userAccount: AccountId32;
       readonly providerId: H256;
       readonly amount: u128;
+      readonly lastTickCharged: u32;
+      readonly chargedAtTick: u32;
     } & Struct;
     readonly isLastChargeableInfoUpdated: boolean;
     readonly asLastChargeableInfoUpdated: {

api-augment/src/interfaces/augment-api-events.ts

@@ -1177,12 +1177,24 @@ declare module "@polkadot/api-base/types/events" {
       >;
       /**
        * Event emitted when a payment is charged. Provides information about the user that was charged,
-       * the Provider that received the funds, and the amount that was charged.
+       * the Provider that received the funds, the tick up to which it was charged and the amount that was charged.
        **/
       PaymentStreamCharged: AugmentedEvent<
         ApiType,
-        [userAccount: AccountId32, providerId: H256, amount: u128],
-        { userAccount: AccountId32; providerId: H256; amount: u128 }
+        [
+          userAccount: AccountId32,
+          providerId: H256,
+          amount: u128,
+          lastTickCharged: u32,
+          chargedAtTick: u32
+        ],
+        {
+          userAccount: AccountId32;
+          providerId: H256;
+          amount: u128;
+          lastTickCharged: u32;
+          chargedAtTick: u32;
+        }
       >;
       /**
        * Event emitted when a User that has been flagged as not having enough funds to pay for their contracted services has paid all its outstanding debt.

api-augment/src/interfaces/lookup.ts

@@ -1733,7 +1733,9 @@ export default {
       PaymentStreamCharged: {
         userAccount: "AccountId32",
         providerId: "H256",
-        amount: "u128"
+        amount: "u128",
+        lastTickCharged: "u32",
+        chargedAtTick: "u32"
       },
       LastChargeableInfoUpdated: {
         providerId: "H256",

api-augment/src/interfaces/types-lookup.ts

@@ -2257,6 +2257,8 @@ declare module "@polkadot/types/lookup" {
       readonly userAccount: AccountId32;
       readonly providerId: H256;
       readonly amount: u128;
+      readonly lastTickCharged: u32;
+      readonly chargedAtTick: u32;
     } & Struct;
     readonly isLastChargeableInfoUpdated: boolean;
     readonly asLastChargeableInfoUpdated: {

client/indexer-db/migrations/2024-10-01-112655_create_paymentstream/down.sql

@@ -0,0 +1,2 @@
+-- Drop the paymentstream table
+DROP TABLE IF EXISTS paymentstream;

client/indexer-db/migrations/2024-10-01-112655_create_paymentstream/up.sql

@@ -0,0 +1,14 @@
+CREATE TABLE paymentstream (
+    id SERIAL PRIMARY KEY,
+    account VARCHAR NOT NULL,
+    provider VARCHAR NOT NULL,
+    total_amount_paid NUMERIC(38, 0) NOT NULL DEFAULT 0,
+    last_tick_charged BIGINT NOT NULL DEFAULT 0,
+    charged_at_tick BIGINT NOT NULL DEFAULT 0
+);
+
+-- Create an index on the account column for faster lookups
+CREATE INDEX idx_paymentstream_account ON paymentstream(account);
+
+-- Create an index on the provider column for faster lookups
+CREATE INDEX idx_paymentstream_provider ON paymentstream(provider);

client/indexer-db/src/models/mod.rs

@@ -2,10 +2,12 @@ pub mod bsp;
 pub mod bucket;
 pub mod msp;
 pub mod multiaddress;
+pub mod payment_stream;
 pub mod service_state;
 
 pub use bsp::*;
 pub use bucket::*;
 pub use msp::*;
 pub use multiaddress::*;
+pub use payment_stream::*;
 pub use service_state::*;

client/indexer-db/src/models/payment_stream.rs

@@ -0,0 +1,76 @@
+use bigdecimal::BigDecimal;
+use diesel::prelude::*;
+use diesel_async::RunQueryDsl;
+
+use crate::{schema::paymentstream, DbConnection};
+
+#[derive(Debug, Queryable, Insertable, Selectable)]
+#[diesel(table_name = paymentstream)]
+pub struct PaymentStream {
+    // postgres attributed ID for this payment stream
+    pub id: i32,
+    // Account ID of the payer
+    pub account: String,
+    // ID of the payee (msp or bsp)
+    pub provider: String,
+    // Total amount already paid to this provider from this account for this payment stream
+    pub total_amount_paid: BigDecimal,
+    // The last tick for which the payment stream has recorded a payment
+    pub last_tick_charged: i64,
+    // The tick at which the payment actually happened
+    pub charged_at_tick: i64,
+}
+
+impl PaymentStream {
+    pub async fn create<'a>(
+        conn: &mut DbConnection<'a>,
+        account: String,
+        provider: String,
+    ) -> Result<Self, diesel::result::Error> {
+        let ps = diesel::insert_into(paymentstream::table)
+            .values((
+                paymentstream::account.eq(account),
+                paymentstream::provider.eq(provider),
+            ))
+            .returning(PaymentStream::as_select())
+            .get_result(conn)
+            .await?;
+        Ok(ps)
+    }
+
+    pub async fn get<'a>(
+        conn: &mut DbConnection<'a>,
+        account: String,
+        provider: String,
+    ) -> Result<Self, diesel::result::Error> {
+        // Looking by a payment stream by provider and the account associated
+        let ps = paymentstream::table
+            .filter(
+                paymentstream::account
+                    .eq(account)
+                    .and(paymentstream::provider.eq(provider)),
+            )
+            .first(conn)
+            .await?;
+        Ok(ps)
+    }
+
+    pub async fn update_total_amount<'a>(
+        conn: &mut DbConnection<'a>,
+        ps_id: i32,
+        new_total_amount: BigDecimal,
+        last_tick_charged: i64,
+        charged_at_tick: i64,
+    ) -> Result<(), diesel::result::Error> {
+        diesel::update(paymentstream::table)
+            .filter(paymentstream::id.eq(ps_id))
+            .set((
+                paymentstream::total_amount_paid.eq(new_total_amount),
+                paymentstream::last_tick_charged.eq(last_tick_charged),
+                paymentstream::charged_at_tick.eq(charged_at_tick),
+            ))
+            .execute(conn)
+            .await?;
+        Ok(())
+    }
+}

client/indexer-db/src/schema.rs

@@ -60,6 +60,17 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    paymentstream (id) {
+        id -> Int4,
+        account -> Varchar,
+        provider -> Varchar,
+        total_amount_paid -> Numeric,
+        last_tick_charged -> Int8,
+        charged_at_tick -> Int8,
+    }
+}
+
 diesel::table! {
     service_state (id) {
         id -> Int4,
@@ -82,5 +93,6 @@ diesel::allow_tables_to_appear_in_same_query!(
     msp,
     msp_multiaddress,
     multiaddress,
+    paymentstream,
     service_state,
 );

client/indexer-service/src/handler.rs

@@ -239,17 +239,57 @@ impl IndexerService {
 
     async fn index_payment_streams_event<'a, 'b: 'a>(
         &'b self,
-        _conn: &mut DbConnection<'a>,
+        conn: &mut DbConnection<'a>,
         event: &pallet_payment_streams::Event<storage_hub_runtime::Runtime>,
     ) -> Result<(), diesel::result::Error> {
         match event {
-            pallet_payment_streams::Event::DynamicRatePaymentStreamCreated { .. } => {}
-            pallet_payment_streams::Event::DynamicRatePaymentStreamUpdated { .. } => {}
+            pallet_payment_streams::Event::DynamicRatePaymentStreamCreated {
+                provider_id,
+                user_account,
+                amount_provided: _amount_provided,
+            } => {
+                PaymentStream::create(conn, provider_id.to_string(), user_account.to_string())
+                    .await?;
+            }
+            pallet_payment_streams::Event::DynamicRatePaymentStreamUpdated { .. } => {
+                // TODO: Currently we are not treating the info of dynamic rate update
+            }
             pallet_payment_streams::Event::DynamicRatePaymentStreamDeleted { .. } => {}
-            pallet_payment_streams::Event::FixedRatePaymentStreamCreated { .. } => {}
-            pallet_payment_streams::Event::FixedRatePaymentStreamUpdated { .. } => {}
+            pallet_payment_streams::Event::FixedRatePaymentStreamCreated {
+                provider_id,
+                user_account,
+                rate: _rate,
+            } => {
+                PaymentStream::create(conn, provider_id.to_string(), user_account.to_string())
+                    .await?;
+            }
+            pallet_payment_streams::Event::FixedRatePaymentStreamUpdated { .. } => {
+                // TODO: Currently we are not treating the info of fixed rate update
+            }
             pallet_payment_streams::Event::FixedRatePaymentStreamDeleted { .. } => {}
-            pallet_payment_streams::Event::PaymentStreamCharged { .. } => {}
+            pallet_payment_streams::Event::PaymentStreamCharged {
+                user_account,
+                provider_id,
+                amount,
+                last_tick_charged,
+                charged_at_tick,
+            } => {
+                // We want to handle this and update the payment stream total amount
+                let ps =
+                    PaymentStream::get(conn, user_account.to_string(), provider_id.to_string())
+                        .await?;
+                let new_total_amount = ps.total_amount_paid + amount;
+                let last_tick_charged: i64 = (*last_tick_charged).into();
+                let charged_at_tick: i64 = (*charged_at_tick).into();
+                PaymentStream::update_total_amount(
+                    conn,
+                    ps.id,
+                    new_total_amount,
+                    last_tick_charged,
+                    charged_at_tick,
+                )
+                .await?;
+            }
             pallet_payment_streams::Event::LastChargeableInfoUpdated { .. } => {}
             pallet_payment_streams::Event::UserWithoutFunds { .. } => {}
             pallet_payment_streams::Event::UserPaidDebts { .. } => {}

pallets/payment-streams/src/lib.rs

@@ -284,11 +284,13 @@ pub mod pallet {
             provider_id: ProviderIdFor<T>,
         },
         /// Event emitted when a payment is charged. Provides information about the user that was charged,
-        /// the Provider that received the funds, and the amount that was charged.
+        /// the Provider that received the funds, the tick up to which it was charged and the amount that was charged.
         PaymentStreamCharged {
             user_account: T::AccountId,
             provider_id: ProviderIdFor<T>,
             amount: BalanceOf<T>,
+            last_tick_charged: BlockNumberFor<T>,
+            charged_at_tick: BlockNumberFor<T>,
         },
         /// Event emitted when a Provider's last chargeable tick and price index are updated. Provides information about the Provider of the stream,
         /// the tick number of the last chargeable tick and the price index at that tick.
@@ -656,7 +658,7 @@ pub mod pallet {
         ///    4. Charge the user (if the user does not have enough funds, it gets flagged and a `UserWithoutFunds` event is emitted)
         ///    5. Update the last charged tick number of the payment stream
         /// 4. If there is a dynamic-rate payment stream:
-        ///    1. Get the amount provided by the Provider
+        ///    1. Get the amount provided by the Provider   
         ///    2. Get the difference between price index when the stream was last charged and the price index at the last chargeable tick
         ///    3. Calculate the amount to charge doing `amount_provided * difference`
         ///    4. Charge the user (if the user does not have enough funds, it gets flagged and a `UserWithoutFunds` event is emitted)
@@ -681,13 +683,19 @@ pub mod pallet {
                     .ok_or(Error::<T>::NotAProvider)?;
 
             // Execute checks and logic, update storage
-            let amount_charged = Self::do_charge_payment_streams(&provider_id, &user_account)?;
+            let (amount_charged, last_tick_charged) =
+                Self::do_charge_payment_streams(&provider_id, &user_account)?;
+
+            // Get the last tick to add it to the event
+            let charged_at_tick = Self::get_current_tick();
 
             // Emit the corresponding event (we always emit it even if the charged amount was 0)
             Self::deposit_event(Event::<T>::PaymentStreamCharged {
                 user_account,
                 provider_id: provider_id,
                 amount: amount_charged,
+                last_tick_charged,
+                charged_at_tick,
             });
 
             // Return a successful DispatchResultWithPostInfo