Clean ml feed (#473)

* reset ml feed

* reset ml feed

* test

* inc tick

* add more tick

* test print

* test print

* test print

* final push

* review changes

src/canister/individual_user_template/can.did

@@ -377,11 +377,11 @@ type Result_19 = variant { Ok : vec record { nat64; nat8 }; Err : text };
 type Result_2 = variant { Ok : bool; Err : CdaoTokenError };
 type Result_20 = variant { Ok; Err : MigrationErrors };
 type Result_21 = variant { Ok; Err : AirdropError };
-type Result_22 = variant { Ok : IndividualUserCreatorDaoEntry; Err : text };
-type Result_23 = variant { Committed : Committed; Aborted : record {} };
-type Result_24 = variant { Ok : Ok; Err : GovernanceError };
-type Result_25 = variant { Ok; Err : CdaoTokenError };
-type Result_26 = variant { Ok : text; Err : text };
+type Result_22 = variant { Ok : text; Err : text };
+type Result_23 = variant { Ok : IndividualUserCreatorDaoEntry; Err : text };
+type Result_24 = variant { Committed : Committed; Aborted : record {} };
+type Result_25 = variant { Ok : Ok; Err : GovernanceError };
+type Result_26 = variant { Ok; Err : CdaoTokenError };
 type Result_27 = variant {
   Ok : UserProfileDetailsForFrontend;
   Err : UpdateProfileDetailsError;
@@ -408,10 +408,10 @@ type RoomDetails = record {
 };
 type SessionType = variant { AnonymousSession; RegisteredSession };
 type SettleNeuronsFundParticipationRequest = record {
-  result : opt Result_23;
+  result : opt Result_24;
   nns_proposal_id : opt nat64;
 };
-type SettleNeuronsFundParticipationResponse = record { result : opt Result_24 };
+type SettleNeuronsFundParticipationResponse = record { result : opt Result_25 };
 type SlotDetails = record { room_details : vec record { nat64; RoomDetails } };
 type SnsInitPayload = record {
   url : opt text;
@@ -631,20 +631,21 @@ service : (IndividualUserTemplateInitArgs) -> {
   receive_bet_winnings_when_distributed : (nat64, BetOutcomeForBetMaker) -> ();
   receive_data_from_hotornot : (principal, nat64, vec Post) -> (Result_20);
   request_airdrop : (principal, opt blob, nat, principal) -> (Result_21);
+  reset_ml_feed_cache : () -> (Result_22);
   return_cycles_to_user_index_canister : (opt nat) -> ();
   save_snapshot_json : () -> (nat32);
-  send_creator_dao_stats_to_subnet_orchestrator : () -> (Result_22);
+  send_creator_dao_stats_to_subnet_orchestrator : () -> (Result_23);
   set_controller_as_subnet_orchestrator : (principal) -> ();
   settle_neurons_fund_participation : (
       SettleNeuronsFundParticipationRequest,
     ) -> (SettleNeuronsFundParticipationResponse);
   transfer_token_to_user_canister : (principal, principal, opt blob, nat) -> (
-      Result_25,
+      Result_26,
     );
   transfer_tokens_and_posts : (principal, principal) -> (Result_20);
-  update_last_access_time : () -> (Result_26);
+  update_last_access_time : () -> (Result_22);
   update_last_canister_functionality_access_time : () -> ();
-  update_ml_feed_cache : (vec MLFeedCacheItem) -> (Result_26);
+  update_ml_feed_cache : (vec MLFeedCacheItem) -> (Result_22);
   update_post_add_view_details : (nat64, PostViewDetailsFromFrontend) -> ();
   update_post_as_ready_to_view : (nat64) -> ();
   update_post_increment_share_count : (nat64) -> (nat64);
@@ -661,11 +662,11 @@ service : (IndividualUserTemplateInitArgs) -> {
   update_profiles_that_follow_me_toggle_list_with_specified_profile : (
       FollowerArg,
     ) -> (Result_8);
-  update_referrer_details : (UserCanisterDetails) -> (Result_26);
-  update_session_type : (SessionType) -> (Result_26);
-  update_success_history : (SuccessHistoryItemV1) -> (Result_26);
-  update_user_propensity : (float64) -> (Result_26);
-  update_watch_history : (WatchHistoryItem) -> (Result_26);
+  update_referrer_details : (UserCanisterDetails) -> (Result_22);
+  update_session_type : (SessionType) -> (Result_22);
+  update_success_history : (SuccessHistoryItemV1) -> (Result_22);
+  update_user_propensity : (float64) -> (Result_22);
+  update_watch_history : (WatchHistoryItem) -> (Result_22);
   update_well_known_principal : (KnownPrincipalType, principal) -> ();
   upgrade_creator_dao_governance_canisters : (blob) -> (Result_28);
   write_key_value_pair : (nat64, text, text) -> (Result_5);

src/canister/individual_user_template/src/api/ml_ops/ml_feed_cache.rs

@@ -61,6 +61,17 @@ fn get_ml_feed_cache_paginated_impl(
     canister_data.ml_feed_cache[start_index..end_index].to_vec()
 }
 
+#[update(guard = "is_caller_controller_or_global_admin")]
+fn reset_ml_feed_cache() -> Result<String, String> {
+    let _ = CANISTER_DATA.with(|canister_data| {
+        let mut canister_data = canister_data.borrow_mut();
+
+        canister_data.ml_feed_cache.clear();
+    });
+
+    Ok("Success".into())
+}
+
 #[cfg(test)]
 mod test {
     use candid::Principal;

src/canister/individual_user_template/src/api/ml_ops/update_success_history.rs

@@ -19,6 +19,11 @@ fn update_success_history(success_history_item: SuccessHistoryItemV1) -> Result<
         canister_data
             .success_history
             .insert(success_history_item, ());
+
+        // keep removing oldest items until the len is less than or equal to 3000
+        while canister_data.success_history.len() > 3000 {
+            canister_data.success_history.pop_first();
+        }
     });
 
     Ok("Success".into())

src/canister/individual_user_template/src/api/ml_ops/update_watch_history.rs

@@ -17,6 +17,11 @@ fn update_watch_history(watch_history_item: WatchHistoryItem) -> Result<String,
     CANISTER_DATA.with(|canister_data| {
         let mut canister_data = canister_data.borrow_mut();
         canister_data.watch_history.insert(watch_history_item, ());
+
+        // keep removing oldest items until the len is less than or equal to 3000
+        while canister_data.watch_history.len() > 3000 {
+            canister_data.watch_history.pop_first();
+        }
     });
 
     Ok("Success".into())

src/canister/platform_orchestrator/can.did

@@ -124,6 +124,7 @@ service : (PlatformOrchestratorInitArgs) -> {
   remove_principal_from_global_admins : (principal) -> ();
   remove_subnet_orchestrators_from_available_list : (principal) -> (Result);
   report_subnet_upgrade_status : (UpgradeStatus) -> (Result_1);
+  reset_canisters_ml_feed_cache : () -> (Result);
   set_reserved_cycle_limit_for_subnet_orchestrator : (principal, nat) -> (
       Result_1,
     );

src/canister/platform_orchestrator/src/api/canister_management/mod.rs

@@ -20,6 +20,7 @@ pub mod register_new_subnet_orhestrator;
 mod reinstall_yral_post_cache_canister;
 pub mod remove_subnet_orchestrator_from_available_list;
 pub mod report_subnet_upgrade_status;
+pub mod reset_canisters_ml_feed_cache;
 pub mod set_reserved_cycle_limit_for_subnet_orchestrator;
 pub mod start_subnet_orchestrator_canister;
 mod stop_upgrades_for_individual_user_canisters;

src/canister/platform_orchestrator/src/api/canister_management/reset_canisters_ml_feed_cache.rs

@@ -0,0 +1,27 @@
+use ic_cdk::{api::call::CallResult, call};
+use ic_cdk_macros::update;
+
+use crate::{guard::is_caller::is_caller_global_admin_or_controller, CANISTER_DATA};
+
+#[update(guard = "is_caller_global_admin_or_controller")]
+async fn reset_canisters_ml_feed_cache() -> Result<String, String> {
+    let subnet_orchestrator_list = CANISTER_DATA
+        .with_borrow(|canister_data| canister_data.all_subnet_orchestrator_canisters_list.clone());
+
+    for subnet_orchestrator in subnet_orchestrator_list {
+        let result: CallResult<()> = call(
+            subnet_orchestrator,
+            "reset_user_canisters_ml_feed_cache",
+            (),
+        )
+        .await;
+        result.map_err(|e| {
+            format!(
+                "failed to call reset_user_canisters_ml_feed_cache for {} {}",
+                subnet_orchestrator, e.1
+            )
+        })?;
+    }
+
+    Ok("Success".into())
+}

src/canister/user_index/can.did

@@ -178,6 +178,7 @@ service : (UserIndexInitArgs) -> {
   recharge_individual_user_canister : () -> (Result_4);
   reclaim_cycles_from_individual_canisters : () -> ();
   request_cycles : (nat) -> (Result_4);
+  reset_user_canisters_ml_feed_cache : () -> (text);
   reset_user_individual_canisters : (vec principal) -> (Result_2);
   return_cycles_to_platform_orchestrator_canister : () -> (Result_2);
   set_permission_to_upgrade_individual_canisters : (bool) -> (text);

src/canister/user_index/src/api/canister_management/mod.rs

@@ -43,6 +43,7 @@ pub mod provision_empty_canisters;
 pub mod recharge_individual_user_canister;
 pub mod recycle_canisters;
 pub mod request_cycles;
+pub mod reset_user_canister_ml_feed_cache;
 pub mod start_upgrades_for_individual_canisters;
 pub mod update_canisters_access_time;
 pub mod update_user_canister_restart_timers;

src/canister/user_index/src/api/canister_management/reset_user_canister_ml_feed_cache.rs

@@ -0,0 +1,30 @@
+use futures::StreamExt;
+use ic_cdk::api::call::CallResult;
+use ic_cdk_macros::update;
+
+use crate::CANISTER_DATA;
+use shared_utils::common::utils::permissions::is_caller_controller;
+
+#[update(guard = "is_caller_controller")]
+async fn reset_user_canisters_ml_feed_cache() -> String {
+    ic_cdk::spawn(reset_user_canisters_ml_feed_cache_impl());
+    "Success".to_string()
+}
+
+async fn reset_user_canisters_ml_feed_cache_impl() {
+    let canisters = CANISTER_DATA.with_borrow_mut(|canister_data| {
+        canister_data
+            .user_principal_id_to_canister_id_map
+            .values()
+            .cloned()
+            .collect::<Vec<_>>()
+    });
+
+    let futures = canisters.iter().map(|canister_id| async {
+        let _: CallResult<()> = ic_cdk::call(*canister_id, "reset_ml_feed_cache", ()).await;
+    });
+
+    let stream = futures::stream::iter(futures).boxed().buffer_unordered(25);
+
+    let _ = stream.collect::<Vec<()>>().await;
+}

src/lib/integration_tests/tests/platform_orchestrator/main.rs

@@ -3,4 +3,5 @@ pub mod provision_empty_canisters_in_a_subnet_test;
 pub mod provision_subnet_orchestrator_test;
 pub mod recharge_subnet_orchestrator_test;
 pub mod register_and_deregister_new_subnet_orchestrator_test;
+pub mod reset_ml_feed_cache_test;
 pub mod update_canisters_last_access_time_test;