Compare commits

..

10 Commits

Author SHA1 Message Date
6444b300b3 version: 1.0.63 2026-03-15 23:03:15 -04:00
c529c8b8ea added jspg_schemas for mixer 2026-03-15 23:03:03 -04:00
2f15ae3d41 version: 1.0.62 2026-03-15 10:10:19 -04:00
f8528aa85e version: 1.0.61 2026-03-15 10:09:27 -04:00
b6f383e700 version: 1.0.60 2026-03-15 09:46:23 -04:00
db5183930d queryer supports subfiltering now 2026-03-15 07:49:05 -04:00
6de75ba525 merger notification process order testing 2026-03-15 07:31:14 -04:00
6632570712 merger improvements 2026-03-15 03:24:00 -04:00
d4347072f2 stripping schema nulls from stems 2026-03-15 00:13:33 -04:00
290464adc1 gjson pathing for stem paths 2026-03-13 23:35:37 -04:00
12 changed files with 666 additions and 384 deletions

View File

@ -80,6 +80,8 @@ The Merger provides an automated, high-performance graph synchronization engine
* **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.
* **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.
* **Sparse Updates**: Empty JSON strings `""` are directly bound as explicit SQL `NULL` directives to clear data, whilst omitted (missing) properties skip UPDATE execution entirely, ensuring partial UI submissions do not wipe out sibling fields.
* **Unified Return Structure**: To eliminate UI hydration race conditions and multi-user duplication, `jspg_merge` explicitly strips the response graph and returns only the root `{ "id": "uuid" }` (or an array of IDs for list insertions). External APIs can then explicitly call read APIs to fetch the resulting graph, while the UI relies 100% implicitly on the flat `pg_notify` pipeline for reactive state synchronization.
@ -101,22 +103,24 @@ The Queryer transforms Postgres into a pre-compiled Semantic Query Engine via th
* **Array Inclusion**: `{"$in": [values]}`, `{"$nin": [values]}` use native `jsonb_array_elements_text()` bindings to enforce `IN` and `NOT IN` logic without runtime SQL injection risks.
* **Text Matching (ILIKE)**: Evaluates `$eq` or `$ne` against string fields containing the `%` character natively into Postgres `ILIKE` and `NOT ILIKE` partial substring matches.
* **Type Casting**: Safely resolves dynamic combinations by casting values instantly into the physical database types mapped in the schema (e.g. parsing `uuid` bindings to `::uuid`, formatting DateTimes to `::timestamptz`, and numbers to `::numeric`).
### 4. The Stem Engine
### The Stem Engine
Rather than over-fetching heavy Entity payloads and trimming them, Punc Framework Websockets depend on isolated subgraphs defined as **Stems**.
A `Stem` is **not a JSON Pointer** or a physical path string (like `/properties/contacts/items/phone_number`). It is simply a declaration of an **Entity Type boundary** that exists somewhere within the compiled JSON Schema graph.
A `Stem` is a declaration of an **Entity Type boundary** that exists somewhere within the compiled JSON Schema graph, expressed using **`gjson` multipath syntax** (e.g., `contacts.#.phone_numbers.#`).
Because `pg_notify` (Beats) fire rigidly from physical Postgres tables (e.g. `{"type": "phone_number"}`), the Go Framework only ever needs to know: "Does the schema `with_contacts.person` contain the `phone_number` Entity anywhere inside its tree?"
Because `pg_notify` (Beats) fire rigidly from physical Postgres tables (e.g. `{"type": "phone_number"}`), the Go Framework only ever needs to know: "Does the schema `with_contacts.person` contain the `phone_number` Entity anywhere inside its tree, and if so, what is the gjson path to iterate its payload?"
* **Initialization:** During startup (`jspg_stems()`), the database crawls all Schemas and maps out every physical Entity Type it references. It builds a flat dictionary of `Schema ID -> [Entity Types]` (e.g. `with_contacts.person -> ["person", "contact", "phone_number", "email_address"]`).
* **Identifier Prioritization**: When determining if a nested object boundary is an Entity, JSPG natively prioritizes defined `$id` tags over `$ref` inheritance pointers to prevent polymorphic boundaries from devolving into their generic base classes.
* **Cyclical Deduplication**: Because Punc relationships often reference back on themselves via deeply nested classes, the Stem Engine applies intelligent path deduplication. If the active `current_path` already ends with the target entity string, it traverses the inheritance properties without appending the entity to the stem path again, eliminating infinite powerset loops.
* **Relationship Path Squashing:** When calculating nested string paths structurally to discover these boundaries, JSPG intentionally **omits** properties natively named `target` or `source` if they belong to a native database `relationship` table override. This ensures paths like `phone_numbers/contact/target` correctly register their beat resolution pattern as `phone_numbers/contact/phone_number`.
* **Initialization:** During startup (`jspg_stems()`), the database crawls all Schemas and maps out every physical Entity Type it references. It builds a highly optimized `HashMap<String, HashMap<String, Arc<Stem>>>` providing strictly `O(1)` memory lookups mapping `Schema ID -> { Stem Path -> Entity Type }`.
* **GJSON Pathing:** Unlike standard JSON Pointers, stems utilize `.#` array iterator syntax. The Go web server consumes this native path (e.g. `lines.#`) across the raw Postgres JSON byte payload, extracting all active UUIDs in one massive sub-millisecond sweep without unmarshaling Go ASTs.
* **Polymorphic Condition Selectors:** When trailing paths would otherwise collide because of abstract polymorphic type definitions (e.g., a `target` property bounded by a `oneOf` taking either `phone_number` or `email_address`), JSPG natively appends evaluated `gjson` type conditions into the path (e.g. `contacts.#.target#(type=="phone_number")`). This guarantees `O(1)` key uniqueness in the HashMap while retaining extreme array extraction speeds natively without runtime AST evaluation.
* **Identifier Prioritization:** When determining if a nested object boundary is an Entity, JSPG natively prioritizes defined `$id` tags over `$ref` inheritance pointers to prevent polymorphic boundaries from devolving into their generic base classes.
* **Cyclical Deduplication:** Because Punc relationships often reference back on themselves via deeply nested classes, the Stem Engine applies intelligent path deduplication. If the active `current_path` already ends with the target entity string, it traverses the inheritance properties without appending the entity to the stem path again, eliminating infinite powerset loops.
* **Relationship Path Squashing:** When calculating string paths structurally, JSPG intentionally **omits** properties natively named `target` or `source` if they belong to a native database `relationship` table override.
* **The Go Router**: The Golang Punc framework uses this exact mapping to register WebSocket Beat frequencies exclusively on the Entity types discovered.
* **The Queryer Execution**: When the Go framework asks JSPG to hydrate a partial `phone_number` stem for the `with_contacts.person` schema, instead of jumping through string paths, the SQL Compiler simply reaches into the Schema's AST using the `phone_number` Type string, pulls out exactly that entity's mapping rules, and returns a fully correlated `SELECT` block! This natively handles nested array properties injected via `oneOf` or array references efficiently bypassing runtime powerset expansion.
* **Performance:** These Stem execution structures are fully statically compiled via SPI and map perfectly to `O(1)` real-time routing logic on the application tier.
## 5. Testing & Execution Architecture
JSPG implements a strict separation of concerns to bypass the need to boot a full PostgreSQL cluster for unit and integration testing. Because `pgrx::spi::Spi` directly links to PostgreSQL C-headers, building the library with `cargo test` on macOS natively normally results in fatal `dyld` crashes.

View File

@ -1213,21 +1213,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"date_of_birth\":\"2000-01-01\",",
" \"first_name\":\"Bob\",",
" \"id\":\"{{uuid:customer_id}}\",",
" \"last_name\":\"Smith\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"type\":\"person\"",
" }",
"}')"
],
[
"INSERT INTO agreego.\"entity\" (",
" \"created_at\",",
@ -1295,6 +1280,21 @@
" \"type\":\"order\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"date_of_birth\":\"2000-01-01\",",
" \"first_name\":\"Bob\",",
" \"id\":\"{{uuid:customer_id}}\",",
" \"last_name\":\"Smith\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"type\":\"person\"",
" }",
"}')"
]
]
}
@ -1409,21 +1409,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:line_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"order_id\":\"abc\",",
" \"price\":99.0,",
" \"product\":\"Widget\",",
" \"type\":\"order_line\"",
" }",
"}')"
],
[
"INSERT INTO agreego.change (",
" changes,",
@ -1457,6 +1442,21 @@
" \"type\":\"order\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:line_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"order_id\":\"abc\",",
" \"price\":99.0,",
" \"product\":\"Widget\",",
" \"type\":\"order_line\"",
" }",
"}')"
]
]
}
@ -1587,19 +1587,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:phone1_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"number\":\"555-0001\",",
" \"type\":\"phone_number\"",
" }",
"}')"
],
[
"INSERT INTO agreego.\"entity\" (",
" \"created_at\",",
@ -1661,23 +1648,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:contact1_id}}\",",
" \"is_primary\":true,",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"source_id\":\"{{uuid:person_id}}\",",
" \"source_type\":\"person\",",
" \"target_id\":\"{{uuid:phone1_id}}\",",
" \"target_type\":\"phone_number\",",
" \"type\":\"contact\"",
" }",
"}')"
],
[
"INSERT INTO agreego.\"entity\" (",
" \"created_at\",",
@ -1722,19 +1692,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:phone2_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"number\":\"555-0002\",",
" \"type\":\"phone_number\"",
" }",
"}')"
],
[
"INSERT INTO agreego.\"entity\" (",
" \"created_at\",",
@ -1796,23 +1753,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:contact2_id}}\",",
" \"is_primary\":false,",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"source_id\":\"{{uuid:person_id}}\",",
" \"source_type\":\"person\",",
" \"target_id\":\"{{uuid:phone2_id}}\",",
" \"target_type\":\"phone_number\",",
" \"type\":\"contact\"",
" }",
"}')"
],
[
"INSERT INTO agreego.\"entity\" (",
" \"created_at\",",
@ -1857,19 +1797,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"address\":\"test@example.com\",",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:email1_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"type\":\"email_address\"",
" }",
"}')"
],
[
"INSERT INTO agreego.\"entity\" (",
" \"created_at\",",
@ -1931,23 +1858,6 @@
" '00000000-0000-0000-0000-000000000000'",
")"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:contact3_id}}\",",
" \"is_primary\":false,",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"source_id\":\"{{uuid:person_id}}\",",
" \"source_type\":\"person\",",
" \"target_id\":\"{{uuid:email1_id}}\",",
" \"target_type\":\"email_address\",",
" \"type\":\"contact\"",
" }",
"}')"
],
[
"INSERT INTO agreego.change (",
" changes,",
@ -1982,6 +1892,96 @@
" \"type\":\"person\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:contact1_id}}\",",
" \"is_primary\":true,",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"source_id\":\"{{uuid:person_id}}\",",
" \"source_type\":\"person\",",
" \"target_id\":\"{{uuid:phone1_id}}\",",
" \"target_type\":\"phone_number\",",
" \"type\":\"contact\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:phone1_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"number\":\"555-0001\",",
" \"type\":\"phone_number\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:contact2_id}}\",",
" \"is_primary\":false,",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"source_id\":\"{{uuid:person_id}}\",",
" \"source_type\":\"person\",",
" \"target_id\":\"{{uuid:phone2_id}}\",",
" \"target_type\":\"phone_number\",",
" \"type\":\"contact\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:phone2_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"number\":\"555-0002\",",
" \"type\":\"phone_number\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:contact3_id}}\",",
" \"is_primary\":false,",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"source_id\":\"{{uuid:person_id}}\",",
" \"source_type\":\"person\",",
" \"target_id\":\"{{uuid:email1_id}}\",",
" \"target_type\":\"email_address\",",
" \"type\":\"contact\"",
" }",
"}')"
],
[
"SELECT pg_notify('entity', '{",
" \"complete\":{",
" \"address\":\"test@example.com\",",
" \"created_at\":\"{{timestamp}}\",",
" \"created_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"id\":\"{{uuid:email1_id}}\",",
" \"modified_at\":\"{{timestamp}}\",",
" \"modified_by\":\"00000000-0000-0000-0000-000000000000\",",
" \"type\":\"email_address\"",
" }",
"}')"
]
]
}

View File

@ -947,6 +947,9 @@
"$eq": true,
"$ne": false
},
"contacts.#.is_primary": {
"$eq": true
},
"created_at": {
"$eq": "2020-01-01T00:00:00Z",
"$gt": "2019-01-01T00:00:00Z",
@ -970,10 +973,6 @@
"Bob"
]
},
"last_name": {
"$eq": "%Doe%",
"$ne": "%Smith%"
},
"id": {
"$eq": "00000000-0000-0000-0000-000000000001",
"$in": [
@ -983,6 +982,13 @@
"$nin": [
"00000000-0000-0000-0000-000000000002"
]
},
"last_name": {
"$eq": "%Doe%",
"$ne": "%Smith%"
},
"phone_numbers.#.target.number": {
"$eq": "555-1234"
}
},
"expect": {
@ -1037,6 +1043,7 @@
" JOIN agreego.entity t1_obj_t2_contacts_t3 ON t1_obj_t2_contacts_t3.id = t1_obj_t2_contacts_t2.id",
" WHERE",
" NOT t1_obj_t2_contacts_t1.archived",
" AND t1_obj_t2_contacts_t1.is_primary = ($11#>>'{}')::boolean",
" AND t1_obj_t2_contacts_t1.parent_id = t1_obj_t2.id),",
" 'created_at', t1_obj_t2.created_at,",
" 'email_addresses',",
@ -1093,6 +1100,7 @@
" JOIN agreego.entity t1_obj_t2_phone_numbers_t3_target_t2 ON t1_obj_t2_phone_numbers_t3_target_t2.id = t1_obj_t2_phone_numbers_t3_target_t1.id",
" WHERE",
" NOT t1_obj_t2_phone_numbers_t3_target_t1.archived",
" AND t1_obj_t2_phone_numbers_t3_target_t1.number ILIKE $32#>>'{}'",
" AND t1_obj_t2_phone_numbers_t3_target_t1.parent_id = t1_obj_t2_phone_numbers_t3.id",
" ),",
" 'type', t1_obj_t2_phone_numbers_t3.type",
@ -1119,26 +1127,26 @@
" AND t1_obj_t1.age NOT IN (SELECT value::numeric FROM jsonb_array_elements_text(($8#>>'{}')::jsonb))",
" AND t1_obj_t2.archived = ($9#>>'{}')::boolean",
" AND t1_obj_t2.archived != ($10#>>'{}')::boolean",
" AND t1_obj_t2.created_at = ($11#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at > ($12#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at >= ($13#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at < ($14#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at <= ($15#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at != ($16#>>'{}')::timestamptz",
" AND t1_obj_t1.first_name ILIKE $17#>>'{}'",
" AND t1_obj_t1.first_name > ($18#>>'{}')",
" AND t1_obj_t1.first_name >= ($19#>>'{}')",
" AND t1_obj_t1.first_name IN (SELECT value FROM jsonb_array_elements_text(($20#>>'{}')::jsonb))",
" AND t1_obj_t1.first_name < ($21#>>'{}')",
" AND t1_obj_t1.first_name <= ($22#>>'{}')",
" AND t1_obj_t1.first_name NOT ILIKE $23#>>'{}'",
" AND t1_obj_t1.first_name NOT IN (SELECT value FROM jsonb_array_elements_text(($24#>>'{}')::jsonb))",
" AND t1_obj_t2.id = ($25#>>'{}')::uuid",
" AND t1_obj_t2.id IN (SELECT value::uuid FROM jsonb_array_elements_text(($26#>>'{}')::jsonb))",
" AND t1_obj_t2.id != ($27#>>'{}')::uuid",
" AND t1_obj_t2.id NOT IN (SELECT value::uuid FROM jsonb_array_elements_text(($28#>>'{}')::jsonb))",
" AND t1_obj_t1.last_name ILIKE $29#>>'{}'",
" AND t1_obj_t1.last_name NOT ILIKE $30#>>'{}'",
" AND t1_obj_t2.created_at = ($12#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at > ($13#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at >= ($14#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at < ($15#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at <= ($16#>>'{}')::timestamptz",
" AND t1_obj_t2.created_at != ($17#>>'{}')::timestamptz",
" AND t1_obj_t1.first_name ILIKE $18#>>'{}'",
" AND t1_obj_t1.first_name > ($19#>>'{}')",
" AND t1_obj_t1.first_name >= ($20#>>'{}')",
" AND t1_obj_t1.first_name IN (SELECT value FROM jsonb_array_elements_text(($21#>>'{}')::jsonb))",
" AND t1_obj_t1.first_name < ($22#>>'{}')",
" AND t1_obj_t1.first_name <= ($23#>>'{}')",
" AND t1_obj_t1.first_name NOT ILIKE $24#>>'{}'",
" AND t1_obj_t1.first_name NOT IN (SELECT value FROM jsonb_array_elements_text(($25#>>'{}')::jsonb))",
" AND t1_obj_t2.id = ($26#>>'{}')::uuid",
" AND t1_obj_t2.id IN (SELECT value::uuid FROM jsonb_array_elements_text(($27#>>'{}')::jsonb))",
" AND t1_obj_t2.id != ($28#>>'{}')::uuid",
" AND t1_obj_t2.id NOT IN (SELECT value::uuid FROM jsonb_array_elements_text(($29#>>'{}')::jsonb))",
" AND t1_obj_t1.last_name ILIKE $30#>>'{}'",
" AND t1_obj_t1.last_name NOT ILIKE $31#>>'{}'",
")"
]
]
@ -1148,7 +1156,7 @@
"description": "Full person stem query on phone number contact",
"action": "query",
"schema_id": "full.person",
"stem": "phone_numbers/contact",
"stem": "phone_numbers.#",
"expect": {
"success": true,
"sql": [
@ -1159,7 +1167,22 @@
" 'id', t1_obj_t3.id,",
" 'is_primary', t1_obj_t1.is_primary,",
" 'name', t1_obj_t3.name,",
" 'type', t1_obj_t3.type)",
" 'target',",
" (SELECT jsonb_build_object(",
" 'archived', t1_obj_t3_target_t2.archived,",
" 'created_at', t1_obj_t3_target_t2.created_at,",
" 'id', t1_obj_t3_target_t2.id,",
" 'name', t1_obj_t3_target_t2.name,",
" 'number', t1_obj_t3_target_t1.number,",
" 'type', t1_obj_t3_target_t2.type",
" )",
" FROM agreego.phone_number t1_obj_t3_target_t1",
" JOIN agreego.entity t1_obj_t3_target_t2 ON t1_obj_t3_target_t2.id = t1_obj_t3_target_t1.id",
" WHERE",
" NOT t1_obj_t3_target_t1.archived",
" AND t1_obj_t3_target_t1.parent_id = t1_obj_t3.id),",
" 'type', t1_obj_t3.type",
")",
"FROM agreego.contact t1_obj_t1",
"JOIN agreego.relationship t1_obj_t2 ON t1_obj_t2.id = t1_obj_t1.id",
"JOIN agreego.entity t1_obj_t3 ON t1_obj_t3.id = t1_obj_t2.id",
@ -1172,7 +1195,7 @@
"description": "Full person stem query on phone number contact on phone number",
"action": "query",
"schema_id": "full.person",
"stem": "phone_numbers/contact/phone_number",
"stem": "phone_numbers.#.target",
"expect": {
"success": true,
"sql": [
@ -1195,7 +1218,7 @@
"description": "Full person stem query on contact email address",
"action": "query",
"schema_id": "full.person",
"stem": "contacts/contact/email_address",
"stem": "contacts.#.target#(type==\"email_address\")",
"expect": {
"success": true,
"sql": [

View File

@ -10,9 +10,13 @@
"type": "relation",
"constraint": "fk_contact_entity",
"source_type": "contact",
"source_columns": ["entity_id"],
"source_columns": [
"entity_id"
],
"destination_type": "person",
"destination_columns": ["id"],
"destination_columns": [
"id"
],
"prefix": null
},
{
@ -20,88 +24,132 @@
"type": "relation",
"constraint": "fk_relationship_target",
"source_type": "relationship",
"source_columns": ["target_id", "target_type"],
"source_columns": [
"target_id",
"target_type"
],
"destination_type": "entity",
"destination_columns": ["id", "type"],
"destination_columns": [
"id",
"type"
],
"prefix": "target"
}
],
"types": [
{
"name": "entity",
"hierarchy": ["entity"],
"schemas": [{
"$id": "entity",
"type": "object",
"properties": {}
}]
"hierarchy": [
"entity"
],
"schemas": [
{
"$id": "entity",
"type": "object",
"properties": {}
}
]
},
{
"name": "person",
"hierarchy": ["person", "entity"],
"schemas": [{
"$id": "person",
"$ref": "entity",
"properties": {}
}]
"hierarchy": [
"person",
"entity"
],
"schemas": [
{
"$id": "person",
"$ref": "entity",
"properties": {}
}
]
},
{
"name": "email_address",
"hierarchy": ["email_address", "entity"],
"schemas": [{
"$id": "email_address",
"$ref": "entity",
"properties": {}
}]
"hierarchy": [
"email_address",
"entity"
],
"schemas": [
{
"$id": "email_address",
"$ref": "entity",
"properties": {}
}
]
},
{
"name": "phone_number",
"hierarchy": ["phone_number", "entity"],
"schemas": [{
"$id": "phone_number",
"$ref": "entity",
"properties": {}
}]
"hierarchy": [
"phone_number",
"entity"
],
"schemas": [
{
"$id": "phone_number",
"$ref": "entity",
"properties": {}
}
]
},
{
"name": "relationship",
"relationship": true,
"hierarchy": ["relationship", "entity"],
"schemas": [{
"$id": "relationship",
"$ref": "entity",
"properties": {}
}]
"hierarchy": [
"relationship",
"entity"
],
"schemas": [
{
"$id": "relationship",
"$ref": "entity",
"properties": {}
}
]
},
{
"name": "contact",
"relationship": true,
"hierarchy": ["contact", "relationship", "entity"],
"schemas": [{
"$id": "contact",
"$ref": "relationship",
"properties": {
"target": {
"oneOf": [
{ "$ref": "phone_number" },
{ "$ref": "email_address" }
]
"hierarchy": [
"contact",
"relationship",
"entity"
],
"schemas": [
{
"$id": "contact",
"$ref": "relationship",
"properties": {
"target": {
"oneOf": [
{
"$ref": "phone_number"
},
{
"$ref": "email_address"
}
]
}
}
}
}]
]
},
{
"name": "save_person",
"schemas": [{
"$id": "save_person.response",
"$ref": "person",
"properties": {
"contacts": {
"type": "array",
"items": { "$ref": "contact" }
"schemas": [
{
"$id": "save_person.response",
"$ref": "person",
"properties": {
"contacts": {
"type": "array",
"items": {
"$ref": "contact"
}
}
}
}
}]
]
}
]
},
@ -112,59 +160,148 @@
"expect": {
"success": true,
"stems": {
"save_person.response": {
"": {
"type": "person"
},
"contacts/contact": {
"type": "contact",
"relation": "contacts_id"
},
"contacts/contact/email_address": {
"type": "email_address",
"relation": "target_id"
},
"contacts/contact/phone_number": {
"type": "phone_number",
"relation": "target_id"
}
},
"contact": {
"": {
"schema": {
"$id": "contact",
"$ref": "relationship",
"properties": {
"target": {
"oneOf": [
{
"$ref": "phone_number"
},
{
"$ref": "email_address"
}
]
}
}
},
"type": "contact"
},
"email_address": {
"type": "email_address",
"relation": "target_id"
"target#(type==\"email_address\")": {
"relation": "target_id",
"schema": {
"$id": "email_address",
"$ref": "entity",
"properties": {}
},
"type": "email_address"
},
"phone_number": {
"type": "phone_number",
"relation": "target_id"
}
},
"person": {
"": {
"type": "person"
"target#(type==\"phone_number\")": {
"relation": "target_id",
"schema": {
"$id": "phone_number",
"$ref": "entity",
"properties": {}
},
"type": "phone_number"
}
},
"email_address": {
"": {
"schema": {
"$id": "email_address",
"$ref": "entity",
"properties": {}
},
"type": "email_address"
}
},
"phone_number": {
"entity": {
"": {
"schema": {
"$id": "entity",
"properties": {},
"type": "object"
},
"type": "entity"
}
},
"person": {
"": {
"schema": {
"$id": "person",
"$ref": "entity",
"properties": {}
},
"type": "person"
}
},
"phone_number": {
"": {
"schema": {
"$id": "phone_number",
"$ref": "entity",
"properties": {}
},
"type": "phone_number"
}
},
"relationship": {
"": {
"schema": {
"$id": "relationship",
"$ref": "entity",
"properties": {}
},
"type": "relationship"
}
},
"entity": {
"save_person.response": {
"": {
"type": "entity"
"schema": {
"$id": "save_person.response",
"$ref": "person",
"properties": {
"contacts": {
"items": {
"$ref": "contact"
},
"type": "array"
}
}
},
"type": "person"
},
"contacts.#": {
"relation": "contacts_id",
"schema": {
"$id": "contact",
"$ref": "relationship",
"properties": {
"target": {
"oneOf": [
{
"$ref": "phone_number"
},
{
"$ref": "email_address"
}
]
}
}
},
"type": "contact"
},
"contacts.#.target#(type==\"email_address\")": {
"relation": "target_id",
"schema": {
"$id": "email_address",
"$ref": "entity",
"properties": {}
},
"type": "email_address"
},
"contacts.#.target#(type==\"phone_number\")": {
"relation": "target_id",
"schema": {
"$id": "phone_number",
"$ref": "entity",
"properties": {}
},
"type": "phone_number"
}
}
}

View File

@ -265,12 +265,12 @@ impl Database {
String::from(""),
None,
None,
true,
false,
&mut inner_map,
Vec::new(),
&mut errors,
);
if !inner_map.is_empty() {
println!("SCHEMA: {} STEMS: {:?}", schema_id, inner_map.keys());
db_stems.insert(schema_id, inner_map);
}
}
@ -288,11 +288,12 @@ impl Database {
db: &Database,
root_schema_id: &str,
schema: &Schema,
mut current_path: String,
current_path: String,
parent_type: Option<String>,
property_name: Option<String>,
is_root: bool,
is_polymorphic: bool,
inner_map: &mut HashMap<String, Arc<Stem>>,
seen_entities: Vec<String>,
errors: &mut Vec<crate::drop::Error>,
) {
let mut is_entity = false;
@ -323,6 +324,12 @@ impl Database {
}
}
if is_entity {
if seen_entities.contains(&entity_type) {
return; // Break cyclical schemas!
}
}
let mut relation_col = None;
if is_entity {
if let (Some(pt), Some(prop)) = (&parent_type, &property_name) {
@ -344,46 +351,21 @@ impl Database {
}
}
let mut final_path = current_path.clone();
if is_polymorphic && !final_path.is_empty() && !final_path.ends_with(&entity_type) {
if final_path.ends_with(".#") {
final_path = format!("{}(type==\"{}\")", final_path, entity_type);
} else {
final_path = format!("{}#(type==\"{}\")", final_path, entity_type);
}
}
let stem = Stem {
r#type: entity_type.clone(),
relation: relation_col,
schema: Arc::new(schema.clone()),
};
let mut branch_path = if is_root {
String::new()
} else if current_path.is_empty() {
entity_type.clone()
} else {
format!("{}/{}", current_path, entity_type)
};
// DEDUPLICATION: If we just recursed into the EXACT same entity type definition,
// do not append again and do not re-register the stem.
let already_registered =
if current_path == entity_type || current_path.ends_with(&format!("/{}", entity_type)) {
branch_path = current_path.clone();
true
} else {
false
};
if !already_registered {
if inner_map.contains_key(&branch_path) {
errors.push(crate::drop::Error {
code: "STEM_COLLISION".to_string(),
message: format!("The stem path `{}` resolves to multiple Entity boundaries. This usually occurs during un-wrapped $family or oneOf polymorphic schemas where multiple Entities are directly assigned to the same property. To fix this, encapsulate the polymorphic branch.", branch_path),
details: crate::drop::ErrorDetails {
path: root_schema_id.to_string(),
},
});
}
inner_map.insert(branch_path.clone(), Arc::new(stem));
}
// Update current_path for structural children
current_path = branch_path;
inner_map.insert(final_path, Arc::new(stem));
}
let next_parent = if is_entity {
@ -392,34 +374,22 @@ impl Database {
parent_type.clone()
};
let pass_seen = if is_entity {
let mut ns = seen_entities.clone();
ns.push(entity_type.clone());
ns
} else {
seen_entities.clone()
};
// Properties branch
if let Some(props) = &schema.obj.properties {
for (k, v) in props {
// Bypass target and source properties if we are in a relationship
if let Some(parent_str) = &next_parent {
if let Some(pt) = db.types.get(parent_str) {
if pt.relationship && (k == "target" || k == "source") {
Self::discover_stems(
db,
root_schema_id,
v,
current_path.clone(),
next_parent.clone(),
Some(k.clone()),
false,
inner_map,
errors,
);
continue;
}
}
}
// Standard Property Pathing
let next_path = if current_path.is_empty() {
k.clone()
} else {
format!("{}/{}", current_path, k)
format!("{}.{}", current_path, k)
};
Self::discover_stems(
@ -431,6 +401,7 @@ impl Database {
Some(k.clone()),
false,
inner_map,
pass_seen.clone(),
errors,
);
}
@ -438,15 +409,22 @@ impl Database {
// Array Item branch
if let Some(items) = &schema.obj.items {
let next_path = if current_path.is_empty() {
String::from("#")
} else {
format!("{}.#", current_path)
};
Self::discover_stems(
db,
root_schema_id,
items,
current_path.clone(),
next_path,
next_parent.clone(),
property_name.clone(),
false, // Arrays themselves aren't polymorphic branches, their items might be
false,
inner_map,
pass_seen.clone(),
errors,
);
}
@ -463,8 +441,9 @@ impl Database {
current_path.clone(),
next_parent.clone(),
property_name.clone(),
false,
is_polymorphic,
inner_map,
seen_entities.clone(),
errors,
);
}
@ -481,8 +460,9 @@ impl Database {
current_path.clone(),
next_parent.clone(),
property_name.clone(),
false,
true,
inner_map,
pass_seen.clone(),
errors,
);
}
@ -496,8 +476,9 @@ impl Database {
current_path.clone(),
next_parent.clone(),
property_name.clone(),
false,
is_polymorphic,
inner_map,
pass_seen.clone(),
errors,
);
}

View File

@ -5,7 +5,6 @@ use std::sync::Arc;
// Schema mirrors the Go Punc Generator's schema struct for consistency.
// It is an order-preserving representation of a JSON Schema.
pub fn deserialize_some<'de, D>(deserializer: D) -> Result<Option<Value>, D::Error>
where
D: serde::Deserializer<'de>,
@ -13,117 +12,159 @@ where
let v = Value::deserialize(deserializer)?;
Ok(Some(v))
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct SchemaObject {
// Core Schema Keywords
#[serde(rename = "$id")]
#[serde(skip_serializing_if = "Option::is_none")]
pub id: Option<String>,
#[serde(rename = "$ref")]
#[serde(skip_serializing_if = "Option::is_none")]
pub r#ref: Option<String>,
/*
Note: The `Ref` field in the Go struct is a pointer populated by the linker.
In Rust, we might handle this differently (e.g., separate lookup or Rc/Arc),
so we omit the direct recursive `Ref` field for now and rely on `ref_string`.
*/
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub title: Option<String>,
#[serde(default)] // Allow missing type
#[serde(rename = "type")]
#[serde(skip_serializing_if = "Option::is_none")]
pub type_: Option<SchemaTypeOrArray>, // Handles string or array of strings
// Object Keywords
#[serde(skip_serializing_if = "Option::is_none")]
pub properties: Option<BTreeMap<String, Arc<Schema>>>,
#[serde(rename = "patternProperties")]
#[serde(skip_serializing_if = "Option::is_none")]
pub pattern_properties: Option<BTreeMap<String, Arc<Schema>>>,
#[serde(rename = "additionalProperties")]
#[serde(skip_serializing_if = "Option::is_none")]
pub additional_properties: Option<Arc<Schema>>,
#[serde(rename = "$family")]
#[serde(skip_serializing_if = "Option::is_none")]
pub family: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub required: Option<Vec<String>>,
// dependencies can be schema dependencies or property dependencies
#[serde(skip_serializing_if = "Option::is_none")]
pub dependencies: Option<BTreeMap<String, Dependency>>,
// Array Keywords
#[serde(rename = "items")]
#[serde(skip_serializing_if = "Option::is_none")]
pub items: Option<Arc<Schema>>,
#[serde(rename = "prefixItems")]
#[serde(skip_serializing_if = "Option::is_none")]
pub prefix_items: Option<Vec<Arc<Schema>>>,
// String Validation
#[serde(rename = "minLength")]
#[serde(skip_serializing_if = "Option::is_none")]
pub min_length: Option<f64>,
#[serde(rename = "maxLength")]
#[serde(skip_serializing_if = "Option::is_none")]
pub max_length: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub pattern: Option<String>,
// Array Validation
#[serde(rename = "minItems")]
#[serde(skip_serializing_if = "Option::is_none")]
pub min_items: Option<f64>,
#[serde(rename = "maxItems")]
#[serde(skip_serializing_if = "Option::is_none")]
pub max_items: Option<f64>,
#[serde(rename = "uniqueItems")]
#[serde(skip_serializing_if = "Option::is_none")]
pub unique_items: Option<bool>,
#[serde(rename = "contains")]
#[serde(skip_serializing_if = "Option::is_none")]
pub contains: Option<Arc<Schema>>,
#[serde(rename = "minContains")]
#[serde(skip_serializing_if = "Option::is_none")]
pub min_contains: Option<f64>,
#[serde(rename = "maxContains")]
#[serde(skip_serializing_if = "Option::is_none")]
pub max_contains: Option<f64>,
// Object Validation
#[serde(rename = "minProperties")]
#[serde(skip_serializing_if = "Option::is_none")]
pub min_properties: Option<f64>,
#[serde(rename = "maxProperties")]
#[serde(skip_serializing_if = "Option::is_none")]
pub max_properties: Option<f64>,
#[serde(rename = "propertyNames")]
#[serde(skip_serializing_if = "Option::is_none")]
pub property_names: Option<Arc<Schema>>,
// Numeric Validation
#[serde(skip_serializing_if = "Option::is_none")]
pub format: Option<String>,
#[serde(rename = "enum")]
#[serde(skip_serializing_if = "Option::is_none")]
pub enum_: Option<Vec<Value>>, // `enum` is a reserved keyword in Rust
#[serde(
default,
rename = "const",
deserialize_with = "crate::database::schema::deserialize_some"
)]
#[serde(skip_serializing_if = "Option::is_none")]
pub const_: Option<Value>,
// Numeric Validation
#[serde(rename = "multipleOf")]
#[serde(skip_serializing_if = "Option::is_none")]
pub multiple_of: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub minimum: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub maximum: Option<f64>,
#[serde(rename = "exclusiveMinimum")]
#[serde(skip_serializing_if = "Option::is_none")]
pub exclusive_minimum: Option<f64>,
#[serde(rename = "exclusiveMaximum")]
#[serde(skip_serializing_if = "Option::is_none")]
pub exclusive_maximum: Option<f64>,
// Combining Keywords
#[serde(rename = "allOf")]
#[serde(skip_serializing_if = "Option::is_none")]
pub all_of: Option<Vec<Arc<Schema>>>,
#[serde(rename = "oneOf")]
#[serde(skip_serializing_if = "Option::is_none")]
pub one_of: Option<Vec<Arc<Schema>>>,
#[serde(rename = "not")]
#[serde(skip_serializing_if = "Option::is_none")]
pub not: Option<Arc<Schema>>,
#[serde(rename = "if")]
#[serde(skip_serializing_if = "Option::is_none")]
pub if_: Option<Arc<Schema>>,
#[serde(rename = "then")]
#[serde(skip_serializing_if = "Option::is_none")]
pub then_: Option<Arc<Schema>>,
#[serde(rename = "else")]
#[serde(skip_serializing_if = "Option::is_none")]
pub else_: Option<Arc<Schema>>,
// Custom Vocabularies
#[serde(skip_serializing_if = "Option::is_none")]
pub form: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub display: Option<Vec<String>>,
#[serde(rename = "enumNames")]
#[serde(skip_serializing_if = "Option::is_none")]
pub enum_names: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub control: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub actions: Option<BTreeMap<String, Action>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub computer: Option<String>,
#[serde(default)]
#[serde(skip_serializing_if = "Option::is_none")]
pub extensible: Option<bool>,
#[serde(skip)]
@ -331,7 +372,9 @@ pub enum SchemaTypeOrArray {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Action {
#[serde(skip_serializing_if = "Option::is_none")]
pub navigate: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub punc: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]

View File

@ -8,10 +8,5 @@ pub struct Stem {
#[serde(skip_serializing_if = "Option::is_none")]
pub relation: Option<String>,
// The actual database schema node mapping for
// O(1) jump table execution for queryer.
//
// Automatically skipped from `jspg_stems()` JSON payload output.
#[serde(skip)]
pub schema: Arc<Schema>,
}

View File

@ -111,9 +111,7 @@ pub fn jspg_validate(schema_id: &str, instance: JsonB) -> JsonB {
}
#[cfg_attr(not(test), pg_extern)]
pub fn jspg_stems() -> JsonB {
use serde_json::{Map, Value};
pub fn jspg_schemas() -> JsonB {
let engine_opt = {
let lock = GLOBAL_JSPG.read().unwrap();
lock.clone()
@ -121,9 +119,28 @@ pub fn jspg_stems() -> JsonB {
match engine_opt {
Some(engine) => {
JsonB(serde_json::to_value(&engine.database.stems).unwrap_or(Value::Object(Map::new())))
let schemas_json = serde_json::to_value(&engine.database.schemas)
.unwrap_or(serde_json::Value::Object(serde_json::Map::new()));
JsonB(schemas_json)
}
None => JsonB(Value::Object(Map::new())),
None => JsonB(serde_json::Value::Object(serde_json::Map::new())),
}
}
#[cfg_attr(not(test), pg_extern)]
pub fn jspg_stems() -> JsonB {
let engine_opt = {
let lock = GLOBAL_JSPG.read().unwrap();
lock.clone()
};
match engine_opt {
Some(engine) => {
let stems_json = serde_json::to_value(&engine.database.stems)
.unwrap_or(serde_json::Value::Object(serde_json::Map::new()));
JsonB(stems_json)
}
None => JsonB(serde_json::Value::Object(serde_json::Map::new())),
}
}

View File

@ -21,63 +21,87 @@ impl Merger {
}
pub fn merge(&self, data: Value) -> crate::drop::Drop {
match self.merge_internal(data) {
let mut val_resolved = Value::Null;
let mut notifications_queue = Vec::new();
let result = self.merge_internal(data, &mut notifications_queue);
match result {
Ok(val) => {
let stripped_val = match val {
Value::Object(mut map) => {
val_resolved = val;
}
Err(msg) => {
return crate::drop::Drop::with_errors(vec![crate::drop::Error {
code: "MERGE_FAILED".to_string(),
message: msg,
details: crate::drop::ErrorDetails {
path: "".to_string(),
},
}]);
}
};
// Execute the globally collected, pre-ordered notifications last!
for notify_sql in notifications_queue {
if let Err(e) = self.db.execute(&notify_sql, None) {
return crate::drop::Drop::with_errors(vec![crate::drop::Error {
code: "MERGE_FAILED".to_string(),
message: format!("Executor Error in pre-ordered notify: {:?}", e),
details: crate::drop::ErrorDetails {
path: "".to_string(),
},
}]);
}
}
let stripped_val = match val_resolved {
Value::Object(mut map) => {
let mut out = serde_json::Map::new();
if let Some(id) = map.remove("id") {
out.insert("id".to_string(), id);
}
Value::Object(out)
}
Value::Array(arr) => {
let mut out_arr = Vec::new();
for item in arr {
if let Value::Object(mut map) = item {
let mut out = serde_json::Map::new();
if let Some(id) = map.remove("id") {
out.insert("id".to_string(), id);
}
Value::Object(out)
out_arr.push(Value::Object(out));
} else {
out_arr.push(Value::Null);
}
Value::Array(arr) => {
let mut out_arr = Vec::new();
for item in arr {
if let Value::Object(mut map) = item {
let mut out = serde_json::Map::new();
if let Some(id) = map.remove("id") {
out.insert("id".to_string(), id);
}
out_arr.push(Value::Object(out));
} else {
out_arr.push(Value::Null);
}
}
Value::Array(out_arr)
}
other => other,
};
crate::drop::Drop::success_with_val(stripped_val)
}
Value::Array(out_arr)
}
Err(msg) => crate::drop::Drop::with_errors(vec![crate::drop::Error {
code: "MERGE_FAILED".to_string(),
message: msg,
details: crate::drop::ErrorDetails {
path: "".to_string(),
},
}]),
}
other => other,
};
crate::drop::Drop::success_with_val(stripped_val)
}
pub(crate) fn merge_internal(&self, data: Value) -> Result<Value, String> {
pub(crate) fn merge_internal(&self, data: Value, notifications: &mut Vec<String>) -> Result<Value, String> {
match data {
Value::Array(items) => self.merge_array(items),
Value::Object(map) => self.merge_object(map),
Value::Array(items) => self.merge_array(items, notifications),
Value::Object(map) => self.merge_object(map, notifications),
_ => Err("Invalid merge payload: root must be an Object or Array".to_string()),
}
}
fn merge_array(&self, items: Vec<Value>) -> Result<Value, String> {
fn merge_array(&self, items: Vec<Value>, notifications: &mut Vec<String>) -> Result<Value, String> {
let mut resolved_items = Vec::new();
for item in items {
let resolved = self.merge_internal(item)?;
let resolved = self.merge_internal(item, notifications)?;
resolved_items.push(resolved);
}
Ok(Value::Array(resolved_items))
}
fn merge_object(&self, obj: serde_json::Map<String, Value>) -> Result<Value, String> {
fn merge_object(&self, obj: serde_json::Map<String, Value>, notifications: &mut Vec<String>) -> Result<Value, String> {
let queue_start = notifications.len();
let type_name = match obj.get("type").and_then(|v| v.as_str()) {
Some(t) => t.to_string(),
None => return Err("Missing required 'type' field on object".to_string()),
@ -147,7 +171,7 @@ impl Merger {
}
}
let merged_relative = match self.merge_internal(Value::Object(relative))? {
let merged_relative = match self.merge_internal(Value::Object(relative), notifications)? {
Value::Object(m) => m,
_ => continue,
};
@ -174,7 +198,7 @@ impl Merger {
&entity_fields,
);
let merged_relative = match self.merge_internal(Value::Object(relative))? {
let merged_relative = match self.merge_internal(Value::Object(relative), notifications)? {
Value::Object(m) => m,
_ => continue,
};
@ -242,7 +266,7 @@ impl Merger {
&entity_fields,
);
let merged_relative = match self.merge_internal(Value::Object(relative_item))? {
let merged_relative = match self.merge_internal(Value::Object(relative_item), notifications)? {
Value::Object(m) => m,
_ => continue,
};
@ -255,7 +279,7 @@ impl Merger {
}
// 7. Perform change tracking
self.merge_entity_change(
let notify_sql = self.merge_entity_change(
&entity_fields,
entity_fetched.as_ref(),
entity_change_kind.as_deref(),
@ -263,6 +287,10 @@ impl Merger {
&timestamp,
)?;
if let Some(sql) = notify_sql {
notifications.insert(queue_start, sql);
}
// Produce the full tree response
let mut final_response = serde_json::Map::new();
if let Some(fetched) = entity_fetched {
@ -614,10 +642,10 @@ impl Merger {
entity_change_kind: Option<&str>,
user_id: &str,
timestamp: &str,
) -> Result<(), String> {
) -> Result<Option<String>, String> {
let change_kind = match entity_change_kind {
Some(k) => k,
None => return Ok(()),
None => return Ok(None),
};
let id_str = entity_fields.get("id").unwrap();
@ -697,12 +725,8 @@ impl Merger {
.db
.execute(&change_sql, None)
.map_err(|e| format!("Executor Error in change: {:?}", e))?;
self
.db
.execute(&notify_sql, None)
.map_err(|e| format!("Executor Error in notify: {:?}", e))?;
Ok(())
Ok(Some(notify_sql))
}
fn compare_entities(

View File

@ -47,7 +47,7 @@ impl SqlCompiler {
// We expect the top level to typically be an Object or Array
let is_stem_query = stem_path.is_some();
let (sql, _) = self.walk_schema(target_schema, "t1", None, filter_keys, is_stem_query, 0)?;
let (sql, _) = self.walk_schema(target_schema, "t1", None, filter_keys, is_stem_query, 0, String::new())?;
Ok(sql)
}
@ -61,17 +61,21 @@ impl SqlCompiler {
filter_keys: &[String],
is_stem_query: bool,
depth: usize,
current_path: String,
) -> Result<(String, String), String> {
// Determine the base schema type (could be an array, object, or literal)
match &schema.obj.type_ {
Some(crate::database::schema::SchemaTypeOrArray::Single(t)) if t == "array" => {
// Handle Arrays:
if let Some(items) = &schema.obj.items {
let next_path = if current_path.is_empty() {
String::from("#")
} else {
format!("{}.#", current_path)
};
if let Some(ref_id) = &items.obj.r#ref {
if let Some(type_def) = self.db.types.get(ref_id) {
if is_stem_query && depth > 0 {
return Ok(("".to_string(), "abort".to_string()));
}
return self.compile_entity_node(
items,
type_def,
@ -81,6 +85,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth,
next_path,
);
}
}
@ -91,6 +96,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth + 1,
next_path,
)?;
return Ok((
format!("(SELECT jsonb_agg({}) FROM TODO)", item_sql),
@ -115,9 +121,6 @@ impl SqlCompiler {
}
if let Some(type_def) = resolved_type {
if is_stem_query && depth > 0 {
return Ok(("".to_string(), "abort".to_string()));
}
return self.compile_entity_node(
schema,
type_def,
@ -127,6 +130,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth,
current_path,
);
}
@ -141,6 +145,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth,
current_path,
);
}
return Err(format!("Unresolved $ref: {}", ref_id));
@ -154,6 +159,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth,
current_path,
);
}
@ -201,6 +207,7 @@ impl SqlCompiler {
filter_keys: &[String],
is_stem_query: bool,
depth: usize,
current_path: String,
) -> Result<(String, String), String> {
let local_ctx = format!("{}_{}", parent_alias, prop_name.unwrap_or("obj"));
@ -216,6 +223,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth,
&current_path,
)?;
let jsonb_obj_sql = if select_args.is_empty() {
@ -232,6 +240,7 @@ impl SqlCompiler {
parent_alias,
prop_name,
filter_keys,
&current_path,
)?;
let selection = if is_array {
@ -291,6 +300,7 @@ impl SqlCompiler {
filter_keys: &[String],
is_stem_query: bool,
depth: usize,
current_path: &str,
) -> Result<Vec<String>, String> {
let mut select_args = Vec::new();
let grouped_fields = type_def.grouped_fields.as_ref().and_then(|v| v.as_object());
@ -316,6 +326,12 @@ impl SqlCompiler {
}
}
let next_path = if current_path.is_empty() {
prop_key.clone()
} else {
format!("{}.{}", current_path, prop_key)
};
let (val_sql, val_type) = self.walk_schema(
prop_schema,
&owner_alias,
@ -323,6 +339,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth + 1,
next_path,
)?;
if val_type != "abort" {
@ -340,6 +357,7 @@ impl SqlCompiler {
parent_alias: &str,
prop_name: Option<&str>,
filter_keys: &[String],
current_path: &str,
) -> Result<Vec<String>, String> {
let base_alias = table_aliases
.get(&type_def.name)
@ -349,15 +367,32 @@ impl SqlCompiler {
let mut where_clauses = Vec::new();
where_clauses.push(format!("NOT {}.archived", base_alias));
if parent_alias == "t1" {
for (i, filter_key) in filter_keys.iter().enumerate() {
let mut parts = filter_key.split(':');
let field_name = parts.next().unwrap_or(filter_key);
let op = parts.next().unwrap_or("$eq");
for (i, filter_key) in filter_keys.iter().enumerate() {
let mut parts = filter_key.split(':');
let full_field_path = parts.next().unwrap_or(filter_key);
let op = parts.next().unwrap_or("$eq");
let mut filter_alias = base_alias.clone();
let field_name = if current_path.is_empty() {
if full_field_path.contains('.') || full_field_path.contains('#') {
continue;
}
full_field_path
} else {
let prefix = format!("{}.", current_path);
if full_field_path.starts_with(&prefix) {
let remainder = &full_field_path[prefix.len()..];
if remainder.contains('.') || remainder.contains('#') {
continue;
}
remainder
} else {
continue;
}
};
if let Some(gf) = type_def.grouped_fields.as_ref().and_then(|v| v.as_object()) {
let mut filter_alias = base_alias.clone();
if let Some(gf) = type_def.grouped_fields.as_ref().and_then(|v| v.as_object()) {
for (t_name, fields_val) in gf {
if let Some(fields_arr) = fields_val.as_array() {
if fields_arr.iter().any(|v| v.as_str() == Some(field_name)) {
@ -459,7 +494,6 @@ impl SqlCompiler {
"{}.{} {} {}",
filter_alias, field_name, sql_op, param_sql
));
}
}
}
@ -477,9 +511,16 @@ impl SqlCompiler {
filter_keys: &[String],
is_stem_query: bool,
depth: usize,
current_path: String,
) -> Result<(String, String), String> {
let mut build_args = Vec::new();
for (k, v) in props {
let next_path = if current_path.is_empty() {
k.clone()
} else {
format!("{}.{}", current_path, k)
};
let (child_sql, val_type) = self.walk_schema(
v,
parent_alias,
@ -487,6 +528,7 @@ impl SqlCompiler {
filter_keys,
is_stem_query,
depth + 1,
next_path,
)?;
if val_type == "abort" {
continue;

View File

@ -49,6 +49,22 @@ fn test_library_api() {
})
);
// 3. Validate jspg_schemas
let schemas_drop = jspg_schemas();
assert_eq!(
schemas_drop.0,
json!({
"test_schema": {
"$id": "test_schema",
"type": "object",
"properties": {
"name": { "type": "string" }
},
"required": ["name"]
}
})
);
// 4. Validate Happy Path
let happy_drop = jspg_validate("test_schema", JsonB(json!({"name": "Neo"})));
assert_eq!(

View File

@ -1 +1 @@
1.0.59
1.0.63