version: 1.0.99

GEMINI.md

@@ -23,6 +23,12 @@ To support high-throughput operations while allowing for runtime updates (e.g.,
 3. **Immutable AST Caching**: The `Validator` struct immutably owns the `Database` registry. Schemas themselves are frozen structurally, but utilize `OnceLock` interior mutability during the Compilation Phase to permanently cache resolved `$ref` inheritances, properties, and `compiled_edges` directly onto their AST nodes. This guarantees strict `O(1)` relationship and property validation execution at runtime without locking or recursive DB polling.
 4. **Lock-Free Reads**: Incoming operations acquire a read lock just long enough to clone the `Arc` inside an `RwLock<Option<Arc<Validator>>>`, ensuring zero blocking during schema updates.
 
+### Relational Edge Resolution
+When compiling nested object graphs or arrays, the JSPG engine must dynamically infer which Postgres Foreign Key constraint correctly bridges the parent to the nested schema. It utilizes a strict 3-step hierarchical resolution:
+1.  **Direct Prefix Match**: If an explicitly prefixed Foreign Key (e.g. `fk_invoice_counterparty_entity` -> `prefix: "counterparty"`) matches the exact name of the requested schema property (e.g. `{"counterparty": {...}}`), it is instantly selected.
+2.  **Base Edge Fallback (1:M)**: If no explicit prefix directly matches the property name, the compiler filters for explicitly one remaining relation with a `null` prefix (e.g. `fk_invoice_line_invoice` -> `prefix: null`). A `null` prefix mathematically denotes the standard structural parent-child ownership edge (bypassing any M:M ambiguity) and is safely picked over explicit (but unmatched) property edges.
+3.  **Ambiguity Elimination (M:M)**: If multiple explicitly prefixed relations remain (which happens by design in Many-to-Many junction tables like `contact` utilizing `fk_relationship_source` and `fk_relationship_target`), the compiler uses a process of elimination. It checks which of the prefix names the child schema *natively consumes* as an outbound property (e.g. `contact` defines `{ "target": ... }`). It considers that prefix "used up" and mathematically deduces the *remaining* explicitly prefixed relation (`"source"`) must be the inbound link from the parent.
+
 ### Global API Reference
 These functions operate on the global `GLOBAL_JSPG` engine instance and provide administrative boundaries:
 

fixtures/merger.json

@@ -19,7 +19,7 @@
                 {
                     "id": "22222222-2222-2222-2222-222222222222",
                     "type": "relation",
-                    "constraint": "fk_order_customer",
+                    "constraint": "fk_order_customer_person",
                     "source_type": "order",
                     "source_columns": [
                         "customer_id"
@@ -41,8 +41,7 @@
                     "destination_type": "order",
                     "destination_columns": [
                         "id"
-                    ],
-                    "prefix": "lines"
+                    ]
                 },
                 {
                     "id": "44444444-4444-4444-4444-444444444444",
@@ -1152,6 +1151,69 @@
                     ]
                 }
             },
+            {
+                "description": "Replace existing person with id and no changes (lookup)",
+                "action": "merge",
+                "data": {
+                    "id": "33333333-3333-3333-3333-333333333333",
+                    "type": "person",
+                    "first_name": "LookupFirst",
+                    "last_name": "LookupLast",
+                    "date_of_birth": "1990-01-01T00:00:00Z",
+                    "pronouns": "they/them"
+                },
+                "mocks": [
+                    {
+                        "id": "22222222-2222-2222-2222-222222222222",
+                        "type": "person",
+                        "first_name": "LookupFirst",
+                        "last_name": "LookupLast",
+                        "date_of_birth": "1990-01-01T00:00:00Z",
+                        "pronouns": "they/them",
+                        "contact_id": "old-contact"
+                    }
+                ],
+                "schema_id": "person",
+                "expect": {
+                    "success": true,
+                    "sql": [
+                        [
+                            "SELECT to_jsonb(t1.*) || to_jsonb(t2.*) || to_jsonb(t3.*) || to_jsonb(t4.*)",
+                            "FROM agreego.\"person\" t1",
+                            "LEFT JOIN agreego.\"user\" t2 ON t2.id = t1.id",
+                            "LEFT JOIN agreego.\"organization\" t3 ON t3.id = t1.id",
+                            "LEFT JOIN agreego.\"entity\" t4 ON t4.id = t1.id",
+                            "WHERE",
+                            "    t1.id = '33333333-3333-3333-3333-333333333333'",
+                            "    OR (",
+                            "        \"first_name\" = 'LookupFirst'",
+                            "        AND \"last_name\" = 'LookupLast'",
+                            "        AND \"date_of_birth\" = '1990-01-01T00:00:00Z'",
+                            "        AND \"pronouns\" = 'they/them'",
+                            "    )"
+                        ],
+                        [
+                            "SELECT pg_notify('entity', '{",
+                            "        \"complete\":{",
+                            "            \"contact_id\":\"old-contact\",",
+                            "            \"date_of_birth\":\"1990-01-01T00:00:00Z\",",
+                            "            \"first_name\":\"LookupFirst\",",
+                            "            \"id\":\"22222222-2222-2222-2222-222222222222\",",
+                            "            \"last_name\":\"LookupLast\",",
+                            "            \"modified_at\":\"2026-03-10T00:00:00Z\",",
+                            "            \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
+                            "            \"pronouns\":\"they/them\",",
+                            "            \"type\":\"person\"",
+                            "        },",
+                            "        \"new\":{",
+                            "            \"type\":\"person\"",
+                            "        },",
+                            "        \"replaces\":\"33333333-3333-3333-3333-333333333333\"",
+                            "    }')"
+                        ]
+                    ]
+                }
+            },
             {
                 "description": "Update existing person with id (no lookup)",
                 "action": "merge",

fixtures/queryer.json

@@ -27,7 +27,9 @@
             {
               "$id": "get_orders.response",
               "type": "array",
-              "items": { "$ref": "light.order" }
+              "items": {
+                "$ref": "light.order"
+              }
             }
           ]
         }
@@ -69,7 +71,7 @@
         {
           "id": "22222222-2222-2222-2222-222222222222",
           "type": "relation",
-          "constraint": "fk_order_customer",
+          "constraint": "fk_order_customer_person",
           "source_type": "order",
           "source_columns": [
             "customer_id"
@@ -80,6 +82,22 @@
           ],
           "prefix": "customer"
         },
+        {
+          "id": "22222222-2222-2222-2222-222222222227",
+          "type": "relation",
+          "constraint": "fk_order_counterparty_entity",
+          "source_type": "order",
+          "source_columns": [
+            "counterparty_id",
+            "counterparty_type"
+          ],
+          "destination_type": "entity",
+          "destination_columns": [
+            "id",
+            "type"
+          ],
+          "prefix": "counterparty"
+        },
         {
           "id": "33333333-3333-3333-3333-333333333333",
           "type": "relation",
@@ -91,8 +109,7 @@
           "destination_type": "order",
           "destination_columns": [
             "id"
-          ],
-          "prefix": "lines"
+          ]
         }
       ],
       "types": [
@@ -713,14 +730,18 @@
             "created_by",
             "modified_at",
             "modified_by",
-            "archived"
+            "archived",
+            "counterparty_id",
+            "counterparty_type"
           ],
           "grouped_fields": {
             "order": [
               "id",
               "type",
               "total",
-              "customer_id"
+              "customer_id",
+              "counterparty_id",
+              "counterparty_type"
             ],
             "entity": [
               "id",
@@ -748,7 +769,9 @@
             "created_at": "timestamptz",
             "created_by": "uuid",
             "modified_at": "timestamptz",
-            "modified_by": "uuid"
+            "modified_by": "uuid",
+            "counterparty_id": "uuid",
+            "counterparty_type": "text"
           },
           "variations": [
             "order"

src/database/schema.rs

@@ -622,7 +622,23 @@ pub(crate) fn resolve_relation<'a>(
     }
   }
 
+  if !resolved {
+    // 1. If there's EXACTLY ONE relation with a null prefix, it's the base structural edge. Pick it.
+    let mut null_prefix_ids = Vec::new();
+    for (i, rel) in matching_rels.iter().enumerate() {
+      if rel.prefix.is_none() {
+        null_prefix_ids.push(i);
+      }
+    }
+    if null_prefix_ids.len() == 1 {
+      chosen_idx = null_prefix_ids[0];
+      resolved = true;
+    }
+  }
+
   if !resolved && relative_keys.is_some() {
+    // 2. M:M Disambiguation: The child schema will explicitly define an outbound property
+    // matching one of the relational prefixes (e.g. "target"). We use the OTHER one (e.g. "source").
     let keys = relative_keys.unwrap();
     let mut missing_prefix_ids = Vec::new();
     for (i, rel) in matching_rels.iter().enumerate() {
@@ -634,6 +650,7 @@ pub(crate) fn resolve_relation<'a>(
     }
     if missing_prefix_ids.len() == 1 {
       chosen_idx = missing_prefix_ids[0];
+      // resolved = true;
     }
   }
 

src/merger/mod.rs

@@ -3,8 +3,8 @@
 
 pub mod cache;
 
-use crate::database::r#type::Type;
 use crate::database::Database;
+use crate::database::r#type::Type;
 use serde_json::Value;
 use std::sync::Arc;
 
@@ -25,19 +25,19 @@ impl Merger {
     let mut notifications_queue = Vec::new();
 
     let target_schema = match self.db.schemas.get(schema_id) {
-        Some(s) => Arc::new(s.clone()),
-        None => {
-            return crate::drop::Drop::with_errors(vec![crate::drop::Error {
-                code: "MERGE_FAILED".to_string(),
-                message: format!("Unknown schema_id: {}", schema_id),
-                details: crate::drop::ErrorDetails {
-                    path: "".to_string(),
-                    cause: None,
-                    context: Some(data),
-                    schema: None,
-                },
-            }]);
-        }
+      Some(s) => Arc::new(s.clone()),
+      None => {
+        return crate::drop::Drop::with_errors(vec![crate::drop::Error {
+          code: "MERGE_FAILED".to_string(),
+          message: format!("Unknown schema_id: {}", schema_id),
+          details: crate::drop::ErrorDetails {
+            path: "".to_string(),
+            cause: None,
+            context: Some(data),
+            schema: None,
+          },
+        }]);
+      }
     };
 
     let result = self.merge_internal(target_schema, data.clone(), &mut notifications_queue);
@@ -50,18 +50,24 @@ impl Merger {
         let mut final_cause = None;
 
         if let Ok(Value::Object(map)) = serde_json::from_str::<Value>(&msg) {
-          if let (Some(Value::String(e_msg)), Some(Value::String(e_code))) = (map.get("error"), map.get("code")) {
+          if let (Some(Value::String(e_msg)), Some(Value::String(e_code))) =
+            (map.get("error"), map.get("code"))
+          {
             final_message = e_msg.clone();
             final_code = e_code.clone();
             let mut cause_parts = Vec::new();
             if let Some(Value::String(d)) = map.get("detail") {
-              if !d.is_empty() { cause_parts.push(d.clone()); }
+              if !d.is_empty() {
+                cause_parts.push(d.clone());
+              }
             }
             if let Some(Value::String(h)) = map.get("hint") {
-              if !h.is_empty() { cause_parts.push(h.clone()); }
+              if !h.is_empty() {
+                cause_parts.push(h.clone());
+              }
             }
             if !cause_parts.is_empty() {
-                final_cause = Some(cause_parts.join("\n"));
+              final_cause = Some(cause_parts.join("\n"));
             }
           }
         }
@@ -144,11 +150,11 @@ impl Merger {
   ) -> Result<Value, String> {
     let mut item_schema = schema.clone();
     if let Some(crate::database::schema::SchemaTypeOrArray::Single(t)) = &schema.obj.type_ {
-        if t == "array" {
-            if let Some(items_def) = &schema.obj.items {
-                item_schema = items_def.clone();
-            }
+      if t == "array" {
+        if let Some(items_def) = &schema.obj.items {
+          item_schema = items_def.clone();
         }
+      }
     }
 
     let mut resolved_items = Vec::new();
@@ -178,48 +184,48 @@ impl Merger {
     };
 
     let compiled_props = match schema.obj.compiled_properties.get() {
-        Some(props) => props,
-        None => return Err("Schema has no compiled properties for merging".to_string()),
+      Some(props) => props,
+      None => return Err("Schema has no compiled properties for merging".to_string()),
     };
 
     let mut entity_fields = serde_json::Map::new();
     let mut entity_objects = std::collections::BTreeMap::new();
     let mut entity_arrays = std::collections::BTreeMap::new();
 
-    for (k, v) in obj {
+    for (k, v) in obj.clone() {
       // Always retain system and unmapped core fields natively implicitly mapped to the Postgres tables
       if k == "id" || k == "type" || k == "created" {
-          entity_fields.insert(k.clone(), v.clone());
-          continue;
+        entity_fields.insert(k.clone(), v.clone());
+        continue;
       }
 
       if let Some(prop_schema) = compiled_props.get(&k) {
-         let mut is_edge = false;
-         if let Some(edges) = schema.obj.compiled_edges.get() {
-             if edges.contains_key(&k) {
-                 is_edge = true;
-             }
-         }
-         
-         if is_edge {
-             let typeof_v = match &v {
-                Value::Object(_) => "object",
-                Value::Array(_) => "array",
-                _ => "field", // Malformed edge data?
-             };
-             if typeof_v == "object" {
-                 entity_objects.insert(k.clone(), (v.clone(), prop_schema.clone()));
-             } else if typeof_v == "array" {
-                 entity_arrays.insert(k.clone(), (v.clone(), prop_schema.clone()));
-             } else {
-                 entity_fields.insert(k.clone(), v.clone());
-             }
-         } else {
-             // Not an edge! It's a raw Postgres column (e.g., JSONB, text[])
-             entity_fields.insert(k.clone(), v.clone());
-         }
-      } else if type_def.fields.contains(&k) {
+        let mut is_edge = false;
+        if let Some(edges) = schema.obj.compiled_edges.get() {
+          if edges.contains_key(&k) {
+            is_edge = true;
+          }
+        }
+
+        if is_edge {
+          let typeof_v = match &v {
+            Value::Object(_) => "object",
+            Value::Array(_) => "array",
+            _ => "field", // Malformed edge data?
+          };
+          if typeof_v == "object" {
+            entity_objects.insert(k.clone(), (v.clone(), prop_schema.clone()));
+          } else if typeof_v == "array" {
+            entity_arrays.insert(k.clone(), (v.clone(), prop_schema.clone()));
+          } else {
+            entity_fields.insert(k.clone(), v.clone());
+          }
+        } else {
+          // Not an edge! It's a raw Postgres column (e.g., JSONB, text[])
           entity_fields.insert(k.clone(), v.clone());
+        }
+      } else if type_def.fields.contains(&k) {
+        entity_fields.insert(k.clone(), v.clone());
       }
     }
 
@@ -253,12 +259,16 @@ impl Merger {
       };
 
       if let Some(compiled_edges) = schema.obj.compiled_edges.get() {
-        println!("Compiled Edges keys for relation {}: {:?}", relation_name, compiled_edges.keys().collect::<Vec<_>>());
+        println!(
+          "Compiled Edges keys for relation {}: {:?}",
+          relation_name,
+          compiled_edges.keys().collect::<Vec<_>>()
+        );
         if let Some(edge) = compiled_edges.get(&relation_name) {
           println!("FOUND EDGE {} -> {:?}", relation_name, edge.constraint);
           if let Some(relation) = self.db.relations.get(&edge.constraint) {
             let parent_is_source = edge.forward;
-            
+
             if parent_is_source {
               if !relative.contains_key("organization_id") {
                 if let Some(org_id) = entity_fields.get("organization_id") {
@@ -266,15 +276,16 @@ impl Merger {
                 }
               }
 
-              let mut merged_relative = match self.merge_internal(rel_schema.clone(), Value::Object(relative), notifications)? {
+              let mut merged_relative = match self.merge_internal(
+                rel_schema.clone(),
+                Value::Object(relative),
+                notifications,
+              )? {
                 Value::Object(m) => m,
                 _ => continue,
               };
 
-              merged_relative.insert(
-                "type".to_string(),
-                Value::String(relative_type_name),
-              );
+              merged_relative.insert("type".to_string(), Value::String(relative_type_name));
 
               Self::apply_entity_relation(
                 &mut entity_fields,
@@ -297,7 +308,11 @@ impl Merger {
                 &entity_fields,
               );
 
-              let merged_relative = match self.merge_internal(rel_schema.clone(), Value::Object(relative), notifications)? {
+              let merged_relative = match self.merge_internal(
+                rel_schema.clone(),
+                Value::Object(relative),
+                notifications,
+              )? {
                 Value::Object(m) => m,
                 _ => continue,
               };
@@ -318,6 +333,20 @@ impl Merger {
       entity_replaces = replaces;
     }
 
+    #[cfg(not(test))]
+    if type_name == "contact" || type_name == "person" {
+      pgrx::notice!("=== DEBUG {} PAYLOAD ===", type_name);
+      pgrx::notice!("1. Incoming obj iteration: {:?}", obj);
+      pgrx::notice!("2. Final mapped entity_fields: {:?}", entity_fields);
+      pgrx::notice!(
+        "3. TypeDef fields check: {:?}",
+        type_def.fields.contains(&"source_id".to_string())
+      );
+      if !entity_fields.contains_key("source_id") {
+        pgrx::notice!("CRITICAL ERROR: source_id was dropped during mapping loop!");
+      }
+    }
+
     self.merge_entity_fields(
       entity_change_kind.as_deref().unwrap_or(""),
       &type_name,
@@ -360,19 +389,24 @@ impl Merger {
                 );
 
                 let mut item_schema = rel_schema.clone();
-                if let Some(crate::database::schema::SchemaTypeOrArray::Single(t)) = &rel_schema.obj.type_ {
-                    if t == "array" {
-                        if let Some(items_def) = &rel_schema.obj.items {
-                            item_schema = items_def.clone();
-                        }
+                if let Some(crate::database::schema::SchemaTypeOrArray::Single(t)) =
+                  &rel_schema.obj.type_
+                {
+                  if t == "array" {
+                    if let Some(items_def) = &rel_schema.obj.items {
+                      item_schema = items_def.clone();
                     }
+                  }
                 }
 
-                let merged_relative =
-                  match self.merge_internal(item_schema, Value::Object(relative_item), notifications)? {
-                    Value::Object(m) => m,
-                    _ => continue,
-                  };
+                let merged_relative = match self.merge_internal(
+                  item_schema,
+                  Value::Object(relative_item),
+                  notifications,
+                )? {
+                  Value::Object(m) => m,
+                  _ => continue,
+                };
 
                 relative_responses.push(Value::Object(merged_relative));
               }
@@ -433,8 +467,8 @@ impl Merger {
     // An anchor is STRICTLY a struct containing merely an `id` and `type`.
     // We aggressively bypass Database SPI `SELECT` fetches because there are no primitive
     // mutations to apply to the row. PostgreSQL inherently protects relationships via Foreign Keys downstream.
-    let is_anchor = entity_fields.len() == 2 
-      && entity_fields.contains_key("id") 
+    let is_anchor = entity_fields.len() == 2
+      && entity_fields.contains_key("id")
       && entity_fields.contains_key("type");
 
     let has_valid_id = entity_fields
@@ -450,13 +484,13 @@ impl Merger {
 
     let mut replaces_id = None;
     if let Some(ref fetched_row) = entity_fetched {
-        let provided_id = entity_fields.get("id").and_then(|v| v.as_str());
-        let fetched_id = fetched_row.get("id").and_then(|v| v.as_str());
-        if let (Some(pid), Some(fid)) = (provided_id, fetched_id) {
-            if !pid.is_empty() && pid != fid {
-                replaces_id = Some(pid.to_string());
-            }
+      let provided_id = entity_fields.get("id").and_then(|v| v.as_str());
+      let fetched_id = fetched_row.get("id").and_then(|v| v.as_str());
+      if let (Some(pid), Some(fid)) = (provided_id, fetched_id) {
+        if !pid.is_empty() && pid != fid {
+          replaces_id = Some(pid.to_string());
         }
+      }
     }
 
     let system_keys = vec![
@@ -508,7 +542,7 @@ impl Merger {
       );
 
       entity_fields = new_fields;
-    } else if changes.is_empty() {
+    } else if changes.is_empty() && replaces_id.is_none() {
       let mut new_fields = serde_json::Map::new();
       new_fields.insert(
         "id".to_string(),
@@ -524,6 +558,8 @@ impl Merger {
         .unwrap_or(false);
       entity_change_kind = if is_archived {
         Some("delete".to_string())
+      } else if changes.is_empty() && replaces_id.is_some() {
+        Some("replace".to_string())
       } else {
         Some("update".to_string())
       };
@@ -546,7 +582,12 @@ impl Merger {
       entity_fields = new_fields;
     }
 
-    Ok((entity_fields, entity_change_kind, entity_fetched, replaces_id))
+    Ok((
+      entity_fields,
+      entity_change_kind,
+      entity_fetched,
+      replaces_id,
+    ))
   }
 
   fn fetch_entity(
@@ -733,9 +774,7 @@ impl Merger {
           columns.join(", "),
           values.join(", ")
         );
-        self
-          .db
-          .execute(&sql, None)?;
+        self.db.execute(&sql, None)?;
       } else if change_kind == "update" || change_kind == "delete" {
         entity_pairs.remove("id");
         entity_pairs.remove("type");
@@ -767,9 +806,7 @@ impl Merger {
           set_clauses.join(", "),
           Self::quote_literal(&Value::String(id_str.to_string()))
         );
-        self
-          .db
-          .execute(&sql, None)?;
+        self.db.execute(&sql, None)?;
       }
     }
 
@@ -796,9 +833,9 @@ impl Merger {
 
     let mut old_vals = serde_json::Map::new();
     let mut new_vals = serde_json::Map::new();
-    let is_update = change_kind == "update" || change_kind == "delete";
+    let exists = change_kind == "update" || change_kind == "delete" || change_kind == "replace";
 
-    if !is_update {
+    if !exists {
       let system_keys = vec![
         "id".to_string(),
         "created_by".to_string(),
@@ -835,7 +872,7 @@ impl Merger {
     }
 
     let mut complete = entity_fields.clone();
-    if is_update {
+    if exists {
       if let Some(fetched) = entity_fetched {
         let mut temp = fetched.clone();
         for (k, v) in entity_fields {
@@ -855,17 +892,17 @@ impl Merger {
     let mut notification = serde_json::Map::new();
     notification.insert("complete".to_string(), Value::Object(complete));
     notification.insert("new".to_string(), new_val_obj.clone());
-    
+
     if old_val_obj != Value::Null {
-        notification.insert("old".to_string(), old_val_obj.clone());
+      notification.insert("old".to_string(), old_val_obj.clone());
     }
-    
+
     if let Some(rep) = replaces_id {
-        notification.insert("replaces".to_string(), Value::String(rep.to_string()));
+      notification.insert("replaces".to_string(), Value::String(rep.to_string()));
     }
 
     let mut notify_sql = None;
-    if type_obj.historical {
+    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 ({}, {}, {}, {}, {}, {}, {})",
         Self::quote_literal(&old_val_obj),

src/tests/fixtures.rs

@@ -8602,3 +8602,9 @@ fn test_merger_0_11() {
     let path = format!("{}/fixtures/merger.json", env!("CARGO_MANIFEST_DIR"));
     crate::tests::runner::run_test_case(&path, 0, 11).unwrap();
 }
+
+#[test]
+fn test_merger_0_12() {
+    let path = format!("{}/fixtures/merger.json", env!("CARGO_MANIFEST_DIR"));
+    crate::tests::runner::run_test_case(&path, 0, 12).unwrap();
+}

version

@@ -1 +1 @@
-1.0.95
+1.0.99