Track user_registered and diamond_membership_payment events

Cargo.toml

@@ -161,10 +161,10 @@ clap = "4.4.8"
 dfx-core = { git = "https://github.com/hpeebles/dfinity-sdk", rev = "d39dde0611fda8f30a6796b40b1dbb13a0541597" }
 dirs = "5.0.1"
 dotenv = "0.15.0"
-event_sink_canister = { git = "https://github.com/open-chat-labs/event-sink", rev = "84f5ffb8aa638e4ed43d0392ad4e119963fad3df" }
-event_sink_client = { git = "https://github.com/open-chat-labs/event-sink", rev = "84f5ffb8aa638e4ed43d0392ad4e119963fad3df" }
-event_sink_client_cdk_runtime = { git = "https://github.com/open-chat-labs/event-sink", rev = "84f5ffb8aa638e4ed43d0392ad4e119963fad3df" }
-event_sink_utils = { git = "https://github.com/open-chat-labs/event-sink", rev = "84f5ffb8aa638e4ed43d0392ad4e119963fad3df" }
+event_sink_canister = { git = "https://github.com/open-chat-labs/event-sink", rev = "e8e4a547f767ef568ddf978c6cc421a1ef05485e" }
+event_sink_client = { git = "https://github.com/open-chat-labs/event-sink", rev = "e8e4a547f767ef568ddf978c6cc421a1ef05485e" }
+event_sink_client_cdk_runtime = { git = "https://github.com/open-chat-labs/event-sink", rev = "e8e4a547f767ef568ddf978c6cc421a1ef05485e" }
+event_sink_utils = { git = "https://github.com/open-chat-labs/event-sink", rev = "e8e4a547f767ef568ddf978c6cc421a1ef05485e" }
 futures = "0.3.29"
 getrandom = { version = "0.2.11", features = ["custom"] }
 hex = "0.4.3"

backend/canister_installer/src/lib.rs

@@ -69,6 +69,7 @@ async fn install_service_canisters_impl(
         storage_index_canister_id: canister_ids.storage_index,
         cycles_dispenser_canister_id: canister_ids.cycles_dispenser,
         escrow_canister_id: canister_ids.escrow,
+        event_relay_canister_id: canister_ids.event_relay,
         nns_governance_canister_id: canister_ids.nns_governance,
         internet_identity_canister_id: canister_ids.nns_internet_identity,
         translations_canister_id: canister_ids.translations,

backend/canisters/user_index/CHANGELOG.md

@@ -5,6 +5,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ## [unreleased]
 
+### Added
+
+- Track `user_registered` and `diamond_membership_payment` events ([#5342](https://github.com/open-chat-labs/open-chat/pull/5342))
+
 ## [[2.0.1047](https://github.com/open-chat-labs/open-chat/releases/tag/v2.0.1047-user_index)] - 2024-02-05
 
 ### Added

backend/canisters/user_index/api/src/lifecycle/init.rs

@@ -14,6 +14,7 @@ pub struct Args {
     pub cycles_dispenser_canister_id: CanisterId,
     pub storage_index_canister_id: CanisterId,
     pub escrow_canister_id: CanisterId,
+    pub event_relay_canister_id: CanisterId,
     pub nns_governance_canister_id: CanisterId,
     pub internet_identity_canister_id: CanisterId,
     pub translations_canister_id: CanisterId,

backend/canisters/user_index/impl/Cargo.toml

@@ -20,6 +20,8 @@ canister_tracing_macros = { path = "../../../libraries/canister_tracing_macros"
 chat_events = { path = "../../../libraries/chat_events" }
 community_canister = { path = "../../community/api" }
 community_canister_c2c_client = { path = "../../community/c2c_client" }
+event_sink_client = { workspace = true }
+event_sink_client_cdk_runtime = { workspace = true }
 futures = { workspace = true }
 group_canister = { path = "../../group/api" }
 group_canister_c2c_client = { path = "../../group/c2c_client" }

backend/canisters/user_index/impl/src/lib.rs

@@ -7,6 +7,8 @@ use crate::timer_job_types::TimerJob;
 use candid::Principal;
 use canister_state_macros::canister_state;
 use canister_timer_jobs::TimerJobs;
+use event_sink_client::{EventSinkClient, EventSinkClientBuilder};
+use event_sink_client_cdk_runtime::CdkRuntime;
 use fire_and_forget_handler::FireAndForgetHandler;
 use icrc_ledger_types::icrc1::account::{Account, Subaccount};
 use local_user_index_canister::Event as LocalUserIndexEvent;
@@ -20,6 +22,7 @@ use nns_governance_canister::types::{Empty, ManageNeuron, NeuronId};
 use serde::{Deserialize, Serialize};
 use std::cell::RefCell;
 use std::collections::{HashMap, HashSet, VecDeque};
+use std::time::Duration;
 use types::{
     BuildVersion, CanisterId, CanisterWasm, ChatId, Cryptocurrency, Cycles, DiamondMembershipFees, Milliseconds,
     TimestampMillis, Timestamped, UserId,
@@ -148,6 +151,18 @@ impl RuntimeState {
         jobs::sync_events_to_local_user_index_canisters::try_run_now(self);
     }
 
+    pub fn track_event<T: Serialize>(&mut self, name: &str, timestamp: TimestampMillis, user: Option<UserId>, payload: T) {
+        let payload_json = serde_json::to_vec(&payload).unwrap();
+
+        self.data.event_sink_client.push_event(event_sink_client::Event {
+            name: name.to_string(),
+            timestamp,
+            user: user.map(|u| u.to_string()),
+            source: Some(self.env.canister_id().to_text()),
+            payload: payload_json,
+        });
+    }
+
     pub fn queue_payment(&mut self, pending_payment: PendingPayment) {
         self.data.pending_payments_queue.push(pending_payment);
         jobs::make_pending_payments::start_job_if_required(self);
@@ -220,13 +235,12 @@ struct Data {
     pub cycles_dispenser_canister_id: CanisterId,
     pub storage_index_canister_id: CanisterId,
     pub escrow_canister_id: CanisterId,
-    #[serde(default = "translations_canister_id")]
     pub translations_canister_id: CanisterId,
+    #[serde(default = "event_sink_client")]
+    pub event_sink_client: EventSinkClient<CdkRuntime>,
     pub storage_index_user_sync_queue: OpenStorageUserSyncQueue,
     pub user_index_event_sync_queue: CanisterEventSyncQueue<LocalUserIndexEvent>,
-    #[serde(default)]
     pub user_principal_updates_queue: UserPrincipalUpdatesQueue,
-    #[serde(default)]
     pub legacy_principals_sync_queue: VecDeque<Principal>,
     pub pending_payments_queue: PendingPaymentsQueue,
     pub pending_modclub_submissions_queue: PendingModclubSubmissionsQueue,
@@ -247,12 +261,14 @@ struct Data {
     pub nns_8_year_neuron: Option<NnsNeuron>,
     pub rng_seed: [u8; 32],
     pub diamond_membership_fees: DiamondMembershipFees,
-    #[serde(default)]
     pub legacy_principals_synced: bool,
 }
 
-fn translations_canister_id() -> CanisterId {
-    Principal::from_text("lxq5i-mqaaa-aaaaf-bih7q-cai").unwrap()
+fn event_sink_client() -> EventSinkClient<CdkRuntime> {
+    let event_relay_canister_id = CanisterId::from_text("6ofpc-2aaaa-aaaaf-biibq-cai").unwrap();
+    EventSinkClientBuilder::new(event_relay_canister_id, CdkRuntime::default())
+        .with_flush_delay(Duration::from_secs(60))
+        .build()
 }
 
 impl Data {
@@ -268,6 +284,7 @@ impl Data {
         cycles_dispenser_canister_id: CanisterId,
         storage_index_canister_id: CanisterId,
         escrow_canister_id: CanisterId,
+        event_relay_canister_id: CanisterId,
         nns_governance_canister_id: CanisterId,
         internet_identity_canister_id: CanisterId,
         translations_canister_id: CanisterId,
@@ -289,6 +306,9 @@ impl Data {
             storage_index_canister_id,
             escrow_canister_id,
             translations_canister_id,
+            event_sink_client: EventSinkClientBuilder::new(event_relay_canister_id, CdkRuntime::default())
+                .with_flush_delay(Duration::from_secs(60))
+                .build(),
             storage_index_user_sync_queue: OpenStorageUserSyncQueue::default(),
             user_index_event_sync_queue: CanisterEventSyncQueue::default(),
             user_principal_updates_queue: UserPrincipalUpdatesQueue::default(),
@@ -375,6 +395,7 @@ impl Default for Data {
             storage_index_canister_id: Principal::anonymous(),
             escrow_canister_id: Principal::anonymous(),
             translations_canister_id: Principal::anonymous(),
+            event_sink_client: EventSinkClientBuilder::new(Principal::anonymous(), CdkRuntime::default()).build(),
             storage_index_user_sync_queue: OpenStorageUserSyncQueue::default(),
             user_index_event_sync_queue: CanisterEventSyncQueue::default(),
             user_principal_updates_queue: UserPrincipalUpdatesQueue::default(),
@@ -459,6 +480,12 @@ pub struct NnsNeuron {
     pub subaccount: Subaccount,
 }
 
+#[derive(Serialize)]
+struct UserRegisteredEventPayload {
+    referred: bool,
+    is_bot: bool,
+}
+
 #[derive(Serialize, Debug)]
 pub struct CanisterIds {
     pub group_index: CanisterId,

backend/canisters/user_index/impl/src/lifecycle/init.rs

@@ -25,6 +25,7 @@ fn init(args: Args) {
         args.cycles_dispenser_canister_id,
         args.storage_index_canister_id,
         args.escrow_canister_id,
+        args.event_relay_canister_id,
         args.nns_governance_canister_id,
         args.internet_identity_canister_id,
         args.translations_canister_id,

backend/canisters/user_index/impl/src/updates/c2c_notify_events.rs

@@ -1,6 +1,6 @@
 use crate::guards::caller_is_local_user_index_canister;
 use crate::timer_job_types::{JoinUserToGroup, TimerJob};
-use crate::{mutate_state, RuntimeState, ONE_MB};
+use crate::{mutate_state, RuntimeState, UserRegisteredEventPayload, ONE_MB};
 use candid::Principal;
 use canister_api_macros::update_msgpack;
 use canister_tracing_macros::trace;
@@ -114,6 +114,17 @@ fn process_new_user(
         }),
         Some(local_user_index_canister_id),
     );
+
+    state.track_event(
+        "user_registered",
+        now,
+        Some(user_id),
+        UserRegisteredEventPayload {
+            referred: referred_by.is_some(),
+            is_bot: false,
+        },
+    );
+
     if let Some(original_username) = original_username {
         state.push_event_to_local_user_index(
             user_id,

backend/canisters/user_index/impl/src/updates/c2c_register_bot.rs

@@ -60,5 +60,15 @@ fn c2c_register_bot_impl(args: Args, state: &mut RuntimeState) -> Response {
         None,
     );
 
+    state.track_event(
+        "user_registered",
+        now,
+        Some(user_id),
+        crate::UserRegisteredEventPayload {
+            referred: false,
+            is_bot: true,
+        },
+    );
+
     Success
 }

backend/canisters/user_index/impl/src/updates/pay_for_diamond_membership.rs

@@ -10,6 +10,7 @@ use icrc_ledger_types::icrc1;
 use icrc_ledger_types::icrc1::account::Account;
 use local_user_index_canister::{DiamondMembershipPaymentReceived, Event};
 use rand::Rng;
+use serde::Serialize;
 use storage_index_canister::add_or_update_users::UserConfig;
 use tracing::error;
 use types::{Cryptocurrency, DiamondMembershipFees, DiamondMembershipPlanDuration, UserId, ICP};
@@ -97,9 +98,22 @@ fn process_charge(
 ) -> Response {
     let share_with = referrer_to_share_payment(user_id, state);
     let recurring = args.recurring && !args.duration.is_lifetime();
+    let now = state.env.now();
+
+    state.track_event(
+        "diamond_membership_payment",
+        now,
+        Some(user_id),
+        PayForDiamondMembershipEventPayload {
+            token: args.token.token_symbol().to_string(),
+            amount: args.expected_price_e8s,
+            duration: args.duration.to_string(),
+            recurring: args.recurring,
+            manual_payment,
+        },
+    );
 
     if let Some(diamond_membership) = state.data.users.diamond_membership_details_mut(&user_id) {
-        let now = state.env.now();
         let has_ever_been_diamond_member = diamond_membership.has_ever_been_diamond_member();
 
         diamond_membership.add_payment(
@@ -281,3 +295,12 @@ fn process_error_v2(transfer_error: icrc1::transfer::TransferError) -> Response
         error => TransferFailed(format!("{error:?}")),
     }
 }
+
+#[derive(Serialize)]
+struct PayForDiamondMembershipEventPayload {
+    token: String,
+    amount: u64,
+    duration: String,
+    recurring: bool,
+    manual_payment: bool,
+}

backend/integration_tests/src/setup.rs

@@ -121,6 +121,7 @@ fn install_canisters(env: &mut PocketIc, controller: Principal) -> CanisterIds {
         cycles_dispenser_canister_id,
         storage_index_canister_id,
         escrow_canister_id,
+        event_relay_canister_id,
         nns_governance_canister_id,
         internet_identity_canister_id: NNS_INTERNET_IDENTITY_CANISTER_ID,
         translations_canister_id,

backend/libraries/types/src/diamond_membership.rs

@@ -1,6 +1,7 @@
 use crate::{Milliseconds, TimestampMillis};
 use candid::CandidType;
 use serde::{Deserialize, Serialize};
+use std::fmt::{Display, Formatter};
 
 #[derive(CandidType, Serialize, Deserialize, Clone, Debug, Default)]
 pub struct DiamondMembershipDetails {
@@ -145,3 +146,14 @@ impl TryFrom<DiamondMembershipSubscription> for DiamondMembershipPlanDuration {
         }
     }
 }
+
+impl Display for DiamondMembershipPlanDuration {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            DiamondMembershipPlanDuration::OneMonth => "1 month",
+            DiamondMembershipPlanDuration::ThreeMonths => "3 months",
+            DiamondMembershipPlanDuration::OneYear => "1 year",
+            DiamondMembershipPlanDuration::Lifetime => "Lifetime",
+        })
+    }
+}