diff --git a/GEMINI.md b/GEMINI.md index 3bf2adb..77b8839 100644 --- a/GEMINI.md +++ b/GEMINI.md @@ -314,13 +314,13 @@ The Merger provides an automated, high-performance graph synchronization engine. ### Core Features -* **Caching Strategy**: The Merger leverages the native `compiled_edges` permanently cached onto the Schema AST via `OnceLock` to instantly resolve Foreign Key mapping graphs natively in absolute `O(1)` time. It additionally utilizes the concurrent `GLOBAL_JSPG` application memory (`DashMap`) to cache statically constructed SQL `SELECT` strings used during deduplication (`lk_`) and difference tracking calculations. +* **Caching Strategy**: The Merger leverages the native `compiled_edges` permanently cached onto the Schema AST via `OnceLock` to instantly resolve Foreign Key mapping graphs natively in absolute `O(1)` time. It additionally utilizes the concurrent `GLOBAL_JSPG` application memory (`DashMap`) to cache statically constructed SQL `SELECT` strings used during deduplication (`lookup_fields`) and difference tracking calculations. * **Deep Graph Merging**: The Merger walks arbitrary levels of deeply nested JSON schemas (e.g. tracking an `order`, its `customer`, and an array of its `lines`). It intelligently discovers the correct parent-to-child or child-to-parent Foreign Keys stored in the registry and automatically maps the UUIDs across the relationships during UPSERT. * **Prefix Foreign Key Matching**: Handles scenario where multiple relations point to the same table by using database Foreign Key constraint prefixes (`fk_`). For example, if a schema has `shipping_address` and `billing_address`, the merger resolves against `fk_shipping_address_entity` vs `fk_billing_address_entity` automatically to correctly route object properties. -* **Dynamic Deduplication & Lookups**: If a nested object is provided without an `id`, the Merger utilizes Postgres `lk_` index constraints defined in the schema registry (e.g. `lk_person` mapped to `first_name` and `last_name`). It dynamically queries these unique matching constraints to discover the correct UUID to perform an UPDATE, preventing data duplication. +* **Dynamic Deduplication & Lookups**: If a nested object is provided without an `id`, the Merger utilizes custom `lookup_fields` declared directly in the schema registry JSON comments. It validates at setup compile-time that a corresponding unique index exists in PostgreSQL for these fields. When merging, it dynamically builds query predicates for any satisfied `lookup_fields` sets in the entity's type hierarchy (checking child-to-parent hierarchies order-independently and combining satisfied keys with `UNION` queries) to discover the correct UUID to perform an UPDATE, preventing data duplication. * **Hierarchical Table Inheritance**: The Punc system uses distributed table inheritance (e.g. `person` inherits `user` inherits `organization` inherits `entity`). The Merger splits the incoming JSON payload and performs atomic row updates across *all* relevant tables in the lineage map. * **The Archive Paradigm**: Data is never deleted in the Punc system. The Merger securely enforces referential integrity by toggling the `archived` Boolean flag on the base `entity` table rather than issuing SQL `DELETE` commands. -* **Change Tracking & Reactivity**: The Merger diffs the incoming JSON against the existing database row (utilizing static, `DashMap`-cached `lk_` SELECT string templates). Every detected change is recorded into the `agreego.change` audit table, tracking the user mapping. It then natively uses `pg_notify` to broadcast a completely flat row-level diff out to the Go WebSocket server for O(1) routing. +* **Change Tracking & Reactivity**: The Merger diffs the incoming JSON against the existing database row (utilizing static, `DashMap`-cached `lookup` SELECT string templates). Every detected change is recorded into the `agreego.change` audit table, tracking the user mapping. It then natively uses `pg_notify` to broadcast a completely flat row-level diff out to the Go WebSocket server for O(1) routing. * **Flat Structural Beats (Unidirectional Flow)**: The Merger purposefully DOES NOT trace or hydrate outbound Foreign Keys or nested parent structures during writes. It emits completely flat, mathematically perfect structural deltas via `pg_notify` representing only the exact Postgres rows that changed. This guarantees the write-path remains O(1) lightning fast. It is the strict responsibility of the upstream Punc Framework (the Go `Speaker`) to intercept these flat beats, evaluate them against active Websocket Schema Topologies, and dynamically issue targeted `jspg_query` reads to hydrate the exact contextual subgraphs required by listening clients. * **Pre-Order Notification Traversal**: To support proper topological hydration on the upstream Go Framework, the Merger decouples the `pg_notify` execution from the physical database write execution. The engine collects structural changes and explicitly fires `pg_notify` SQL statements in strict **Pre-Order** (Parent -> Relations -> Children). This guarantees that WebSocket clients receive the parent entity `Beat` prior to any nested child entities, ensuring stable unidirectional data flows without hydration race conditions. * **Many-to-Many Graph Edge Management**: Operates seamlessly with the global `agreego.relationship` table, allowing the system to represent and merge arbitrary reified M:M relationships directionally between any two entities. diff --git a/fixtures/database.json b/fixtures/database.json index 8a7024f..b532814 100644 --- a/fixtures/database.json +++ b/fixtures/database.json @@ -911,9 +911,7 @@ "archived" ] }, - "lookup_fields": [ - "id" - ], + "lookup_fields": [], "historical": true, "relationship": false, "field_types": { @@ -1018,5 +1016,100 @@ } } ] + }, + { + "description": "Validation - lookup fields without matching index", + "database": { + "types": [ + { + "id": "22222222-2222-2222-2222-222222222222", + "type": "type", + "name": "user", + "module": "test", + "source": "test", + "hierarchy": ["user"], + "variations": ["user"], + "fields": ["id", "email"], + "lookup_fields": ["email"], + "schemas": { + "user": { + "type": "object", + "properties": { + "id": { "type": "string", "format": "uuid" }, + "email": { "type": "string", "format": "email" } + } + } + } + } + ] + }, + "tests": [ + { + "description": "Fails setup compilation with MISSING_LOOKUP_INDEX if unique index is missing", + "action": "compile", + "expect": { + "success": false, + "errors": [ + { + "code": "MISSING_LOOKUP_INDEX", + "values": { + "type": "user", + "lookup_fields": "[\"email\"]" + }, + "details": { + "path": "/types/user", + "schema": "user" + } + } + ] + } + } + ] + }, + { + "description": "Validation - lookup fields with matching index", + "database": { + "indexes": [ + { + "table": "user", + "columns": ["email"] + } + ], + "types": [ + { + "id": "22222222-2222-2222-2222-222222222222", + "type": "type", + "name": "user", + "module": "test", + "source": "test", + "hierarchy": ["user"], + "variations": ["user"], + "fields": ["id", "email"], + "lookup_fields": ["email"], + "schemas": { + "user": { + "type": "object", + "properties": { + "id": { "type": "string", "format": "uuid" }, + "email": { "type": "string", "format": "email" } + } + } + } + } + ] + }, + "tests": [ + { + "description": "Compiles successfully when unique index matches lookup_fields", + "action": "compile", + "expect": { + "success": true, + "schemas": { + "user": {}, + "user.filter": {} + } + } + } + ] } ] \ No newline at end of file diff --git a/fixtures/merger.json b/fixtures/merger.json index aabe9e1..9154a16 100644 --- a/fixtures/merger.json +++ b/fixtures/merger.json @@ -117,6 +117,16 @@ ] } ], + "indexes": [ + { + "table": "person", + "columns": ["first_name", "last_name", "date_of_birth", "pronouns"] + }, + { + "table": "user", + "columns": ["name"] + } + ], "types": [ { "name": "entity", @@ -296,7 +306,9 @@ "archived" ] }, - "lookup_fields": [], + "lookup_fields": [ + "name" + ], "historical": true, "notify": true, "relationship": false @@ -478,9 +490,7 @@ "organization_id" ] }, - "lookup_fields": [ - "id" - ], + "lookup_fields": [], "historical": true, "notify": true, "relationship": false @@ -1068,9 +1078,7 @@ "archived" ] }, - "lookup_fields": [ - "id" - ], + "lookup_fields": [], "historical": true, "relationship": false, "field_types": { @@ -1593,6 +1601,180 @@ ] } }, + { + "description": "Update existing person with id and multiple inherited lookup keys", + "action": "merge", + "data": { + "id": "33333333-3333-3333-3333-333333333333", + "type": "person", + "name": "LookupName", + "first_name": "LookupFirst", + "last_name": "LookupLast", + "date_of_birth": "1990-01-01T00:00:00Z", + "pronouns": "they/them", + "contact_id": "abc-contact" + }, + "mocks": [ + { + "id": "22222222-2222-2222-2222-222222222222", + "type": "person", + "name": "LookupName", + "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.\"entity\" t1", + "LEFT JOIN agreego.\"organization\" t2 ON t2.id = t1.id", + "LEFT JOIN agreego.\"user\" t3 ON t3.id = t1.id", + "LEFT JOIN agreego.\"person\" t4 ON t4.id = t1.id", + "WHERE", + " t1.id = '{{uuid:data.id}}'", + "UNION SELECT to_jsonb(t1.*) || to_jsonb(t2.*) || to_jsonb(t3.*) || to_jsonb(t4.*)", + "FROM agreego.\"entity\" t1", + "LEFT JOIN agreego.\"organization\" t2 ON t2.id = t1.id", + "LEFT JOIN agreego.\"user\" t3 ON t3.id = t1.id", + "LEFT JOIN agreego.\"person\" t4 ON t4.id = t1.id", + "WHERE", + " (\"first_name\" = 'LookupFirst'", + " AND \"last_name\" = 'LookupLast'", + " AND \"date_of_birth\" = '{{timestamp}}'", + " AND \"pronouns\" = 'they/them')", + "UNION SELECT to_jsonb(t1.*) || to_jsonb(t2.*) || to_jsonb(t3.*) || to_jsonb(t4.*)", + "FROM agreego.\"entity\" t1", + "LEFT JOIN agreego.\"organization\" t2 ON t2.id = t1.id", + "LEFT JOIN agreego.\"user\" t3 ON t3.id = t1.id", + "LEFT JOIN agreego.\"person\" t4 ON t4.id = t1.id", + "WHERE", + " (\"name\" = 'LookupName'))" + ], + [ + "INSERT INTO agreego.\"entity\" (", + " \"created_at\",", + " \"created_by\",", + " \"id\",", + " \"modified_at\",", + " \"modified_by\",", + " \"type\"", + ")", + "VALUES (", + " '{{timestamp}}',", + " '00000000-0000-0000-0000-000000000000',", + " '{{uuid:data.id}}',", + " '{{timestamp}}',", + " '00000000-0000-0000-0000-000000000000',", + " 'person'", + ")" + ], + [ + "INSERT INTO agreego.\"organization\" (", + " \"id\",", + " \"name\",", + " \"type\"", + ")", + "VALUES (", + " '{{uuid:data.id}}',", + " 'LookupName',", + " 'person'", + ")" + ], + [ + "INSERT INTO agreego.\"user\" (", + " \"id\",", + " \"type\"", + ")", + "VALUES (", + " '{{uuid:data.id}}',", + " 'person'", + ")" + ], + [ + "INSERT INTO agreego.\"person\" (", + " \"contact_id\",", + " \"date_of_birth\",", + " \"first_name\",", + " \"id\",", + " \"last_name\",", + " \"pronouns\",", + " \"type\"", + ")", + "VALUES (", + " 'abc-contact',", + " '{{timestamp}}',", + " 'LookupFirst',", + " '{{uuid:data.id}}',", + " 'LookupLast',", + " 'they/them',", + " 'person'", + ")" + ], + [ + "INSERT INTO agreego.change (", + " \"old\",", + " \"new\",", + " \"entity_id\",", + " \"id\",", + " \"kind\",", + " \"modified_at\",", + " \"modified_by\"", + ")", + "VALUES (", + " NULL,", + " '{", + " \"name\": \"LookupName\",", + " \"first_name\": \"LookupFirst\",", + " \"last_name\": \"LookupLast\",", + " \"date_of_birth\": \"{{timestamp}}\",", + " \"pronouns\": \"they/them\",", + " \"contact_id\": \"abc-contact\",", + " \"type\": \"person\"", + " }',", + " '{{uuid:data.id}}',", + " '{{uuid:generated_0}}',", + " 'create',", + " '{{timestamp}}',", + " '00000000-0000-0000-0000-000000000000'", + ")" + ], + [ + "(SELECT pg_notify('entity', '{", + " \"kind\": \"create\",", + " \"complete\": {", + " \"name\": \"LookupName\",", + " \"first_name\": \"LookupFirst\",", + " \"last_name\": \"LookupLast\",", + " \"date_of_birth\": \"{{timestamp}}\",", + " \"pronouns\": \"they/them\",", + " \"contact_id\": \"abc-contact\",", + " \"id\": \"{{uuid:data.id}}\",", + " \"type\": \"person\",", + " \"created_by\": \"00000000-0000-0000-0000-000000000000\",", + " \"created_at\": \"{{timestamp}}\",", + " \"modified_by\": \"00000000-0000-0000-0000-000000000000\",", + " \"modified_at\": \"{{timestamp}}\"", + " },", + " \"new\": {", + " \"name\": \"LookupName\",", + " \"first_name\": \"LookupFirst\",", + " \"last_name\": \"LookupLast\",", + " \"date_of_birth\": \"{{timestamp}}\",", + " \"pronouns\": \"they/them\",", + " \"contact_id\": \"abc-contact\",", + " \"type\": \"person\"", + " }", + "}'))" + ] + ] + } + }, { "description": "Replace existing person with id and no changes (lookup)", "action": "merge", @@ -2235,12 +2417,7 @@ "FROM agreego.\"entity\" t1", "LEFT JOIN agreego.\"order\" t2 ON t2.id = t1.id", "WHERE", - " t1.id = 'abc'", - "UNION SELECT to_jsonb(t1.*) || to_jsonb(t2.*)", - "FROM agreego.\"entity\" t1", - "LEFT JOIN agreego.\"order\" t2 ON t2.id = t1.id", - "WHERE", - " (\"id\" = 'abc'))" + " t1.id = 'abc')" ], [ "INSERT INTO agreego.\"entity\" (", @@ -3519,12 +3696,7 @@ "FROM agreego.\"entity\" t1", "LEFT JOIN agreego.\"invoice\" t2 ON t2.id = t1.id", "WHERE", - " t1.id = '{{uuid:data.id}}'", - "UNION SELECT to_jsonb(t1.*) || to_jsonb(t2.*)", - "FROM agreego.\"entity\" t1", - "LEFT JOIN agreego.\"invoice\" t2 ON t2.id = t1.id", - "WHERE", - " (\"id\" = '{{uuid:data.id}}'))" + " t1.id = '{{uuid:data.id}}')" ], [ "INSERT INTO agreego.\"entity\" (", diff --git a/fixtures/queryer.json b/fixtures/queryer.json index a906a42..ed22ca9 100644 --- a/fixtures/queryer.json +++ b/fixtures/queryer.json @@ -258,9 +258,7 @@ "name": "text", "created_at": "timestamptz" }, - "lookup_fields": [ - "id" - ], + "lookup_fields": [], "null_fields": [], "default_fields": [ "id", @@ -806,9 +804,7 @@ "archived" ] }, - "lookup_fields": [ - "id" - ], + "lookup_fields": [], "historical": true, "relationship": false, "field_types": { @@ -1106,9 +1102,7 @@ "archived" ] }, - "lookup_fields": [ - "id" - ], + "lookup_fields": [], "historical": true, "relationship": false, "field_types": { diff --git a/src/database/executors/mock.rs b/src/database/executors/mock.rs index 4a037b6..32ed60d 100644 --- a/src/database/executors/mock.rs +++ b/src/database/executors/mock.rs @@ -85,6 +85,10 @@ impl DatabaseExecutor for MockExecutor { Ok("2026-03-10T00:00:00Z".to_string()) } + fn is_mock(&self) -> bool { + true + } + #[cfg(test)] fn get_queries(&self) -> Vec { MOCK_STATE.with(|state| state.borrow().captured_queries.clone()) diff --git a/src/database/executors/mod.rs b/src/database/executors/mod.rs index 3923d13..5469776 100644 --- a/src/database/executors/mod.rs +++ b/src/database/executors/mod.rs @@ -20,6 +20,11 @@ pub trait DatabaseExecutor: Send + Sync { /// Returns the current transaction timestamp fn timestamp(&self) -> Result; + /// Returns whether this is a mock executor (bypassing pg_catalog index checks) + fn is_mock(&self) -> bool { + false + } + #[cfg(test)] fn get_queries(&self) -> Vec; diff --git a/src/database/mod.rs b/src/database/mod.rs index 64d889d..5b724ab 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -21,16 +21,22 @@ use executors::pgrx::SpiExecutor; #[cfg(test)] use executors::mock::MockExecutor; +use indexmap::IndexMap; use punc::Punc; use relation::Relation; use schema::Schema; use serde_json::Value; -use indexmap::IndexMap; use std::sync::Arc; use r#type::Type; use crate::drop::{Drop, Error, ErrorDetails}; +#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)] +pub struct IndexInfo { + pub table: String, + pub columns: Vec, +} + #[derive(serde::Serialize)] pub struct Database { pub enums: IndexMap, @@ -38,6 +44,8 @@ pub struct Database { pub puncs: IndexMap, pub relations: IndexMap, #[serde(skip)] + pub indexes: Vec, + #[serde(skip)] pub schemas: IndexMap>, #[serde(skip)] pub executor: Box, @@ -50,6 +58,7 @@ impl Database { types: IndexMap::new(), relations: IndexMap::new(), puncs: IndexMap::new(), + indexes: Vec::new(), schemas: IndexMap::new(), #[cfg(not(test))] executor: Box::new(SpiExecutor::new()), @@ -177,6 +186,25 @@ impl Database { } } } + + if let Some(serde_json::Value::Array(arr)) = map.remove("indexes") { + for item in arr { + match serde_json::from_value::(item) { + Ok(idx) => { + db.indexes.push(idx); + } + Err(e) => { + errors.push(Error { + code: "DATABASE_INDEX_PARSE_FAILED".to_string(), + values: Some(IndexMap::from([("reason".to_string(), e.to_string())])), + details: ErrorDetails { + ..Default::default() + }, + }); + } + } + } + } } db.compile(&mut errors); @@ -214,7 +242,6 @@ impl Database { self.executor.timestamp() } - pub fn compile(&mut self, errors: &mut Vec) { // Phase 1: Registration self.collect_schemas(errors); @@ -260,6 +287,43 @@ impl Database { .compile(self, root_id, id.clone(), errors); } } + + // Phase 5: Verify unique indexes for defined lookup fields + for (_, type_def) in &self.types { + if !type_def.lookup_fields.is_empty() { + let mut index_found = false; + for index in &self.indexes { + if index.table == type_def.name { + if index.columns.len() == type_def.lookup_fields.len() + && index + .columns + .iter() + .all(|c| type_def.lookup_fields.contains(c)) + { + index_found = true; + break; + } + } + } + if !index_found { + errors.push(Error { + code: "MISSING_LOOKUP_INDEX".to_string(), + values: Some(IndexMap::from([ + ("type".to_string(), type_def.name.clone()), + ( + "lookup_fields".to_string(), + format!("{:?}", type_def.lookup_fields), + ), + ])), + details: ErrorDetails { + path: Some(format!("/types/{}", type_def.name)), + schema: Some(type_def.name.clone()), + ..Default::default() + }, + }); + } + } + } } /// Synthesizes Composed Filter References for all table-backed boundaries. @@ -395,7 +459,6 @@ impl Database { } } - /// Inspects the Postgres pg_constraint relations catalog to securely identify /// the precise Foreign Key connecting a parent and child hierarchy path. pub fn resolve_relation<'a>( diff --git a/src/merger/mod.rs b/src/merger/mod.rs index 10cac99..027e72a 100644 --- a/src/merger/mod.rs +++ b/src/merger/mod.rs @@ -677,25 +677,32 @@ impl Merger { let id_val = entity_fields.get("id"); let entity_type_name = entity_type.name.as_str(); - let mut lookup_complete = false; - if !entity_type.lookup_fields.is_empty() { - lookup_complete = true; - for column in &entity_type.lookup_fields { - match entity_fields.get(column) { - Some(Value::Null) | None => { - lookup_complete = false; - break; + let mut lookup_satisfied_keys = Vec::new(); + for parent_type_name in entity_type.hierarchy.iter().rev() { + if let Some(parent_type) = self.db.types.get(parent_type_name) { + if !parent_type.lookup_fields.is_empty() { + let mut lookup_complete = true; + for column in &parent_type.lookup_fields { + match entity_fields.get(column) { + Some(Value::Null) | None => { + lookup_complete = false; + break; + } + Some(Value::String(s)) if s.is_empty() => { + lookup_complete = false; + break; + } + _ => {} + } } - Some(Value::String(s)) if s.is_empty() => { - lookup_complete = false; - break; + if lookup_complete { + lookup_satisfied_keys.push(&parent_type.lookup_fields); } - _ => {} } } } - if id_val.is_none() && !lookup_complete { + if id_val.is_none() && lookup_satisfied_keys.is_empty() { return Ok(None); } @@ -727,9 +734,9 @@ impl Merger { where_parts.push(format!("t1.id = {}", Self::quote_literal(id))); } - if lookup_complete { + for lookup_fields in lookup_satisfied_keys { let mut lookup_predicates = Vec::new(); - for column in &entity_type.lookup_fields { + for column in lookup_fields { let val = entity_fields.get(column).unwrap_or(&Value::Null); if column == "type" { lookup_predicates.push(format!("t1.\"{}\" = {}", column, Self::quote_literal(val))); diff --git a/src/tests/fixtures.rs b/src/tests/fixtures.rs index ab10a57..649ffbb 100644 --- a/src/tests/fixtures.rs +++ b/src/tests/fixtures.rs @@ -3407,6 +3407,18 @@ fn test_database_6_0() { crate::tests::runner::run_test_case(&path, 6, 0).unwrap(); } +#[test] +fn test_database_7_0() { + let path = format!("{}/fixtures/database.json", env!("CARGO_MANIFEST_DIR")); + crate::tests::runner::run_test_case(&path, 7, 0).unwrap(); +} + +#[test] +fn test_database_8_0() { + let path = format!("{}/fixtures/database.json", env!("CARGO_MANIFEST_DIR")); + crate::tests::runner::run_test_case(&path, 8, 0).unwrap(); +} + #[test] fn test_cases_0_0() { let path = format!("{}/fixtures/cases.json", env!("CARGO_MANIFEST_DIR")); @@ -7666,3 +7678,9 @@ fn test_merger_0_15() { let path = format!("{}/fixtures/merger.json", env!("CARGO_MANIFEST_DIR")); crate::tests::runner::run_test_case(&path, 0, 15).unwrap(); } + +#[test] +fn test_merger_0_16() { + let path = format!("{}/fixtures/merger.json", env!("CARGO_MANIFEST_DIR")); + crate::tests::runner::run_test_case(&path, 0, 16).unwrap(); +}