fix(trigger): fire AFTER UPDATE on UPSERT and INSERT ON CONFLICT overwrites

Previously UPSERT always fired AFTER INSERT regardless of whether the
target row already existed, causing AFTER UPDATE subscribers (CDC,
materialized views, application triggers) to silently miss overwrite
events.

Add needs_existence_probe to DmlWriteInfo. When set, the pgwire
routing layer fetches the old row before dispatch and reclassifies the
DmlEvent to Insert or Update based on probe result before firing any
BEFORE or AFTER triggers.

The DSL UPSERT handler receives the same treatment: it probes the
primary key and calls fire_sync_after_update_triggers (new helper in
insert_parse.rs) on an overwrite, falling back to fire_sync_after_triggers
on a fresh key.

Author: farhan-syah

nodedb/src/control/server/pgwire/ddl/collection/insert_parse.rs

@@ -398,6 +398,37 @@ pub(super) async fn fire_sync_after_triggers(
     None
 }
 
+/// Fire SYNC AFTER UPDATE triggers, returning an error response on failure.
+///
+/// Used by the UPSERT DSL when the probe finds a pre-existing row —
+/// without this, AFTER UPDATE subscribers would silently miss overwrite
+/// events because the UPSERT handler historically fired only AFTER INSERT.
+pub(super) async fn fire_sync_after_update_triggers(
+    state: &SharedState,
+    identity: &AuthenticatedIdentity,
+    tenant_id: nodedb_types::TenantId,
+    coll_name: &str,
+    old_fields: &std::collections::HashMap<String, nodedb_types::Value>,
+    new_fields: &std::collections::HashMap<String, nodedb_types::Value>,
+) -> Option<PgWireResult<Vec<Response>>> {
+    use crate::control::security::catalog::trigger_types::TriggerExecutionMode;
+    if let Err(e) = crate::control::trigger::fire_after::fire_after_update(
+        state,
+        identity,
+        tenant_id,
+        coll_name,
+        old_fields,
+        new_fields,
+        0,
+        Some(TriggerExecutionMode::Sync),
+    )
+    .await
+    {
+        return Some(Err(sqlstate_error("XX000", &format!("trigger error: {e}"))));
+    }
+    None
+}
+
 /// Fire INSTEAD OF INSERT triggers, returning the result.
 pub(super) async fn fire_instead_triggers(
     state: &SharedState,

nodedb/src/control/server/pgwire/ddl/collection/upsert.rs

@@ -7,7 +7,8 @@ use crate::control::security::identity::AuthenticatedIdentity;
 use crate::control::state::SharedState;
 
 use super::insert_parse::{
-    fire_before_triggers, fire_instead_triggers, fire_sync_after_triggers, parse_write_statement,
+    fire_before_triggers, fire_instead_triggers, fire_sync_after_triggers,
+    fire_sync_after_update_triggers, parse_write_statement,
 };
 
 /// UPSERT INTO <collection> (col1, col2, ...) VALUES (val1, val2, ...)
@@ -88,6 +89,35 @@ pub async fn upsert_document(
         }
     }
 
+    // Probe for an existing row BEFORE dispatch so the correct AFTER
+    // trigger class fires: UPSERT onto an existing primary key is an
+    // UPDATE from every downstream consumer's perspective (AFTER UPDATE
+    // triggers, CDC, materialized views). Probing ahead of dispatch is
+    // safe because the document primary key acts as the upsert key and
+    // the probe + dispatch + AFTER-fire all run serially on this
+    // connection.
+    let pk_for_probe = fields
+        .get("id")
+        .or_else(|| fields.get("document_id"))
+        .or_else(|| fields.get("key"))
+        .map(|v| match v {
+            nodedb_types::Value::String(s) => s.clone(),
+            nodedb_types::Value::Integer(i) => i.to_string(),
+            other => format!("{other:?}"),
+        });
+    let old_fields = if let Some(ref pk) = pk_for_probe {
+        let row = crate::control::trigger::dml_hook::fetch_old_row(
+            state,
+            tenant_id,
+            &parsed.coll_name,
+            pk,
+        )
+        .await;
+        if row.is_empty() { None } else { Some(row) }
+    } else {
+        None
+    };
+
     // Build SQL and route through nodedb-sql → EngineRules → sql_plan_convert.
     let upsert_sql = super::insert_parse::fields_to_upsert_sql(&parsed.coll_name, &fields);
     if let Err(e) =
@@ -96,8 +126,25 @@ pub async fn upsert_document(
         return Some(Err(e));
     }
 
-    // Fire SYNC AFTER INSERT triggers.
-    if let Some(err) =
+    // Fire the AFTER trigger family that matches the actual mutation:
+    // AFTER UPDATE when a prior row existed, AFTER INSERT otherwise.
+    // Firing AFTER INSERT unconditionally would silently skip AFTER
+    // UPDATE subscribers on overwrites — the exact bug this routing
+    // fixes.
+    if let Some(ref old) = old_fields {
+        if let Some(err) = fire_sync_after_update_triggers(
+            state,
+            identity,
+            tenant_id,
+            &parsed.coll_name,
+            old,
+            &fields,
+        )
+        .await
+        {
+            return Some(err);
+        }
+    } else if let Some(err) =
         fire_sync_after_triggers(state, identity, tenant_id, &parsed.coll_name, &fields).await
     {
         return Some(err);

nodedb/src/control/server/pgwire/handler/routing/mod.rs

@@ -340,16 +340,21 @@ impl NodeDbPgHandler {
             let plan_for_response = task.plan.clone();
 
             // --- Trigger interception for DML writes ---
-            let dml_info = crate::control::trigger::dml_hook::classify_dml_write(&task.plan);
+            let mut dml_info = crate::control::trigger::dml_hook::classify_dml_write(&task.plan);
 
             // Fetch OLD row and fire BEFORE/INSTEAD OF triggers if applicable.
+            // UPSERT sets `needs_existence_probe` so the probe decides whether
+            // AFTER INSERT or AFTER UPDATE triggers fire — post-dispatch routing
+            // branches on `info.event`, which we override here based on the
+            // probe result before the BEFORE / AFTER hooks run.
             let old_row = if let Some(ref info) = dml_info
                 && info.document_id.is_some()
-                && matches!(
+                && (matches!(
                     info.event,
                     crate::control::trigger::DmlEvent::Update
                         | crate::control::trigger::DmlEvent::Delete
-                ) {
+                ) || info.needs_existence_probe)
+            {
                 let doc_id = info.document_id.as_deref().unwrap_or("");
                 let row = crate::control::trigger::dml_hook::fetch_old_row(
                     &self.state,
@@ -363,6 +368,20 @@ impl NodeDbPgHandler {
                 None
             };
 
+            // Probe-driven reclassification: UPSERT onto an existing row is an
+            // UPDATE for trigger purposes; onto a fresh key it's an INSERT.
+            // `needs_existence_probe` is the signal that `event` was a
+            // placeholder and must be refined from the probe result.
+            if let Some(ref mut info) = dml_info
+                && info.needs_existence_probe
+            {
+                info.event = if old_row.is_some() {
+                    crate::control::trigger::DmlEvent::Update
+                } else {
+                    crate::control::trigger::DmlEvent::Insert
+                };
+            }
+
             if let Some(ref info) = dml_info {
                 use crate::control::trigger::dml_hook::PreDispatchResult;
                 match crate::control::trigger::dml_hook::fire_pre_dispatch_triggers(

nodedb/src/control/trigger/dml_hook.rs

@@ -35,9 +35,22 @@ pub struct DmlWriteInfo {
     /// Document ID (for point operations). None for bulk operations.
     pub document_id: Option<String>,
     /// DML event type.
+    ///
+    /// For UPSERT the initial value is a best guess — the true event is
+    /// not known until the routing layer probes the pre-write row via
+    /// `fetch_old_row`. When `needs_existence_probe` is set, routing
+    /// overrides this field based on probe results before firing
+    /// post-dispatch triggers.
     pub event: DmlEvent,
     /// NEW row fields extracted from the write plan. None for DELETE.
     pub new_fields: Option<HashMap<String, nodedb_types::Value>>,
+    /// True when the operation's real event type depends on whether the
+    /// target row already exists (currently: UPSERT / INSERT ... ON
+    /// CONFLICT). Routing uses this flag to force a pre-dispatch
+    /// existence probe so the correct AFTER INSERT vs AFTER UPDATE
+    /// triggers fire — otherwise an UPSERT onto an existing row would
+    /// silently fire AFTER INSERT, which is the wrong trigger class.
+    pub needs_existence_probe: bool,
 }
 
 /// Attempt to classify a PhysicalPlan as a document DML write.
@@ -74,6 +87,7 @@ fn classify_document_op(op: &DocumentOp) -> Option<DmlWriteInfo> {
                 document_id: Some(document_id.clone()),
                 event: DmlEvent::Insert,
                 new_fields: Some(new_fields),
+                needs_existence_probe: false,
             })
         }
         DocumentOp::Upsert {
@@ -82,12 +96,17 @@ fn classify_document_op(op: &DocumentOp) -> Option<DmlWriteInfo> {
             value,
             ..
         } => {
+            // UPSERT's event type depends on whether the primary key
+            // already exists — routing must probe before firing
+            // post-dispatch SYNC triggers. `event` starts at Insert as a
+            // harmless default; the probe result overrides it.
             let new_fields = deserialize_value_to_fields(value);
             Some(DmlWriteInfo {
                 collection: collection.clone(),
                 document_id: Some(document_id.clone()),
-                event: DmlEvent::Insert, // Upsert treated as INSERT for trigger purposes
+                event: DmlEvent::Insert,
                 new_fields: Some(new_fields),
+                needs_existence_probe: true,
             })
         }
         DocumentOp::PointDelete {
@@ -98,6 +117,7 @@ fn classify_document_op(op: &DocumentOp) -> Option<DmlWriteInfo> {
             document_id: Some(document_id.clone()),
             event: DmlEvent::Delete,
             new_fields: None,
+            needs_existence_probe: false,
         }),
         DocumentOp::PointUpdate {
             collection,
@@ -108,30 +128,35 @@ fn classify_document_op(op: &DocumentOp) -> Option<DmlWriteInfo> {
             document_id: Some(document_id.clone()),
             event: DmlEvent::Update,
             new_fields: None, // NEW fields computed after applying updates to OLD
+            needs_existence_probe: false,
         }),
         DocumentOp::BatchInsert { collection, .. } => Some(DmlWriteInfo {
             collection: collection.clone(),
             document_id: None,
             event: DmlEvent::Insert,
             new_fields: None, // Batch — individual rows not available here
+            needs_existence_probe: false,
         }),
         DocumentOp::BulkUpdate { collection, .. } => Some(DmlWriteInfo {
             collection: collection.clone(),
             document_id: None,
             event: DmlEvent::Update,
             new_fields: None,
+            needs_existence_probe: false,
         }),
         DocumentOp::BulkDelete { collection, .. } => Some(DmlWriteInfo {
             collection: collection.clone(),
             document_id: None,
             event: DmlEvent::Delete,
             new_fields: None,
+            needs_existence_probe: false,
         }),
         DocumentOp::Truncate { collection, .. } => Some(DmlWriteInfo {
             collection: collection.clone(),
             document_id: None,
             event: DmlEvent::Delete,
             new_fields: None,
+            needs_existence_probe: false,
         }),
         DocumentOp::InsertSelect {
             target_collection, ..
@@ -140,6 +165,7 @@ fn classify_document_op(op: &DocumentOp) -> Option<DmlWriteInfo> {
             document_id: None,
             event: DmlEvent::Insert,
             new_fields: None,
+            needs_existence_probe: false,
         }),
         // Not a write operation.
         DocumentOp::PointGet { .. }