feat: propagate origin and trigger to cdc and changes

src/merger/mod.rs

@@ -946,7 +946,25 @@ impl Merger {
       Value::Object(old_vals)
     };
 
+    let origin = match self.db.auth_origin() {
+      Ok(Some(orig)) => orig,
+      _ => serde_json::json!({
+        "kind": "user",
+        "user_id": user_id
+      }),
+    };
+
+    let trigger = match self.db.punc_trigger() {
+      Ok(Some(trig)) => trig,
+      _ => "merge_entity".to_string(),
+    };
+
+    let entity_type_name = type_name.as_str().unwrap_or(&type_obj.name);
+
     let mut notification = serde_json::Map::new();
+    notification.insert("type".to_string(), Value::String(entity_type_name.to_string()));
+    notification.insert("trigger".to_string(), Value::String(trigger));
+    notification.insert("origin".to_string(), origin.clone());
     notification.insert("complete".to_string(), Value::Object(complete));
     notification.insert("new".to_string(), new_val_obj.clone());
 
@@ -961,14 +979,16 @@ impl Merger {
     let mut notify_sql = None;
     if type_obj.historical && change_kind != "replace" {
       let change_sql = format!(
-        "INSERT INTO agreego.change (\"old\", \"new\", entity_id, id, kind, modified_at, modified_by) VALUES ({}, {}, {}, {}, {}, {}, {})",
+        "INSERT INTO agreego.change (\"old\", \"new\", entity_id, id, kind, modified_at, modified_by, origin, entity_type) VALUES ({}, {}, {}, {}, {}, {}, {}, {}, {})",
         Self::quote_literal(&old_val_obj),
         Self::quote_literal(&new_val_obj),
         Self::quote_literal(id_str),
         Self::quote_literal(&Value::String(uuid::Uuid::new_v4().to_string())),
         Self::quote_literal(&Value::String(change_kind.to_string())),
         Self::quote_literal(&Value::String(timestamp.to_string())),
-        Self::quote_literal(&Value::String(user_id.to_string()))
+        Self::quote_literal(&Value::String(user_id.to_string())),
+        Self::quote_literal(&origin),
+        Self::quote_literal(&Value::String(entity_type_name.to_string()))
       );
 
       self.db.execute(&change_sql, None)?;