Add v2 endpoint to support old and new version event producers (#58)

rs/canister/api/src/lib.rs

@@ -23,6 +23,17 @@ pub struct IdempotentEvent {
     pub payload: Vec<u8>,
 }
 
+#[derive(CandidType, Serialize, Deserialize, Clone, Debug)]
+pub struct IdempotentEventPrevious {
+    pub idempotency_key: u128,
+    pub name: String,
+    pub timestamp: TimestampMillis,
+    pub user: Option<String>,
+    pub source: Option<String>,
+    #[serde(with = "serde_bytes")]
+    pub payload: Vec<u8>,
+}
+
 #[derive(CandidType, Serialize, Deserialize, Clone, Debug)]
 pub struct IndexedEvent {
     pub index: u64,

rs/canister/api/src/updates/mod.rs

@@ -1,3 +1,5 @@
 mod push_events;
+mod push_events_v2;
 
 pub use push_events::*;
+pub use push_events_v2::*;

rs/canister/api/src/updates/push_events.rs

@@ -1,8 +1,8 @@
-use crate::IdempotentEvent;
+use crate::IdempotentEventPrevious;
 use candid::{CandidType, Deserialize};
 use serde::Serialize;
 
 #[derive(CandidType, Serialize, Deserialize, Clone, Debug)]
-pub struct PushEventsArgs {
-    pub events: Vec<IdempotentEvent>,
+pub struct PushEventsArgsPrevious {
+    pub events: Vec<IdempotentEventPrevious>,
 }

rs/canister/api/src/updates/push_events_v2.rs

@@ -0,0 +1,8 @@
+use crate::IdempotentEvent;
+use candid::{CandidType, Deserialize};
+use serde::Serialize;
+
+#[derive(CandidType, Serialize, Deserialize, Clone, Debug)]
+pub struct PushEventsArgs {
+    pub events: Vec<IdempotentEvent>,
+}

rs/canister/impl/src/updates/push_events.rs

@@ -1,10 +1,31 @@
 use crate::guards::caller_can_push_events;
 use crate::{env, state};
-use event_store_canister::PushEventsArgs;
+use event_store_canister::{Anonymizable, IdempotentEvent, PushEventsArgs, PushEventsArgsPrevious};
 use ic_cdk::update;
 
 #[update(guard = "caller_can_push_events")]
-fn push_events(args: PushEventsArgs) {
+fn push_events(args: PushEventsArgsPrevious) {
+    let now = env::time();
+
+    state::mutate(|s| {
+        for event in args.events {
+            s.push_event(
+                IdempotentEvent {
+                    idempotency_key: event.idempotency_key,
+                    name: event.name,
+                    timestamp: event.timestamp,
+                    user: event.user.map(Anonymizable::Public),
+                    source: event.source.map(Anonymizable::Public),
+                    payload: event.payload,
+                },
+                now,
+            );
+        }
+    });
+}
+
+#[update(guard = "caller_can_push_events")]
+fn push_events_v2(args: PushEventsArgs) {
     let now = env::time();
 
     state::mutate(|s| {

rs/integration_tests/src/client.rs

@@ -18,7 +18,7 @@ pub fn push_events(
     canister_id: Principal,
     args: &PushEventsArgs,
 ) {
-    execute_update_no_response(env, sender, canister_id, "push_events", args)
+    execute_update_no_response(env, sender, canister_id, "push_events_v2", args)
 }
 
 fn execute_query<P: CandidType, R: CandidType + DeserializeOwned>(

rs/producer/agent_runtime/src/lib.rs

@@ -73,7 +73,7 @@ async fn flush_async<F: FnOnce(FlushOutcome) + Send + 'static>(
     on_complete: F,
 ) {
     if agent
-        .update(&canister_id, "push_events".to_string())
+        .update(&canister_id, "push_events_v2".to_string())
         .with_arg(candid::encode_one(PushEventsArgs { events }).unwrap())
         .call_and_wait()
         .await

rs/producer/cdk_runtime/src/lib.rs

@@ -55,7 +55,7 @@ async fn flush_async<F: FnOnce(FlushOutcome)>(
 ) {
     let events_len = events.len();
     if let Err(error) =
-        ic_cdk::call::<_, ()>(canister_id, "push_events", (PushEventsArgs { events },)).await
+        ic_cdk::call::<_, ()>(canister_id, "push_events_v2", (PushEventsArgs { events },)).await
     {
         on_complete(FLUSH_OUTCOME_FAILED_SHOULD_RETRY);
         error!(%canister_id, events = events_len, ?error, "Failed to call 'push_events'");