From bb2e5720bfc5946f7f6f3bea3ec4c3ce3851bed2 Mon Sep 17 00:00:00 2001 From: Baran Sekin Date: Mon, 14 Sep 2026 13:08:13 +0300 Subject: [PATCH 1/4] feat(indexes): generate offline attribute index SQL --- CLAUDE.md | 12 +- README.md | 68 ++++++++++ bin/workflow.js | 9 ++ src/commands/indexes.js | 45 +++++++ src/lib/indexes/definitions.js | 126 +++++++++++++++++++ src/lib/indexes/sql.js | 220 +++++++++++++++++++++++++++++++++ src/lib/indexes/versions.js | 39 ++++++ test/indexes.postgres.test.js | 97 +++++++++++++++ test/indexes.test.js | 93 ++++++++++++++ 9 files changed, 708 insertions(+), 1 deletion(-) create mode 100644 src/commands/indexes.js create mode 100644 src/lib/indexes/definitions.js create mode 100644 src/lib/indexes/sql.js create mode 100644 src/lib/indexes/versions.js create mode 100644 test/indexes.postgres.test.js create mode 100644 test/indexes.test.js diff --git a/CLAUDE.md b/CLAUDE.md index 16eb68f..416d02c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -17,7 +17,7 @@ node bin/workflow.js # run the CLI directly (same as `npm run dev` / `npm star ``` - **`build` is a no-op** (`echo 'Build not needed for now'`) — nothing to compile. -- **There is no test suite, linter, or formatter configured.** Do not invent `npm test`/`npm run lint` commands; they will fail. Verify changes by running the CLI against a real vNext project directory. +- **Index SQL tests**: `node --test test/indexes.test.js` (Node 18+); real PostgreSQL tests use `VNEXT_INDEX_TEST_URL=... node --test test/indexes.postgres.test.js` against a disposable database. No npm test/lint script is configured. Also verify the CLI against a real vNext project directory. The CLI always treats `process.cwd()` as the project root and requires at least one solution file (`vnext.config.json` or `vnext.{domain}.config.json`) in that directory. To exercise it, `cd` into a vNext workspace (not this repo) before running. A two-solution testbed lives at `../vnext-example` (`core` + `partner`); copy it into a scratch directory before running write-mode commands against it. @@ -63,3 +63,13 @@ Every workspace command runs the table above **once per solution, sequentially** - Library functions (`discover.js`, `workflow.js`, `csx.js`) take a **solution object** (`projectRoot`, `componentsRoot`, `componentTypes`, …) rather than a bare `projectRoot` or reading cwd directly; commands receive it from `runForEachSolution`. Only `config.get('PROJECT_ROOT')` (inside `solutions.loadWorkspace`) reads cwd. - Commands are split into a thin `xxxCommand(options)` (header, prompts that must happen once, then `runForEachSolution`) and an `xxxSolution(solution, options)` body that returns `{ success, failed, errors }` so the workspace summary can aggregate. - DB and API helpers swallow connection errors and return `false`/`null` rather than throwing — callers treat a missing instance as "not in DB". + +## Manual attribute-index SQL + +`wf indexes generate` is strictly offline: `src/lib/indexes/` resolves local Master +references and emits SQL; `src/commands/indexes.js` writes immutable batches. Never add DB execution +or automatic sync/publish hooks. DBA execution owns the maintenance window. Keep physical keys/columns +compatible with runtime `AttributeIndexDefinition` (`v1:latest` SHA-256 first 24 hex); version matching +follows runtime `InstanceDataVersionComparer`, including package revisions. SQL compares actual index +structure, skips equivalent indexes and only rebuilds/removes owned indexes. Projection retirement +requires explicit `--retire-obsolete` and a complete local inventory of active workflow versions. diff --git a/README.md b/README.md index 5db14b7..86bf5eb 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,74 @@ wf reset All workspace commands (`check`, `csx`, `sync`, `update`, `reset`) accept the global `--domain ` option. Without it they run once per solution file found in the project root. +### `wf indexes generate` + +**Purpose**: Generate offline attribute-index SQL for planned execution by your DBA team. + +From a domain workspace containing `vnext.config.json`: + +```bash +wf indexes generate --output ./index-sql +wf indexes generate --flow money-transfer --output ./index-sql +``` + +Each invocation writes a **new batch folder** containing one `.sql` per workflow, a `manifest.json` +(source paths, versions, SHA-256 checksums, physical index definitions), and execution notes. Earlier +batches are preserved. The command never connects to API/DB and never executes SQL; `sync`, `update` +and Master publication do not invoke it. + +The generator resolves local workflow → Master references from the configured component directories. +All local versions of a workflow contribute requirements. Full `-pkg.` revisions remain pinned; `latest`, +artifact, minor and major selectors use runtime version ordering. Build metadata is ignored. Referenced +schemas must exist locally; incomplete references, duplicate versions and incompatible types fail. + +Indexed fields use `x-indexed: true`. Nested scalar string/number/integer/boolean fields are supported; +dates use `format: "date-time"`. Filtering/sorting permissions remain `x-filterOperators`/`x-sortable`. +The generator preserves the runtime `v1:latest` column/key contract; readable index names include a hash +of their physical definition. + +The DBA reviews and executes each SQL file in a maintenance window: + +```bash +psql -X -v ON_ERROR_STOP=1 --dbname=vNext_MyDomainDb --file=index-sql//money_transfer.sql +``` + +SQL takes an advisory lock plus an **ACCESS EXCLUSIVE table lock**. +It validates current/history data, adds stored generated columns in one table rewrite, reuses structurally +matching indexes (including legacy names), rebuilds changed owned indexes, and updates the runtime ready +catalog atomically. Index comparison covers keys, includes, ordering, collation, access method, operator +classes and partial predicates. Unmanaged collisions abort; no `CASCADE` is issued. Replaying an unchanged +batch preserves index OIDs and skips data validation, rewrites and ANALYZE. Missing/invalid values are not +silently coerced; conversion errors identify the field and roll back all changes. + +This is **not concurrent DDL**. Plan disk/WAL/replica capacity for generated-column rewrites; the script's +5-second lock timeout may be adjusted during DBA review. Text-search indexes require `pg_trgm` in `public` +and `tr-TR-x-icu`. Runtime routing uses `AttributeIndexes:Enabled`; catalog refresh defaults to 30 seconds. + +Obsolete projections are retained by default for other deployed versions. To explicitly retire them: + +```bash +wf indexes generate --flow money-transfer --retire-obsolete +``` + +Include **every still-active workflow version** locally and drain readers/writers for this maintenance. +Retirement removes obsolete expressions/owned indexes and marks catalog entries not ready; columns and +stored values remain. This is required when an old numeric/date cast would reject values after a type +change. The offline tool cannot verify deployed versions: compare the source manifest before execution. +Roll back query routing through the runtime's `AttributeIndexes:DisabledFlows`; schedule physical cleanup +separately. See the runtime [maintenance runbook](https://github.com/burgan-tech/vnext/blob/main/docs/runtime/manual-attribute-index-maintenance.md). + +**Tests (Node 18+ test runner):** + +```bash +node --test test/indexes.test.js +# Use only a disposable PostgreSQL database; the test owns uniquely named fixture schemas. +VNEXT_INDEX_TEST_URL=postgresql://user:password@localhost:5432/disposable_test \ + node --test test/indexes.postgres.test.js +``` + +--- + ### `wf check` **Purpose**: System health check diff --git a/bin/workflow.js b/bin/workflow.js index 615ad84..0e97a6f 100755 --- a/bin/workflow.js +++ b/bin/workflow.js @@ -82,6 +82,15 @@ program .description('Reset workflows (force update)') .action(resetCommand); +// Offline index maintenance artifacts; execution is owned by the DBA team. +program.command('indexes').description('Generate DBA-reviewed attribute index SQL') + .command('generate') + .description('Read local workflow/Master definitions and write SQL files without contacting API/DB') + .option('--flow ', 'Generate for one workflow key (all its local versions)') + .option('-o, --output ', 'Parent folder for a new immutable SQL batch', 'index-sql') + .option('--retire-obsolete', 'Retire obsolete projections; requires ALL active workflow versions locally') + .action(require('../src/commands/indexes')); + // Config command program .command('config') diff --git a/src/commands/indexes.js b/src/commands/indexes.js new file mode 100644 index 0000000..8460006 --- /dev/null +++ b/src/commands/indexes.js @@ -0,0 +1,45 @@ +const fs = require('fs'); +const path = require('path'); +const { LOG } = require('../lib/ui'); +const { loadPlans, hash } = require('../lib/indexes/definitions'); +const { generateSql, indexes } = require('../lib/indexes/sql'); + +async function generate(options, projectRoot = process.cwd()) { + const plans = await loadPlans(projectRoot, options.flow); + // Validate/render the whole batch before creating any output; never overwrite an earlier batch. + const rendered = plans.map(plan => ({ plan, sql: generateSql(plan, options) })); + const output = path.resolve(projectRoot, options.output || 'index-sql'); + fs.mkdirSync(output, { recursive: true }); + const batch = fs.mkdtempSync(path.join(output, new Date().toISOString().replace(/[:.]/g, '-') + '-')); + const manifest = { formatVersion: 1, physicalContract: 'v1:latest', generatorVersion: require('../../package.json').version, + domain: plans[0].domain, retireObsolete: !!options.retireObsolete, + generatedAt: new Date().toISOString(), flows: [] }; + for (const { plan, sql } of rendered) { + const file = plan.schema + '.sql'; + fs.writeFileSync(path.join(batch, file), sql, { flag: 'wx' }); + manifest.flows.push({ ...plan, projections: plan.projections.map(p => ({ ...p, indexes: indexes(p) })), file, sha256: hash(sql) }); + } + fs.writeFileSync(path.join(batch, 'manifest.json'), JSON.stringify(manifest, null, 2) + '\n', { flag: 'wx' }); + fs.writeFileSync(path.join(batch, 'README.txt'), [ + `Domain: ${manifest.domain}. Generated offline; nothing was executed.`, + 'DBA: review manifest/source versions and each SQL file. Execute files individually in a maintenance window.', + 'Example: psql -X -v ON_ERROR_STOP=1 --dbname= --file=', + 'Scripts lock InstancesData ACCESS EXCLUSIVE and use one transaction per flow. Plan disk/WAL/replica capacity for table rewrites.', + 'A 5s lock_timeout is included; adjust it during DBA review if necessary. Re-run the SAME script after rollback on failure.', + 'Existing correct indexes are adopted; changed CLI-owned indexes are rebuilt. Unmanaged name collisions fail.', + 'No obsolete projection is retired unless --retire-obsolete was explicitly passed. Retirement requires all active versions locally and drained runtime readers/writers.', + 'Stored values/columns are retained on retirement; physical cleanup remains a separate DBA operation.', + 'Runtime reads use AttributeIndexes:Enabled and refresh the catalog after CatalogCacheSeconds (default 30).', + 'Rollback routing with AttributeIndexes:DisabledFlows. Do not drop columns under running readers.', '' + ].join('\n'), { flag: 'wx' }); + return { batch, count: rendered.length }; +} +async function command(options) { + try { + const { batch, count } = await generate(options); + LOG.success(`Generated ${count} SQL file(s): ${batch}`); + LOG.info('No API or database connection was made. Give this batch to your DBA for review and execution.'); + } catch (error) { LOG.error(error.message); process.exitCode = 1; } +} +module.exports = command; +module.exports.generate = generate; diff --git a/src/lib/indexes/definitions.js b/src/lib/indexes/definitions.js new file mode 100644 index 0000000..a0b83a6 --- /dev/null +++ b/src/lib/indexes/definitions.js @@ -0,0 +1,126 @@ +const crypto = require('crypto'); +const fs = require('fs'); +const path = require('path'); +const { discoverComponents, findJsonInComponent } = require('../discover'); +const { loadVnextConfig, getComponentsRootFromConfig, getComponentTypesFromConfig } = require('../vnextConfig'); + +const hash = value => crypto.createHash('sha256').update(value).digest('hex'); +const compare = (a, b) => a < b ? -1 : a > b ? 1 : 0; +const identifier = /^[a-zA-Z_][a-zA-Z0-9_-]*$/; +function physicalSchema(flow) { + if (typeof flow !== 'string' || !identifier.test(flow) || flow.length > 63) + throw new Error(`Invalid flow key for index generation: ${flow}`); + return flow.toLowerCase().replace(/-/g, '_'); +} + +const { parse: parseVersion, bestMatch } = require('./versions'); +function resolveMaster(schemas, reference, domain) { + if (!reference || reference.domain !== domain || reference.flow !== 'sys-schemas') + throw new Error('Master reference must identify a local sys-schemas component (domain, flow, key).'); + const selected = bestMatch(schemas.filter(s => s.domain === domain && s.key === reference.key), reference.version); + if (!selected) throw new Error(`Master ${reference.key}/${reference.version || 'latest'} is missing locally; include the referenced schema version before generating SQL.`); + return selected; +} + +function fieldsFromSchema(root) { + const fields = []; + function visit(node, fieldPath, supported) { + if (Array.isArray(node)) { node.forEach(n => visit(n, fieldPath, false)); return; } + if (!node || typeof node !== 'object') return; + if ('x-indexed' in node && typeof node['x-indexed'] !== 'boolean') + throw new Error(`Field '${fieldPath}': x-indexed must be boolean.`); + if (node['x-indexed'] === true) { + if (!supported || !/^[a-zA-Z][a-zA-Z0-9_]*(\.[a-zA-Z0-9_]+)*$/.test(fieldPath) || + !['string', 'number', 'integer', 'boolean'].includes(node.type) || '$ref' in node) + throw new Error(`Field '${fieldPath}': x-indexed requires an explicit scalar under object properties; arrays, references and conditional schemas are unsupported.`); + if ('x-sortable' in node && typeof node['x-sortable'] !== 'boolean') throw new Error(`Field '${fieldPath}': x-sortable must be boolean.`); + if ('x-filterOperators' in node && (!Array.isArray(node['x-filterOperators']) || node['x-filterOperators'].some(op => typeof op !== 'string'))) + throw new Error(`Field '${fieldPath}': x-filterOperators must be an array of strings.`); + fields.push({ path: fieldPath, type: node.type, format: node.format, + sortable: node['x-sortable'] === true, operators: node['x-filterOperators'] || [] }); + } + for (const [key, child] of Object.entries(node)) { + if (key === 'properties' && child && typeof child === 'object' && !Array.isArray(child)) { + for (const [name, value] of Object.entries(child)) + visit(value, fieldPath ? `${fieldPath}.${name}` : name, supported && (!node.type || node.type === 'object') && !name.includes('.')); + } else if (['$defs', 'definitions', 'patternProperties'].includes(key)) { + Object.values(child || {}).forEach(value => visit(value, fieldPath, false)); + } else if (['items', 'prefixItems', 'allOf', 'anyOf', 'oneOf', 'if', 'then', 'else', 'additionalProperties'].includes(key)) { + visit(child, fieldPath, false); + } + } + } + visit(root, '', true); + return fields; +} +function projectionDefinitions(fields) { + const merged = new Map(); + for (const field of fields) { + const storage = ['number', 'integer'].includes(field.type) ? 'numeric' : field.type === 'string' && field.format === 'date-time' ? 'timestamptz' : 'text'; + const old = merged.get(field.path); + if (old && old.storage !== storage) throw new Error(`Incompatible indexed types across workflow versions: ${field.path}`); + merged.set(field.path, { ...field, storage, sortable: field.sortable || (old && old.sortable) || false, + operators: [...new Set([...(old ? old.operators : []), ...field.operators])].sort(compare) }); + } + return [...merged.values()].sort((a, b) => compare(a.path, b.path)).flatMap(field => + [...new Set(['text', field.storage])].map(storage => { + const key = hash(`v1:latest:${field.path}:${storage}`).slice(0, 24); + return { key, column: `q_${key}`, path: field.path, storage, + sortable: storage === 'text' && field.sortable, + trigram: storage === 'text' && field.operators.some(op => ['contains', 'like', 'startswith', 'endswith'].includes(op.toLowerCase())) }; + })); +} +async function loadPlans(projectRoot, flow) { + const config = loadVnextConfig(projectRoot); + if (!config.domain || !identifier.test(config.domain)) throw new Error('vnext.config.json requires a valid domain.'); + const dirs = await discoverComponents({ + projectRoot, + componentsRoot: getComponentsRootFromConfig(projectRoot, config), + componentTypes: getComponentTypesFromConfig(config) + }); + if (!dirs.workflows || !dirs.schemas) throw new Error('Configured workflows and schemas directories are required.'); + async function read(dir, type) { + const files = (await findJsonInComponent(dir)).sort(compare); + const seen = new Set(); + return files.map(file => { + const data = JSON.parse(fs.readFileSync(file, 'utf8')); + if (data.flow !== type || !data.key || !data.domain) throw new Error(`Invalid ${type} component: ${file}`); + const version = parseVersion(data.version); + const key = JSON.stringify([data.domain, data.key, version.canonical]); + if (seen.has(key)) throw new Error(`Duplicate component identity ${key}: ${file}`); + seen.add(key); + return { ...data, source: path.relative(projectRoot, file), digest: hash(fs.readFileSync(file)) }; + }); + } + const [workflows, schemas] = await Promise.all([read(dirs.workflows, 'sys-flows'), read(dirs.schemas, 'sys-schemas')]); + const schemaOwners = new Map(); + for (const w of workflows.filter(w => w.domain === config.domain)) { + const name = physicalSchema(w.key); + if (schemaOwners.has(name) && schemaOwners.get(name) !== w.key) + throw new Error(`Workflow keys collide in PostgreSQL schema '${name}'.`); + schemaOwners.set(name, w.key); + } + const grouped = new Map(); + for (const w of workflows.filter(w => w.domain === config.domain && (!flow || w.key === flow))) { + if (!grouped.has(w.key)) grouped.set(w.key, []); + grouped.get(w.key).push(w); + } + if (!grouped.size) throw new Error(`No local workflows found${flow ? ` for '${flow}'` : ''}.`); + const physical = new Set(); + return [...grouped.entries()].sort(([a], [b]) => compare(a, b)).map(([key, versions]) => { + const schema = physicalSchema(key); + if (physical.has(schema)) throw new Error(`Workflow keys collide in PostgreSQL schema '${schema}'.`); + physical.add(schema); + const sources = new Map(), fields = []; + for (const w of versions) { + sources.set(w.source, { file: w.source, sha256: w.digest, key: w.key, version: w.version }); + if (!w.attributes || !w.attributes.schema) continue; + const master = resolveMaster(schemas, w.attributes.schema, config.domain); + if (!master.attributes || !master.attributes.schema) throw new Error(`Master JSON schema is missing in ${master.source}`); + fields.push(...fieldsFromSchema(master.attributes.schema)); + sources.set(master.source, { file: master.source, sha256: master.digest, key: master.key, version: master.version }); + } + return { domain: config.domain, flow: key, schema, projections: projectionDefinitions(fields), sources: [...sources.values()] }; + }); +} +module.exports = { hash, physicalSchema, resolveMaster, fieldsFromSchema, projectionDefinitions, loadPlans }; diff --git a/src/lib/indexes/sql.js b/src/lib/indexes/sql.js new file mode 100644 index 0000000..8ae87bc --- /dev/null +++ b/src/lib/indexes/sql.js @@ -0,0 +1,220 @@ +const { hash, physicalSchema } = require('./definitions'); +const quote = value => '"' + value.replace(/"/g, '""') + '"'; +const literal = value => "'" + value.replace(/'/g, "''") + "'"; +const array = values => `ARRAY[${values.map(literal).join(', ')}]::text[]`; +function expression(p, schema) { + const parts = p.path.split('.'); + const text = parts.length === 1 ? `("Data" ->> ${literal(p.path)})` : `("Data" #>> ${array(parts)})`; + return p.storage === 'numeric' ? `${text}::numeric` : p.storage === 'timestamptz' ? `${quote(schema)}.q_iso_timestamp_v1(${text})` : text; +} +function indexes(p) { + const column = quote(p.column); + const specifications = [{ kind: 'btree', legacy: `ix_${p.key}_btree_v2`, definition: p.storage === 'text' + ? `USING btree (${column}, "InstanceId") WHERE "IsLatest" = true` + : `USING btree (${column}) INCLUDE ("InstanceId") WHERE "IsLatest" = true` }]; + if (p.sortable) specifications.push({ kind: 'desc', legacy: `ix_${p.key}_desc_v2`, definition: `USING btree (${column} DESC, "InstanceId" ASC) WHERE "IsLatest" = true` }); + if (p.trigram) specifications.push({ kind: 'trgm', legacy: `ix_${p.key}_trgm`, definition: `USING gin (${column} COLLATE "tr-TR-x-icu" public.gin_trgm_ops) WHERE "IsLatest" = true` }); + return specifications.map(s => ({ ...s, name: `ix_${p.path.replace(/\./g, '_').toLowerCase().slice(0, 20)}_${s.kind}_${hash(`v1:${p.key}:${s.definition}`).slice(0, 20)}` })); +} + +// The same explicit ISO conversion as runtime physical contract v1:latest. Never wrap a general cast as IMMUTABLE. +const timestampBody = String.raw` +DECLARE m text[]; local_value timestamp; offset_minutes integer := 0; +BEGIN + m := regexp_match(value, '^([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2}(\.[0-9]+)?)(Z|([+-])([0-9]{2}):([0-9]{2}))$'); + IF m IS NULL THEN RAISE EXCEPTION 'Expected ISO-8601 date-time with explicit offset' USING ERRCODE = '22007'; END IF; + IF m[4]::int > 23 OR m[5]::int > 59 OR m[6]::numeric >= 60 + OR COALESCE(m[10]::int, 0) > 15 OR COALESCE(m[11]::int, 0) > 59 THEN + RAISE EXCEPTION 'Invalid ISO-8601 time or offset' USING ERRCODE = '22007'; + END IF; + local_value := make_timestamp(m[1]::int, m[2]::int, m[3]::int, m[4]::int, m[5]::int, m[6]::double precision); + IF m[8] <> 'Z' THEN + offset_minutes := (m[10]::int * 60 + m[11]::int) * CASE WHEN m[9] = '-' THEN -1 ELSE 1 END; + END IF; + RETURN (local_value - make_interval(mins => offset_minutes)) AT TIME ZONE 'UTC'; +END; +`; + +// Canonicalized by PostgreSQL on an empty temporary prototype. Covers includes, collation, +// operator classes, ordering, predicate and access method; names alone never prove readiness. +const helpers = ` +CREATE FUNCTION pg_temp.vnext_index_signature(index_oid oid) RETURNS jsonb +LANGUAGE sql STABLE SET search_path = pg_catalog AS $signature$ + SELECT jsonb_build_object('method', am.amname, 'unique', i.indisunique, + 'keys', i.indnkeyatts, 'attributes', i.indnatts, + 'classes', i.indclass::text, 'collations', i.indcollation::text, 'options', i.indoption::text, + 'definition', (SELECT jsonb_agg(pg_get_indexdef(i.indexrelid, n, false) ORDER BY n) + FROM generate_series(1, i.indnatts) n), + 'predicate', pg_get_expr(i.indpred, i.indrelid), 'reloptions', c.reloptions) + FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid JOIN pg_am am ON am.oid = c.relam + WHERE i.indexrelid = index_oid AND i.indisvalid AND i.indisready AND NOT i.indisexclusion +$signature$; +`; + +function generateSql(plan, { retireObsolete = false } = {}) { + if (physicalSchema(plan.flow) !== plan.schema) throw new Error('Unexpected physical schema.'); + const schema = quote(plan.schema), table = `${schema}."InstancesData"`, catalog = `${schema}."AttributeIndexCatalog"`; + const projections = plan.projections; + const lines = [ + `-- vNext attribute indexes: ${plan.domain}/${plan.flow}`, + '-- Generated offline. Review and execute only in a DBA-approved maintenance window.', + '-- ACCESS EXCLUSIVE lock; stored columns rewrite the table. Transactional, not CONCURRENTLY.', + '-- On any error ROLLBACK the connection. With psql use: psql -X -v ON_ERROR_STOP=1 -f .', + '-- Existing application catalog caches may live for 30s: drain readers/writers before retirement.', + '-- This file never publishes schemas. Confirm the source manifest matches deployed workflow versions.', + 'BEGIN;', "SET LOCAL search_path = pg_catalog, public;", "SET LOCAL standard_conforming_strings = on;", "SET LOCAL lock_timeout = '5s';", + `DO $guard$ BEGIN + IF NOT pg_try_advisory_xact_lock(hashtextextended(${literal('attribute-indexes:' + plan.schema)}, 0)) THEN + RAISE EXCEPTION 'Another attribute-index maintenance transaction is running'; END IF; + IF to_regclass(${literal(table)}) IS NULL THEN RAISE EXCEPTION 'Publish the workflow first: table % does not exist', ${literal(table)}; END IF; +END $guard$;`, + `LOCK TABLE ${table} IN ACCESS EXCLUSIVE MODE;`, + `CREATE TABLE IF NOT EXISTS ${catalog} ( + "Key" text PRIMARY KEY, "ColumnName" text NOT NULL, "PgType" text NOT NULL, + "Indexes" text[] NOT NULL, "Ready" boolean NOT NULL DEFAULT false);` + ]; + if (projections.some(p => p.trigram)) lines.push('CREATE EXTENSION IF NOT EXISTS pg_trgm WITH SCHEMA public;'); + if (projections.some(p => p.storage === 'timestamptz')) { + const fn = `${schema}.q_iso_timestamp_v1`; + // Existing v1 functions must match exactly (whitespace aside); changing the implementation + // in place would silently invalidate stored values, so collisions require a new physical version. + lines.push(`DO $timestamp$ DECLARE existing oid := to_regprocedure(${literal(fn + '(text)')}); BEGIN + IF existing IS NOT NULL AND NOT EXISTS (SELECT 1 FROM pg_proc p JOIN pg_language l ON l.oid=p.prolang + WHERE p.oid=existing AND p.provolatile='i' AND p.proisstrict AND p.proparallel='s' + AND NOT p.prosecdef AND p.prorettype='timestamptz'::regtype AND l.lanname='plpgsql' + AND regexp_replace(p.prosrc, '\\s+', '', 'g') = regexp_replace(${literal(timestampBody)}, '\\s+', '', 'g')) + THEN RAISE EXCEPTION 'Timestamp conversion v1 differs; use a new physical conversion version'; END IF; + IF existing IS NULL THEN EXECUTE ${literal(`CREATE FUNCTION ${fn}(value text) RETURNS timestamptz LANGUAGE plpgsql IMMUTABLE STRICT PARALLEL SAFE AS ${literal(timestampBody)}`)}; END IF; +END $timestamp$;`); + } + lines.push('CREATE TEMP TABLE vnext_index_state (changed boolean) ON COMMIT DROP; INSERT INTO pg_temp.vnext_index_state VALUES (false);', 'CREATE TEMP TABLE vnext_index_prototype ("Data" jsonb, "IsLatest" boolean, "InstanceId" uuid) ON COMMIT DROP;', helpers); + const additionSql = []; + for (const p of projections) { + const expr = expression(p, plan.schema); + const ddl = `${quote(p.column)} ${p.storage} GENERATED ALWAYS AS (CASE WHEN "IsLatest" THEN ${expr} END) STORED`; + lines.push(`ALTER TABLE pg_temp.vnext_index_prototype ADD COLUMN ${ddl};`); + additionSql.push(` + -- ${p.path} (${p.storage}): validate all versions before activating a new projection. + SELECT a.attgenerated, format_type(a.atttypid, a.atttypmod), col_description(a.attrelid,a.attnum), + pg_get_expr(d.adbin,d.adrelid) INTO existing + FROM pg_attribute a LEFT JOIN pg_attrdef d ON d.adrelid=a.attrelid AND d.adnum=a.attnum + WHERE a.attrelid=${literal(table)}::regclass AND a.attname=${literal(p.column)} AND NOT a.attisdropped; + IF FOUND THEN + IF existing.col_description IS DISTINCT FROM ${literal(`vnext-attribute-index:v1:latest:${p.path}:${p.storage}`)} OR + existing.format_type <> ${literal(p.storage === 'timestamptz' ? 'timestamp with time zone' : p.storage)} THEN + RAISE EXCEPTION 'Unrecognized projection collision: %', ${literal(p.path)}; + END IF; + SELECT pg_get_expr(d.adbin,d.adrelid) INTO expected FROM pg_attrdef d JOIN pg_attribute a ON a.attrelid=d.adrelid AND a.attnum=d.adnum + WHERE d.adrelid='pg_temp.vnext_index_prototype'::regclass AND a.attname=${literal(p.column)}; + IF existing.attgenerated = 's' AND existing.pg_get_expr IS DISTINCT FROM expected THEN + RAISE EXCEPTION 'Projection expression changed under a v1 name: %', ${literal(p.path)}; + END IF; + END IF; + IF existing.attgenerated IS DISTINCT FROM 's' THEN + BEGIN PERFORM COUNT(${expr}) FROM ${table}; + EXCEPTION WHEN OTHERS THEN RAISE EXCEPTION 'Cannot index field % as %: % (SQLSTATE %); no changes committed', ${literal(p.path)}, ${literal(p.storage)}, SQLERRM, SQLSTATE; END; + IF existing.format_type IS NOT NULL THEN + -- DROP COLUMN implicitly removes indexes/checks too: reject unexpected dependencies. + IF EXISTS (SELECT 1 FROM pg_depend dep JOIN pg_attribute a + ON a.attrelid=dep.refobjid AND a.attnum=dep.refobjsubid + WHERE a.attrelid=${literal(table)}::regclass AND a.attname=${literal(p.column)} + AND dep.classid IN ('pg_class'::regclass, 'pg_constraint'::regclass)) THEN + RAISE EXCEPTION 'Retired projection % has dependent objects; DBA review is required before reactivation', ${literal(p.path)}; + END IF; + EXECUTE ${literal(`ALTER TABLE ${table} DROP COLUMN ${quote(p.column)}`)}; + END IF; + additions := array_append(additions, ${literal('ADD COLUMN ' + ddl)}); + END IF;`); + } + lines.push(`DO $columns$ DECLARE existing record; expected text; additions text[] := ARRAY[]::text[]; BEGIN +${additionSql.join('\n')} + IF cardinality(additions)>0 THEN UPDATE pg_temp.vnext_index_state SET changed=true; EXECUTE ${literal(`ALTER TABLE ${table} `)} || array_to_string(additions, ', '); END IF; +END $columns$;`); + for (const p of projections) { + lines.push(`COMMENT ON COLUMN ${table}.${quote(p.column)} IS ${literal(`vnext-attribute-index:v1:latest:${p.path}:${p.storage}`)};`); + lines.push(`DO $indexes$ DECLARE desired jsonb; existing_oid oid; existing_name text; selected_names text[] := ARRAY[]::text[]; old_names text[]; old_name text; affected integer; BEGIN + SELECT "Indexes" INTO old_names FROM ${catalog} WHERE "Key"=${literal(p.key)};`); + for (const ix of indexes(p)) { + const marker = `vnext-attribute-index:cli:v1:${p.key}:${ix.kind}`; + lines.push(` + CREATE INDEX ${quote(ix.name)} ON pg_temp.vnext_index_prototype ${ix.definition}; + desired := pg_temp.vnext_index_signature(${literal('pg_temp.' + quote(ix.name))}::regclass); + existing_oid := to_regclass(${literal(schema + '.' + quote(ix.name))}); + IF existing_oid IS NOT NULL AND pg_temp.vnext_index_signature(existing_oid) IS DISTINCT FROM desired THEN + IF NOT EXISTS (SELECT 1 FROM pg_index WHERE indexrelid=existing_oid AND indrelid=${literal(table)}::regclass) + OR obj_description(existing_oid,'pg_class') IS DISTINCT FROM ${literal(marker)} THEN + RAISE EXCEPTION 'Unmanaged index name collision: %', ${literal(ix.name)}; END IF; + EXECUTE ${literal('DROP INDEX ' + schema + '.' + quote(ix.name))}; + RAISE NOTICE 'Rebuilding changed index %', ${literal(ix.name)}; + END IF; + -- Adopt an equivalent existing index (including old ix_ names), avoiding duplicates. + SELECT c.relname INTO existing_name FROM pg_index i JOIN pg_class c ON c.oid=i.indexrelid + WHERE i.indrelid=${literal(table)}::regclass AND pg_temp.vnext_index_signature(i.indexrelid)=desired + ORDER BY (c.relname=${literal(ix.name)}) DESC, c.relname LIMIT 1; + IF existing_name IS NULL THEN + -- Old runtime-created names are owned only when the catalog also references them. + existing_oid := to_regclass(${literal(schema + '.' + quote(ix.legacy))}); + IF existing_oid IS NOT NULL AND EXISTS (SELECT 1 FROM ${catalog} WHERE "Key"=${literal(p.key)} AND ${literal(ix.legacy)}=ANY("Indexes")) + AND EXISTS (SELECT 1 FROM pg_index WHERE indexrelid=existing_oid AND indrelid=${literal(table)}::regclass) THEN + EXECUTE ${literal('DROP INDEX ' + schema + '.' + quote(ix.legacy))}; + END IF; + EXECUTE ${literal(`CREATE INDEX ${quote(ix.name)} ON ${table} ${ix.definition}`)}; + EXECUTE ${literal(`COMMENT ON INDEX ${schema}.${quote(ix.name)} IS ${literal(marker)}`)}; + UPDATE pg_temp.vnext_index_state SET changed=true; + existing_name := ${literal(ix.name)}; + RAISE NOTICE 'Created index %', existing_name; + ELSE RAISE NOTICE 'Reusing index %', existing_name; + END IF; + selected_names := array_append(selected_names, existing_name); + DROP INDEX pg_temp.${quote(ix.name)}; +`); + } + lines.push(` + -- Remove obsolete indexes only when this tool (or the legacy runtime catalog) owns them. + FOREACH old_name IN ARRAY COALESCE(old_names, ARRAY[]::text[]) LOOP + IF NOT (old_name=ANY(selected_names)) THEN + existing_oid := to_regclass(format('%I.%I', ${literal(plan.schema)}, old_name)); + IF EXISTS (SELECT 1 FROM pg_index WHERE indexrelid=existing_oid AND indrelid=${literal(table)}::regclass) + AND (obj_description(existing_oid,'pg_class') LIKE ${literal(`vnext-attribute-index:cli:v1:${p.key}:%`)} + OR old_name=ANY(${array([`ix_${p.key}_btree_v2`, `ix_${p.key}_desc_v2`, `ix_${p.key}_trgm`])})) THEN + EXECUTE format('DROP INDEX %I.%I',${literal(plan.schema)},old_name); + UPDATE pg_temp.vnext_index_state SET changed=true; + RAISE NOTICE 'Removed obsolete managed index %',old_name; + END IF; + END IF; + END LOOP; +INSERT INTO ${catalog} ("Key","ColumnName","PgType","Indexes","Ready") + VALUES (${literal(p.key)},${literal(p.column)},${literal(p.storage === 'timestamptz' ? 'timestamp with time zone' : p.storage)},selected_names,true) + ON CONFLICT ("Key") DO UPDATE SET "ColumnName"=EXCLUDED."ColumnName", "PgType"=EXCLUDED."PgType", "Indexes"=EXCLUDED."Indexes", "Ready"=true + WHERE (${catalog}."ColumnName",${catalog}."PgType",${catalog}."Indexes",${catalog}."Ready") + IS DISTINCT FROM (EXCLUDED."ColumnName",EXCLUDED."PgType",EXCLUDED."Indexes",EXCLUDED."Ready"); + GET DIAGNOSTICS affected = ROW_COUNT; + IF affected > 0 THEN UPDATE pg_temp.vnext_index_state SET changed=true; END IF; + END $indexes$;`); + } + if (retireObsolete) lines.push(` +-- Explicit retirement: local definitions must include EVERY still-active workflow version. +DO $retire$ DECLARE r record; old_name text; existing_oid oid; BEGIN + FOR r IN SELECT c.* FROM ${catalog} c WHERE NOT (c."Key"=ANY(${array(projections.map(p => p.key))})) AND c."Ready" LOOP + IF r."ColumnName" <> 'q_' || r."Key" OR r."Key" !~ '^[a-f0-9]{24}$' OR NOT EXISTS ( + SELECT 1 FROM pg_attribute a WHERE a.attrelid=${literal(table)}::regclass AND a.attname=r."ColumnName" AND NOT a.attisdropped + AND col_description(a.attrelid,a.attnum) LIKE 'vnext-attribute-index:v1:latest:%') + THEN RAISE EXCEPTION 'Unrecognized retired projection: %',r."ColumnName"; END IF; + EXECUTE format('ALTER TABLE %s ALTER COLUMN %I DROP EXPRESSION IF EXISTS',${literal(table)},r."ColumnName"); + UPDATE ${catalog} SET "Ready"=false WHERE "Key"=r."Key"; + FOREACH old_name IN ARRAY r."Indexes" LOOP + existing_oid := to_regclass(format('%I.%I',${literal(plan.schema)},old_name)); + IF EXISTS (SELECT 1 FROM pg_index WHERE indexrelid=existing_oid AND indrelid=${literal(table)}::regclass) + AND (obj_description(existing_oid,'pg_class') LIKE 'vnext-attribute-index:cli:v1:' || r."Key" || ':%' + OR old_name=ANY(ARRAY['ix_' || r."Key" || '_btree_v2','ix_' || r."Key" || '_desc_v2','ix_' || r."Key" || '_trgm'])) THEN + EXECUTE format('DROP INDEX %I.%I',${literal(plan.schema)},old_name); + END IF; + END LOOP; + UPDATE pg_temp.vnext_index_state SET changed=true; + RAISE NOTICE 'Retired projection %, retained stored values',r."ColumnName"; + END LOOP; +END $retire$;`); + lines.push(`DO $analyze$ BEGIN IF (SELECT changed FROM pg_temp.vnext_index_state) THEN ANALYZE ${table}; END IF; END $analyze$;`, 'DROP FUNCTION pg_temp.vnext_index_signature(oid);', 'COMMIT;', ''); + return lines.join('\n'); +} +module.exports = { generateSql, indexes, expression, timestampBody }; diff --git a/src/lib/indexes/versions.js b/src/lib/indexes/versions.js new file mode 100644 index 0000000..e121e43 --- /dev/null +++ b/src/lib/indexes/versions.js @@ -0,0 +1,39 @@ +// Mirrors vNext InstanceDataVersionComparer: artifact first, package second; +// build metadata ignored, prereleases compared case-insensitively as in the runtime. +const canonical = v => v.split('+')[0].toLowerCase(); +const compare = (a,b) => a < b ? -1 : a > b ? 1 : 0; +function parse(v) { + if (typeof v !== 'string') throw new Error(`Invalid component version: ${v}`); + const normalized=canonical(v); + const [artifact,pkg,...extra]=normalized.split('-pkg.'); + if (extra.length || !/^\d+\.\d+\.\d+(?:-[a-z0-9]+(?:\.[a-z0-9]+)*)?$/.test(artifact) || + (pkg !== undefined && !/^\d+\.\d+\.\d+$/.test(pkg))) throw new Error(`Unsupported component version: ${v}`); + return {canonical:normalized,artifact,package:pkg||null}; +} +function semanticCompare(a,b) { + if (!a || !b) return compare(!!a,!!b); + const [ac,ap]=a.split('-'), [bc,bp]=b.split('-'); + const av=ac.split('.').map(BigInt),bv=bc.split('.').map(BigInt); + for(let i=0;i<3;i++){const c=compare(av[i],bv[i]);if(c)return c;} + if (!ap || !bp) return compare(!ap,!bp); + return compare(ap,bp); +} +function versionCompare(a,b) { + const av=parse(a.version),bv=parse(b.version); + return semanticCompare(av.artifact,bv.artifact)||semanticCompare(av.package,bv.package); +} +function bestMatch(candidates,selector) { + const requested=selector==null?'latest':String(selector).trim().toLowerCase(); + const sorted=[...candidates].sort(versionCompare).reverse(); + if (!requested || requested==='latest') return sorted[0]; + if (requested.includes('-pkg.')) {const full=parse(requested);return sorted.find(c=>parse(c.version).canonical===full.canonical);} + const exact=sorted.find(c=>c.version.toLowerCase()===requested); + if (exact) return exact; + if (/^\d+(\.\d+)?$/.test(requested)) return sorted.find(c=>parse(c.version).artifact.startsWith(requested+'.')); + const wanted=parse(requested).artifact; + return sorted.find(c=>parse(c.version).artifact===wanted) || sorted.find(c=> { + const pkg=parse(c.version).package; + return pkg && semanticCompare(pkg,wanted)===0; + }); +} +module.exports={parse,versionCompare,bestMatch}; diff --git a/test/indexes.postgres.test.js b/test/indexes.postgres.test.js new file mode 100644 index 0000000..1e5b4c9 --- /dev/null +++ b/test/indexes.postgres.test.js @@ -0,0 +1,97 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const { Client } = require('pg'); +const { projectionDefinitions } = require('../src/lib/indexes/definitions'); +const { generateSql, indexes } = require('../src/lib/indexes/sql'); +const connectionString = process.env.VNEXT_INDEX_TEST_URL; + +test('DBA-generated SQL contract on real PostgreSQL', {skip:!connectionString}, async t=>{ + const client=new Client({connectionString}); await client.connect(); + const schemas=[]; + const fields=[{path:'amount',type:'number',operators:['gt'],sortable:true},{path:'name',type:'string',operators:['contains'],sortable:true}, + {path:'nested.when',type:'string',format:'date-time',operators:['between']}]; + const planFor=async(values=fields)=>{ + const schema='cli_test_'+require('crypto').randomBytes(8).toString('hex'); schemas.push(schema); + await client.query(`CREATE SCHEMA "${schema}"; CREATE TABLE "${schema}"."InstancesData" ("InstanceId" uuid, "Data" jsonb, "IsLatest" boolean);`); + return {domain:'test',flow:schema,schema,projections:projectionDefinitions(values)}; + }; + const execute=(plan,extra={})=>client.query(generateSql(plan,extra)); + const failed=async(plan,regex,extra={})=>{await assert.rejects(execute(plan,extra),regex);await client.query('ROLLBACK');}; + const listIndexes=async p=>(await client.query('SELECT c.relname,c.oid::text FROM pg_class c JOIN pg_index i ON i.indexrelid=c.oid WHERE i.indrelid=$1::regclass ORDER BY c.relname',[`"${p.schema}"."InstancesData"`])).rows; + const seed=async(p,value,latest=true)=>client.query(`INSERT INTO "${p.schema}"."InstancesData" VALUES ('00000000-0000-0000-0000-000000000001',$1,$2)`,[value,latest]); + try { + await t.test('replay preserves index OIDs, history is null, timestamp is timezone independent, updates track IsLatest',async()=>{ + const p=await planFor(); const data={amount:10,name:'İstanbul_😀',nested:{when:'2026-01-02T03:00:00+03:00'}}; + await seed(p,data);await seed(p,data,false);await execute(p); + const original=await listIndexes(p);await execute(p);assert.deepEqual(await listIndexes(p),original); + const col=p.projections.find(x=>x.storage==='timestamptz').column; + const amount=p.projections.find(x=>x.storage==='numeric').column; + const rows=(await client.query(`SELECT "IsLatest", "${col}" AS time, "${amount}" AS amount FROM "${p.schema}"."InstancesData" ORDER BY "IsLatest" DESC`)).rows; + assert.equal(rows[0].time.toISOString(),'2026-01-02T00:00:00.000Z');assert.equal(rows[1].time,null);assert.equal(rows[1].amount,null); + await client.query("SET TIME ZONE 'America/New_York'; SET DateStyle = 'SQL, DMY'"); + await seed(p,{...data,nested:{when:'2026-01-01T19:00:00-05:00'}}); + assert.equal((await client.query(`SELECT count(DISTINCT "${col}")::int n FROM "${p.schema}"."InstancesData"`)).rows[0].n,1); + await client.query(`UPDATE "${p.schema}"."InstancesData" SET "IsLatest"=false;`); + assert.equal((await client.query(`SELECT count("${amount}")::int n FROM "${p.schema}"."InstancesData"`)).rows[0].n,0); + await client.query("RESET TIME ZONE; RESET DateStyle"); + }); + await t.test('equivalent legacy names are adopted without another index',async()=>{ + const p=await planFor(fields.slice(0,1));await execute(p);const ix=indexes(p.projections[0])[0]; + await client.query(`ALTER INDEX "${p.schema}"."${ix.name}" RENAME TO "${ix.legacy}"; COMMENT ON INDEX "${p.schema}"."${ix.legacy}" IS NULL`); + const before=await listIndexes(p);await execute(p);assert.deepEqual(await listIndexes(p),before); + const catalog=(await client.query(`SELECT "Indexes" FROM "${p.schema}"."AttributeIndexCatalog" WHERE "Key"=$1`,[p.projections[0].key])).rows[0]; + assert(catalog.Indexes.includes(ix.legacy)); + }); + await t.test('changed managed index is rebuilt and unrelated index remains untouched',async()=>{ + const p=await planFor(fields.slice(0,1));await execute(p);const pr=p.projections[0],ix=indexes(pr)[0]; + const before=await listIndexes(p); + await client.query(`DROP INDEX "${p.schema}"."${ix.name}"; CREATE INDEX "${ix.name}" ON "${p.schema}"."InstancesData" ("${pr.column}") WHERE "IsLatest"=false; COMMENT ON INDEX "${p.schema}"."${ix.name}" IS 'vnext-attribute-index:cli:v1:${pr.key}:${ix.kind}'; CREATE INDEX manual_index ON "${p.schema}"."InstancesData" ("IsLatest");`); + const manual=(await listIndexes(p)).find(x=>x.relname==='manual_index'); + await execute(p);const after=await listIndexes(p); + assert.notEqual(after.find(x=>x.relname===ix.name).oid,before.find(x=>x.relname===ix.name).oid); + assert.deepEqual(after.find(x=>x.relname==='manual_index'),manual); + const stable=after.filter(x=>x.relname!==ix.name&&x.relname!=='manual_index'); assert.deepEqual(stable,before.filter(x=>x.relname!==ix.name)); + }); + await t.test('metadata update adds/removes managed trigram and descending indexes without rewriting columns',async()=>{ + const p=await planFor([{path:'name',type:'string',operators:[],sortable:false}]);await execute(p);const before=await listIndexes(p); + const next={...p,projections:projectionDefinitions([fields[1]])};await execute(next);assert.equal((await listIndexes(p)).length,3); + await execute(p);assert.deepEqual(await listIndexes(p),before); + }); + await t.test('invalid historical numeric value rolls back every addition; corrected replay succeeds',async()=>{ + const p=await planFor(fields.slice(0,1));await seed(p,{amount:'invalid'},false); + await failed(p,/Cannot index field amount/); + assert.equal((await client.query('SELECT to_regclass($1) AS value',[`"${p.schema}"."AttributeIndexCatalog"`])).rows[0].value,null); + await client.query(`UPDATE "${p.schema}"."InstancesData" SET "Data"='{"amount":12}'`);await execute(p); + }); + await t.test('invalid timestamp (missing offset) refuses activation and reports field',async()=>{ + const p=await planFor(fields.slice(2));await seed(p,{nested:{when:'2026-01-01T00:00:00'}}); + await failed(p,/Cannot index field nested.when/); + }); + await t.test('unmanaged index name collision fails without dropping it',async()=>{ + const p=await planFor(fields.slice(0,1));const ix=indexes(p.projections[0])[0]; + await client.query(`CREATE INDEX "${ix.name}" ON "${p.schema}"."InstancesData" ("IsLatest")`); + const before=await listIndexes(p);await failed(p,/Unmanaged index name collision/);assert.deepEqual(await listIndexes(p),before); + }); + await t.test('unrecognized generated column collision rolls back without altering it',async()=>{ + const p=await planFor(fields.slice(0,1));const column=p.projections[0].column; + await client.query(`ALTER TABLE "${p.schema}"."InstancesData" ADD COLUMN "${column}" text;`); + await failed(p,/Unrecognized projection collision/);assert.deepEqual(await listIndexes(p),[]); + }); + await t.test('projection retirement is opt-in; retains values, removes owned indexes and permits type change',async()=>{ + const p=await planFor(fields.slice(0,1));await seed(p,{amount:10});await execute(p); + const next={...p,projections:projectionDefinitions([{path:'amount',type:'string',operators:[]}])};await execute(next); + assert.equal((await client.query(`SELECT count(*)::int n FROM "${p.schema}"."AttributeIndexCatalog" WHERE "Ready"`)).rows[0].n,2); + await execute(next,{retireObsolete:true});assert.equal((await listIndexes(p)).length,1); + await execute(p); // A previously retired numeric projection can be prepared again. + await execute(next,{retireObsolete:true}); + await seed(p,{amount:'text-now'}); + assert.equal((await client.query(`SELECT count(*)::int n FROM "${p.schema}"."AttributeIndexCatalog" WHERE "Ready"`)).rows[0].n,1); + }); + await t.test('session concurrency rejects competing maintenance and releases after rollback',async()=>{ + const p=await planFor(fields.slice(0,1));const lock=new Client({connectionString});await lock.connect(); + try {await lock.query('BEGIN');await lock.query('SELECT pg_advisory_xact_lock(hashtextextended($1,0))',['attribute-indexes:'+p.schema]);await failed(p,/Another attribute-index maintenance/);} + finally {await lock.query('ROLLBACK');await lock.end();} + await execute(p); + }); + } finally {await client.query('ROLLBACK'); for(const schema of schemas) await client.query(`DROP SCHEMA "${schema}" CASCADE`);await client.end();} +}); diff --git a/test/indexes.test.js b/test/indexes.test.js new file mode 100644 index 0000000..fc4edad --- /dev/null +++ b/test/indexes.test.js @@ -0,0 +1,93 @@ +// Node >=18 test runner; the CLI itself retains its existing Node requirement. +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { spawnSync } = require('child_process'); +const { fieldsFromSchema, projectionDefinitions, resolveMaster, physicalSchema, loadPlans } = require('../src/lib/indexes/definitions'); +const { generateSql, indexes } = require('../src/lib/indexes/sql'); +const { generate } = require('../src/commands/indexes'); + +function workspace() { + const root=fs.mkdtempSync(path.join(os.tmpdir(),'vnext-indexes-')); + const write=(file, data)=>{ const target=path.join(root,file);fs.mkdirSync(path.dirname(target),{recursive:true});fs.writeFileSync(target,JSON.stringify(data)); }; + write('vnext.config.json',{domain:'test', paths:{componentsRoot:'components',workflows:'Workflows',schemas:'Schemas'}}); + const master={key:'master',version:'1.0.0',domain:'test',flow:'sys-schemas',attributes:{schema:{type:'object',properties:{amount:{type:'number','x-indexed':true,'x-filterOperators':['gt']}}}}}; + const flow={key:'orders',version:'1.0.0',domain:'test',flow:'sys-flows',attributes:{schema:{key:'master',domain:'test',flow:'sys-schemas',version:'latest'}}}; + write('components/Schemas/master.json',master);write('components/Workflows/orders.json',flow); + return {root,write,master,flow}; +} + +test('physical contract and readable names are stable, case-sensitive paths do not collide',()=>{ + const p=projectionDefinitions([{path:'amount',type:'number',operators:[],sortable:false}]); + assert.equal(p[0].key, require('crypto').createHash('sha256').update('v1:latest:amount:text').digest('hex').slice(0,24)); + assert.deepEqual(p.map(x=>x.storage),['text','numeric']); + const long=projectionDefinitions(['ABC','abc','veryLongField'.repeat(8)].map(path=>({path,type:'string',operators:['startsWith'],sortable:true}))); + const names=long.flatMap(indexes).map(i=>i.name); + assert.equal(new Set(names).size,names.length); assert(names.every(n=>Buffer.byteLength(n)<=63)); + assert.equal(physicalSchema('Order-Flow'),'order_flow'); assert.throws(()=>physicalSchema('x;drop schema y')); +}); +test('nested scalars supported; ambiguous, array, ref, conditional and invalid metadata rejected',()=>{ + assert.equal(fieldsFromSchema({properties:{nested:{type:'object',properties:{value:{type:'boolean','x-indexed':true}}}}})[0].path,'nested.value'); + for(const node of [{type:'array','x-indexed':true},{type:['number','null'],'x-indexed':true},{type:'number','x-indexed':'true'},{type:'number','$ref':'x','x-indexed':true},{type:'number','x-indexed':true,'x-filterOperators':'gt'}]) + assert.throws(()=>fieldsFromSchema({properties:{x:node}})); + assert.throws(()=>fieldsFromSchema({allOf:[{properties:{x:{type:'number','x-indexed':true}}}]})); + assert.throws(()=>fieldsFromSchema({properties:{xs:{type:'array',items:{properties:{x:{type:'number','x-indexed':true}}}}}})); + assert.throws(()=>fieldsFromSchema({properties:{'a.b':{type:'number','x-indexed':true}}})); +}); +test('version resolution is numeric, pinned/local, and fails closed for missing references',()=>{ + const schemas=['1.2.0','1.10.0','2.0.0'].map(version=>({domain:'test',key:'m',version})); + const ref={domain:'test',flow:'sys-schemas',key:'m'}; + for(const [version,expected] of [['latest','2.0.0'],['1','1.10.0'],['1.2','1.2.0'],['1.2.0','1.2.0']]) assert.equal(resolveMaster(schemas,{...ref,version},'test').version,expected); + assert.throws(()=>resolveMaster(schemas,{...ref,version:'3'},'test')); + assert.throws(()=>resolveMaster(schemas,{...ref,domain:'remote'},'test')); + assert.throws(()=>resolveMaster(schemas,{...ref,version:'^1'},'test')); + assert.throws(()=>projectionDefinitions([{path:'x',type:'number',operators:[]},{path:'x',type:'string',operators:[]}])); +}); +test('generation is offline, immutable, scoped, and writes manifests matching SQL checksums',async()=>{ + const w=workspace(); + try { + const {root}=w; + const first=await generate({output:'output'},root), second=await generate({output:'output'},root); + assert.notEqual(first.batch,second.batch); + const manifest=JSON.parse(fs.readFileSync(path.join(first.batch,'manifest.json'))); + assert.equal(manifest.flows.length,1);assert.equal(manifest.flows[0].sources.length,2); + const sql=fs.readFileSync(path.join(first.batch,'orders.sql'),'utf8'); + assert.equal(manifest.flows[0].sha256,require('crypto').createHash('sha256').update(sql).digest('hex')); + assert.equal(sql,fs.readFileSync(path.join(second.batch,'orders.sql'),'utf8')); + assert.doesNotMatch(sql,/current_database\(\)/); + assert.equal(Object.hasOwn(manifest, 'database'), false); + // The real CLI runs with network access forbidden. No profile or running API/DB is needed. + const preload=path.join(root,'no-network.js');fs.writeFileSync(preload,"for (const name of ['net','tls','http','https']) { const m=require(name); for(const k of ['connect','createConnection','request','get']) if(m[k]) m[k]=()=>{throw Error('NETWORK FORBIDDEN')}; }"); + const result=spawnSync(process.execPath,['--require',preload,path.resolve(__dirname,'../bin/workflow.js'),'indexes','generate','--flow','orders','--output','from-command'],{cwd:root,encoding:'utf8'}); + assert.equal(result.status,0,result.stdout+result.stderr); + assert.match(result.stdout,/No API or database connection/); + await assert.rejects(loadPlans(root,'missing')); + } finally {fs.rmSync(w.root,{recursive:true,force:true});} +}); +test('all local workflow versions are merged; duplicate identities and schema collisions fail',async()=>{ + const w=workspace(); + try { + w.write('components/Schemas/master2.json',{...w.master,version:'2.0.0',attributes:{schema:{properties:{name:{type:'string','x-indexed':true}}}}}); + w.write('components/Workflows/orders.json',{...w.flow,attributes:{schema:{...w.flow.attributes.schema,version:'1.0.0'}}}); + w.write('components/Workflows/orders2.json',{...w.flow,version:'2.0.0'}); + assert.deepEqual((await loadPlans(w.root))[0].projections.map(p=>p.path),['amount','amount','name']); + w.write('components/Workflows/duplicate.json',w.flow); + await assert.rejects(loadPlans(w.root),/Duplicate/); + } finally {fs.rmSync(w.root,{recursive:true,force:true});} +}); +test('SQL needs no database option; retirement is explicit',()=>{ + const plan={domain:'test',flow:'orders',schema:'orders',projections:[]}; + assert.doesNotMatch(generateSql(plan),/current_database|Wrong database/); + assert.doesNotMatch(generateSql(plan),/DO \$retire\$/); + assert.match(generateSql(plan,{retireObsolete:true}),/DO \$retire\$/); +}); + +test('package revisions match runtime selectors and ignore build metadata',()=>{ + const ref={domain:'test',flow:'sys-schemas',key:'m'}; + const schemas=['1.0.0-pkg.1.2.0+core','1.0.0-pkg.1.10.0+core','2.0.0-pkg.0.1.0+core'].map(version=>({domain:'test',key:'m',version})); + for(const [version,expected] of [['1.0.0','1.0.0-pkg.1.10.0+core'],['1.0.0-pkg.1.2.0','1.0.0-pkg.1.2.0+core'],['latest','2.0.0-pkg.0.1.0+core'],['1.2.0','1.0.0-pkg.1.2.0+core']]) + assert.equal(resolveMaster(schemas,{...ref,version},'test').version,expected); + assert.throws(()=>resolveMaster(schemas,{...ref,version:'1.0.0-pkg.1.3.0'},'test')); +}); From 5fdf7d7476d7339e15566ebb0368d3d46e6a80e5 Mon Sep 17 00:00:00 2001 From: Baran Sekin Date: Mon, 14 Sep 2026 13:31:22 +0300 Subject: [PATCH 2/4] fix(indexes): reject conditional schemas for indexed fields --- src/lib/indexes/definitions.js | 5 +++-- test/indexes.test.js | 11 +++++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/lib/indexes/definitions.js b/src/lib/indexes/definitions.js index a0b83a6..bb9edcb 100644 --- a/src/lib/indexes/definitions.js +++ b/src/lib/indexes/definitions.js @@ -27,6 +27,7 @@ function fieldsFromSchema(root) { function visit(node, fieldPath, supported) { if (Array.isArray(node)) { node.forEach(n => visit(n, fieldPath, false)); return; } if (!node || typeof node !== 'object') return; + supported = supported && !['$ref', 'allOf', 'anyOf', 'oneOf', 'not', 'if', 'then', 'else', 'dependentSchemas'].some(key => key in node); if ('x-indexed' in node && typeof node['x-indexed'] !== 'boolean') throw new Error(`Field '${fieldPath}': x-indexed must be boolean.`); if (node['x-indexed'] === true) { @@ -43,9 +44,9 @@ function fieldsFromSchema(root) { if (key === 'properties' && child && typeof child === 'object' && !Array.isArray(child)) { for (const [name, value] of Object.entries(child)) visit(value, fieldPath ? `${fieldPath}.${name}` : name, supported && (!node.type || node.type === 'object') && !name.includes('.')); - } else if (['$defs', 'definitions', 'patternProperties'].includes(key)) { + } else if (['$defs', 'definitions', 'patternProperties', 'dependentSchemas'].includes(key)) { Object.values(child || {}).forEach(value => visit(value, fieldPath, false)); - } else if (['items', 'prefixItems', 'allOf', 'anyOf', 'oneOf', 'if', 'then', 'else', 'additionalProperties'].includes(key)) { + } else if (['items', 'prefixItems', 'allOf', 'anyOf', 'oneOf', 'if', 'then', 'else', 'not', 'additionalProperties'].includes(key)) { visit(child, fieldPath, false); } } diff --git a/test/indexes.test.js b/test/indexes.test.js index fc4edad..3540c9b 100644 --- a/test/indexes.test.js +++ b/test/indexes.test.js @@ -91,3 +91,14 @@ test('package revisions match runtime selectors and ignore build metadata',()=>{ assert.equal(resolveMaster(schemas,{...ref,version},'test').version,expected); assert.throws(()=>resolveMaster(schemas,{...ref,version:'1.0.0-pkg.1.3.0'},'test')); }); + + +test('conditional indexed nodes and ancestors are rejected without rejecting unrelated unindexed fields',()=>{ + for (const keyword of ['allOf','anyOf','oneOf','not','if','then','else','dependentSchemas']) { + const condition=['allOf','anyOf','oneOf'].includes(keyword)?[{}]:{}; + assert.throws(()=>fieldsFromSchema({properties:{amount:{type:'number','x-indexed':true,[keyword]:condition}}}),/amount/); + assert.throws(()=>fieldsFromSchema({[keyword]:condition,properties:{amount:{type:'number','x-indexed':true}}}),/amount/); + } + const fields=fieldsFromSchema({properties:{amount:{type:'number','x-indexed':true},other:{type:'string',oneOf:[{maxLength:10}]}}}); + assert.deepEqual(fields.map(field=>field.path),['amount']); +}); From 1f89df1cc107edc439bd24b87ba2d39d5ccde957 Mon Sep 17 00:00:00 2001 From: Baran Sekin Date: Mon, 14 Sep 2026 17:41:02 +0300 Subject: [PATCH 3/4] feat(indexes): restrict SQL generation to explicit master schemas --- README.md | 35 +++++++++++++++ src/commands/indexes.js | 1 + src/lib/indexes/definitions.js | 41 ++++++++++++----- test/indexes.test.js | 80 ++++++++++++++++++++++++++++++---- 4 files changed, 137 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 86bf5eb..3f56645 100644 --- a/README.md +++ b/README.md @@ -888,3 +888,38 @@ npm run dev ## 📝 License MIT License - see [LICENSE](LICENSE) for details. + +### Schema purpose for index generation + +Schema components may declare `type` at the document root (alongside `key`, `domain`, and +`flow`): `master`, `transition`, `view`, or `function`. Unknown non-empty values fail validation. +The field is optional and has no default. Missing, null or blank values never mean master. +This is independent of the existing `attributes.type` and JSON Schema `type` keywords. +Only a referenced `type: master` schema contributes index SQL. Other schema purposes are skipped; +`x-indexed` (including `false`) is invalid in their schema nodes. A latest reference resolving to a +non-master schema does not fall back to an older master version. No matching masters means no SQL batch. +Schemas using `x-indexed` must explicitly declare root `type: master` before publication or SQL generation. + +```json +{ + "key": "order-master", + "domain": "sales", + "flow": "sys-schemas", + "version": "1.0.0", + "flowVersion": "1.0.0", + "tags": ["orders"], + "type": "master", + "attributes": { + "type": "workflow", + "schema": { + "type": "object", + "properties": { + "amount": { "type": "number", "x-indexed": true, "x-filterOperators": ["gt"] } + } + } + } +} +``` + +The envelope purpose is validated during publication (including schema seed items); it does not +replace `SchemaDefinition.Type`, which continues to represent `attributes.type`. diff --git a/src/commands/indexes.js b/src/commands/indexes.js index 8460006..e4e034d 100644 --- a/src/commands/indexes.js +++ b/src/commands/indexes.js @@ -6,6 +6,7 @@ const { generateSql, indexes } = require('../lib/indexes/sql'); async function generate(options, projectRoot = process.cwd()) { const plans = await loadPlans(projectRoot, options.flow); + if (!plans.length) throw new Error("No workflows referencing type: master schemas found; no SQL files generated."); // Validate/render the whole batch before creating any output; never overwrite an earlier batch. const rendered = plans.map(plan => ({ plan, sql: generateSql(plan, options) })); const output = path.resolve(projectRoot, options.output || 'index-sql'); diff --git a/src/lib/indexes/definitions.js b/src/lib/indexes/definitions.js index bb9edcb..cb93b9c 100644 --- a/src/lib/indexes/definitions.js +++ b/src/lib/indexes/definitions.js @@ -22,23 +22,27 @@ function resolveMaster(schemas, reference, domain) { return selected; } -function fieldsFromSchema(root) { +function fieldsFromSchema(root, schemaType) { const fields = []; function visit(node, fieldPath, supported) { if (Array.isArray(node)) { node.forEach(n => visit(n, fieldPath, false)); return; } if (!node || typeof node !== 'object') return; supported = supported && !['$ref', 'allOf', 'anyOf', 'oneOf', 'not', 'if', 'then', 'else', 'dependentSchemas'].some(key => key in node); + if ('x-indexed' in node && schemaType !== 'master') + throw new Error(`Field '${fieldPath}': x-indexed is only allowed when root.type is 'master'.`); if ('x-indexed' in node && typeof node['x-indexed'] !== 'boolean') throw new Error(`Field '${fieldPath}': x-indexed must be boolean.`); if (node['x-indexed'] === true) { if (!supported || !/^[a-zA-Z][a-zA-Z0-9_]*(\.[a-zA-Z0-9_]+)*$/.test(fieldPath) || - !['string', 'number', 'integer', 'boolean'].includes(node.type) || '$ref' in node) + !['string', 'number', 'integer', 'boolean'].includes(node.type) || '$ref' in node) throw new Error(`Field '${fieldPath}': x-indexed requires an explicit scalar under object properties; arrays, references and conditional schemas are unsupported.`); if ('x-sortable' in node && typeof node['x-sortable'] !== 'boolean') throw new Error(`Field '${fieldPath}': x-sortable must be boolean.`); if ('x-filterOperators' in node && (!Array.isArray(node['x-filterOperators']) || node['x-filterOperators'].some(op => typeof op !== 'string'))) throw new Error(`Field '${fieldPath}': x-filterOperators must be an array of strings.`); - fields.push({ path: fieldPath, type: node.type, format: node.format, - sortable: node['x-sortable'] === true, operators: node['x-filterOperators'] || [] }); + fields.push({ + path: fieldPath, type: node.type, format: node.format, + sortable: node['x-sortable'] === true, operators: node['x-filterOperators'] || [] + }); } for (const [key, child] of Object.entries(node)) { if (key === 'properties' && child && typeof child === 'object' && !Array.isArray(child)) { @@ -46,7 +50,7 @@ function fieldsFromSchema(root) { visit(value, fieldPath ? `${fieldPath}.${name}` : name, supported && (!node.type || node.type === 'object') && !name.includes('.')); } else if (['$defs', 'definitions', 'patternProperties', 'dependentSchemas'].includes(key)) { Object.values(child || {}).forEach(value => visit(value, fieldPath, false)); - } else if (['items', 'prefixItems', 'allOf', 'anyOf', 'oneOf', 'if', 'then', 'else', 'not', 'additionalProperties'].includes(key)) { + } else if (['items', 'prefixItems', 'allOf', 'anyOf', 'oneOf', 'if', 'then', 'else', 'not', 'additionalProperties', 'contains', 'propertyNames', 'additionalItems', 'unevaluatedProperties', 'unevaluatedItems'].includes(key)) { visit(child, fieldPath, false); } } @@ -60,15 +64,19 @@ function projectionDefinitions(fields) { const storage = ['number', 'integer'].includes(field.type) ? 'numeric' : field.type === 'string' && field.format === 'date-time' ? 'timestamptz' : 'text'; const old = merged.get(field.path); if (old && old.storage !== storage) throw new Error(`Incompatible indexed types across workflow versions: ${field.path}`); - merged.set(field.path, { ...field, storage, sortable: field.sortable || (old && old.sortable) || false, - operators: [...new Set([...(old ? old.operators : []), ...field.operators])].sort(compare) }); + merged.set(field.path, { + ...field, storage, sortable: field.sortable || (old && old.sortable) || false, + operators: [...new Set([...(old ? old.operators : []), ...field.operators])].sort(compare) + }); } return [...merged.values()].sort((a, b) => compare(a.path, b.path)).flatMap(field => [...new Set(['text', field.storage])].map(storage => { const key = hash(`v1:latest:${field.path}:${storage}`).slice(0, 24); - return { key, column: `q_${key}`, path: field.path, storage, + return { + key, column: `q_${key}`, path: field.path, storage, sortable: storage === 'text' && field.sortable, - trigram: storage === 'text' && field.operators.some(op => ['contains', 'like', 'startswith', 'endswith'].includes(op.toLowerCase())) }; + trigram: storage === 'text' && field.operators.some(op => ['contains', 'like', 'startswith', 'endswith'].includes(op.toLowerCase())) + }; })); } async function loadPlans(projectRoot, flow) { @@ -86,6 +94,13 @@ async function loadPlans(projectRoot, flow) { return files.map(file => { const data = JSON.parse(fs.readFileSync(file, 'utf8')); if (data.flow !== type || !data.key || !data.domain) throw new Error(`Invalid ${type} component: ${file}`); + if (type === 'sys-schemas') { + const schemaType = data.type; + if (schemaType != null && !(typeof schemaType === 'string' && schemaType.trim() === '') && + !['master', 'transition', 'view', 'function'].includes(schemaType)) + throw new Error(`Schema type must be one of: master, transition, view, function (${file}).`); + if (schemaType !== 'master') fieldsFromSchema(data.attributes && data.attributes.schema, schemaType); + } const version = parseVersion(data.version); const key = JSON.stringify([data.domain, data.key, version.canonical]); if (seen.has(key)) throw new Error(`Duplicate component identity ${key}: ${file}`); @@ -113,15 +128,19 @@ async function loadPlans(projectRoot, flow) { if (physical.has(schema)) throw new Error(`Workflow keys collide in PostgreSQL schema '${schema}'.`); physical.add(schema); const sources = new Map(), fields = []; + let hasMaster = false; for (const w of versions) { sources.set(w.source, { file: w.source, sha256: w.digest, key: w.key, version: w.version }); if (!w.attributes || !w.attributes.schema) continue; const master = resolveMaster(schemas, w.attributes.schema, config.domain); + if (master.type !== 'master') continue; + hasMaster = true; if (!master.attributes || !master.attributes.schema) throw new Error(`Master JSON schema is missing in ${master.source}`); - fields.push(...fieldsFromSchema(master.attributes.schema)); + fields.push(...fieldsFromSchema(master.attributes.schema, master.type)); sources.set(master.source, { file: master.source, sha256: master.digest, key: master.key, version: master.version }); } + if (!hasMaster) return null; return { domain: config.domain, flow: key, schema, projections: projectionDefinitions(fields), sources: [...sources.values()] }; - }); + }).filter(Boolean); } module.exports = { hash, physicalSchema, resolveMaster, fieldsFromSchema, projectionDefinitions, loadPlans }; diff --git a/test/indexes.test.js b/test/indexes.test.js index 3540c9b..dc7027a 100644 --- a/test/indexes.test.js +++ b/test/indexes.test.js @@ -13,7 +13,7 @@ function workspace() { const root=fs.mkdtempSync(path.join(os.tmpdir(),'vnext-indexes-')); const write=(file, data)=>{ const target=path.join(root,file);fs.mkdirSync(path.dirname(target),{recursive:true});fs.writeFileSync(target,JSON.stringify(data)); }; write('vnext.config.json',{domain:'test', paths:{componentsRoot:'components',workflows:'Workflows',schemas:'Schemas'}}); - const master={key:'master',version:'1.0.0',domain:'test',flow:'sys-schemas',attributes:{schema:{type:'object',properties:{amount:{type:'number','x-indexed':true,'x-filterOperators':['gt']}}}}}; + const master={type:'master',key:'master',version:'1.0.0',domain:'test',flow:'sys-schemas',attributes:{type:'workflow',schema:{type:'object',properties:{amount:{type:'number','x-indexed':true,'x-filterOperators':['gt']}}}}}; const flow={key:'orders',version:'1.0.0',domain:'test',flow:'sys-flows',attributes:{schema:{key:'master',domain:'test',flow:'sys-schemas',version:'latest'}}}; write('components/Schemas/master.json',master);write('components/Workflows/orders.json',flow); return {root,write,master,flow}; @@ -29,12 +29,12 @@ test('physical contract and readable names are stable, case-sensitive paths do n assert.equal(physicalSchema('Order-Flow'),'order_flow'); assert.throws(()=>physicalSchema('x;drop schema y')); }); test('nested scalars supported; ambiguous, array, ref, conditional and invalid metadata rejected',()=>{ - assert.equal(fieldsFromSchema({properties:{nested:{type:'object',properties:{value:{type:'boolean','x-indexed':true}}}}})[0].path,'nested.value'); + assert.equal(fieldsFromSchema({properties:{nested:{type:'object',properties:{value:{type:'boolean','x-indexed':true}}}}}, 'master')[0].path,'nested.value'); for(const node of [{type:'array','x-indexed':true},{type:['number','null'],'x-indexed':true},{type:'number','x-indexed':'true'},{type:'number','$ref':'x','x-indexed':true},{type:'number','x-indexed':true,'x-filterOperators':'gt'}]) - assert.throws(()=>fieldsFromSchema({properties:{x:node}})); - assert.throws(()=>fieldsFromSchema({allOf:[{properties:{x:{type:'number','x-indexed':true}}}]})); - assert.throws(()=>fieldsFromSchema({properties:{xs:{type:'array',items:{properties:{x:{type:'number','x-indexed':true}}}}}})); - assert.throws(()=>fieldsFromSchema({properties:{'a.b':{type:'number','x-indexed':true}}})); + assert.throws(()=>fieldsFromSchema({properties:{x:node}}, 'master')); + assert.throws(()=>fieldsFromSchema({allOf:[{properties:{x:{type:'number','x-indexed':true}}}]}, 'master')); + assert.throws(()=>fieldsFromSchema({properties:{xs:{type:'array',items:{properties:{x:{type:'number','x-indexed':true}}}}}}, 'master')); + assert.throws(()=>fieldsFromSchema({properties:{'a.b':{type:'number','x-indexed':true}}}, 'master')); }); test('version resolution is numeric, pinned/local, and fails closed for missing references',()=>{ const schemas=['1.2.0','1.10.0','2.0.0'].map(version=>({domain:'test',key:'m',version})); @@ -96,9 +96,71 @@ test('package revisions match runtime selectors and ignore build metadata',()=>{ test('conditional indexed nodes and ancestors are rejected without rejecting unrelated unindexed fields',()=>{ for (const keyword of ['allOf','anyOf','oneOf','not','if','then','else','dependentSchemas']) { const condition=['allOf','anyOf','oneOf'].includes(keyword)?[{}]:{}; - assert.throws(()=>fieldsFromSchema({properties:{amount:{type:'number','x-indexed':true,[keyword]:condition}}}),/amount/); - assert.throws(()=>fieldsFromSchema({[keyword]:condition,properties:{amount:{type:'number','x-indexed':true}}}),/amount/); + assert.throws(()=>fieldsFromSchema({properties:{amount:{type:'number','x-indexed':true,[keyword]:condition}}}, 'master'),/amount/); + assert.throws(()=>fieldsFromSchema({[keyword]:condition,properties:{amount:{type:'number','x-indexed':true}}}, 'master'),/amount/); } - const fields=fieldsFromSchema({properties:{amount:{type:'number','x-indexed':true},other:{type:'string',oneOf:[{maxLength:10}]}}}); + const fields=fieldsFromSchema({properties:{amount:{type:'number','x-indexed':true},other:{type:'string',oneOf:[{maxLength:10}]}}}, 'master'); assert.deepEqual(fields.map(field=>field.path),['amount']); }); + + +test('only referenced master schemas produce SQL plans; skip non-master latest versions without falling back',async()=>{ + const w=workspace(); + try { + for(const type of ['transition','view','function']) { + w.write('components/Schemas/master.json',{...w.master,type,attributes:{schema:{type:'object',properties:{name:{type:'string'}}}}}); + assert.deepEqual(await loadPlans(w.root),[]); + await assert.rejects(generate({output:'ignored-output'},w.root),/No workflows referencing type: master/); + assert.equal(fs.existsSync(path.join(w.root,'ignored-output')),false); + } + w.write('components/Schemas/master.json',w.master); + w.write('components/Schemas/next.json',{...w.master,version:'2.0.0',type:'transition',attributes:{schema:{type:'object'}}}); + assert.deepEqual(await loadPlans(w.root),[]); + w.write('components/Workflows/orders.json',{...w.flow,attributes:{schema:{...w.flow.attributes.schema,version:'1.0.0'}}}); + assert.equal((await loadPlans(w.root)).length,1); + } finally {fs.rmSync(w.root,{recursive:true,force:true});} +}); + +test('non-master x-indexed metadata and invalid schema purposes fail validation',async()=>{ + const w=workspace(); + try { + for(const type of ['transition','view','function']) { + for(const indexed of [true,false]) { + w.write('components/Schemas/master.json',{...w.master,type,attributes:{schema:{properties:{nested:{properties:{value:{type:'number','x-indexed':indexed}}}}}}}); + await assert.rejects(loadPlans(w.root),/x-indexed is only allowed.*master/); + } + } + for(const type of ['workflow','json-schema','MASTER',42,{},[]]) { + w.write('components/Schemas/master.json',{...w.master,type,attributes:{schema:{type:'object'}}}); + await assert.rejects(loadPlans(w.root),/Schema type must be one of/); + } + } finally {fs.rmSync(w.root,{recursive:true,force:true});} +}); + + +test('root type controls indexing while existing attributes.type remains independent',async()=>{ + const w=workspace(); + try { + for(const attributeType of ['workflow','task','function','view','schema','extension','headers','json-schema']) { + w.write('components/Schemas/master.json',{...w.master,attributes:{...w.master.attributes,type:attributeType}}); + assert.equal((await loadPlans(w.root)).length,1); + } + w.write('components/Schemas/master.json',{...w.master,type:'view',attributes:{type:'master',schema:{type:'object'}}}); + assert.deepEqual(await loadPlans(w.root),[]); + w.write('components/Schemas/master.json',{...w.master,type:undefined,attributes:{...w.master.attributes,type:'master'}}); + await assert.rejects(loadPlans(w.root),/x-indexed is only allowed/); + } finally {fs.rmSync(w.root,{recursive:true,force:true});} +}); + + +test('optional root purpose has no default and never opts into indexing',async()=>{ + const w=workspace(); + try { + for(const type of [undefined,null,'',' ']) { + w.write('components/Schemas/master.json',{...w.master,type,attributes:{type:'master',schema:{type:'object'}}}); + assert.deepEqual(await loadPlans(w.root),[]); + w.write('components/Schemas/master.json',{...w.master,type}); + await assert.rejects(loadPlans(w.root),/x-indexed is only allowed/); + } + } finally {fs.rmSync(w.root,{recursive:true,force:true});} +}); From c5380ea9286748b97f8144dab018c66743a5e1d7 Mon Sep 17 00:00:00 2001 From: Baran Sekin Date: Tue, 15 Sep 2026 09:44:21 +0300 Subject: [PATCH 4/4] fix(indexes): select master schemas using attributes type --- README.md | 22 +++++++-------- src/commands/indexes.js | 2 +- src/lib/indexes/definitions.js | 13 ++++----- test/indexes.test.js | 50 ++++++++++++---------------------- 4 files changed, 35 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index 3f56645..6e3aa3c 100644 --- a/README.md +++ b/README.md @@ -891,14 +891,13 @@ MIT License - see [LICENSE](LICENSE) for details. ### Schema purpose for index generation -Schema components may declare `type` at the document root (alongside `key`, `domain`, and -`flow`): `master`, `transition`, `view`, or `function`. Unknown non-empty values fail validation. -The field is optional and has no default. Missing, null or blank values never mean master. -This is independent of the existing `attributes.type` and JSON Schema `type` keywords. -Only a referenced `type: master` schema contributes index SQL. Other schema purposes are skipped; -`x-indexed` (including `false`) is invalid in their schema nodes. A latest reference resolving to a -non-master schema does not fall back to an older master version. No matching masters means no SQL batch. -Schemas using `x-indexed` must explicitly declare root `type: master` before publication or SQL generation. +Schema components use the existing `attributes.type` string. Its values are not restricted to an enum; +legacy and custom types remain valid. Only the exact value `master` permits `x-indexed` metadata +(including `false`) and contributes index SQL. Missing, null, blank or other values never mean master. +The existing publication requirement for a non-empty schema type is unchanged. JSON Schema `type` +keywords within `attributes.schema` retain their existing meaning. +A latest reference resolving to a non-master schema does not fall back to an older master version. +No matching masters means no SQL batch. There is no separate component root `type` field. ```json { @@ -908,9 +907,8 @@ Schemas using `x-indexed` must explicitly declare root `type: master` before pub "version": "1.0.0", "flowVersion": "1.0.0", "tags": ["orders"], - "type": "master", "attributes": { - "type": "workflow", + "type": "master", "schema": { "type": "object", "properties": { @@ -921,5 +919,5 @@ Schemas using `x-indexed` must explicitly declare root `type: master` before pub } ``` -The envelope purpose is validated during publication (including schema seed items); it does not -replace `SchemaDefinition.Type`, which continues to represent `attributes.type`. +The schema validator reads `SchemaDefinition.Type` (`attributes.type`) during publication, including +schema seed items. No extra envelope metadata is required. diff --git a/src/commands/indexes.js b/src/commands/indexes.js index e4e034d..47c076d 100644 --- a/src/commands/indexes.js +++ b/src/commands/indexes.js @@ -6,7 +6,7 @@ const { generateSql, indexes } = require('../lib/indexes/sql'); async function generate(options, projectRoot = process.cwd()) { const plans = await loadPlans(projectRoot, options.flow); - if (!plans.length) throw new Error("No workflows referencing type: master schemas found; no SQL files generated."); + if (!plans.length) throw new Error("No workflows referencing attributes.type: master schemas found; no SQL files generated."); // Validate/render the whole batch before creating any output; never overwrite an earlier batch. const rendered = plans.map(plan => ({ plan, sql: generateSql(plan, options) })); const output = path.resolve(projectRoot, options.output || 'index-sql'); diff --git a/src/lib/indexes/definitions.js b/src/lib/indexes/definitions.js index cb93b9c..a743087 100644 --- a/src/lib/indexes/definitions.js +++ b/src/lib/indexes/definitions.js @@ -29,7 +29,7 @@ function fieldsFromSchema(root, schemaType) { if (!node || typeof node !== 'object') return; supported = supported && !['$ref', 'allOf', 'anyOf', 'oneOf', 'not', 'if', 'then', 'else', 'dependentSchemas'].some(key => key in node); if ('x-indexed' in node && schemaType !== 'master') - throw new Error(`Field '${fieldPath}': x-indexed is only allowed when root.type is 'master'.`); + throw new Error(`Field '${fieldPath}': x-indexed is only allowed when attributes.type is 'master'.`); if ('x-indexed' in node && typeof node['x-indexed'] !== 'boolean') throw new Error(`Field '${fieldPath}': x-indexed must be boolean.`); if (node['x-indexed'] === true) { @@ -95,10 +95,9 @@ async function loadPlans(projectRoot, flow) { const data = JSON.parse(fs.readFileSync(file, 'utf8')); if (data.flow !== type || !data.key || !data.domain) throw new Error(`Invalid ${type} component: ${file}`); if (type === 'sys-schemas') { - const schemaType = data.type; - if (schemaType != null && !(typeof schemaType === 'string' && schemaType.trim() === '') && - !['master', 'transition', 'view', 'function'].includes(schemaType)) - throw new Error(`Schema type must be one of: master, transition, view, function (${file}).`); + const schemaType = data.attributes && data.attributes.type; + if (schemaType != null && typeof schemaType !== 'string') + throw new Error(`Schema attributes.type must be a string (${file}).`); if (schemaType !== 'master') fieldsFromSchema(data.attributes && data.attributes.schema, schemaType); } const version = parseVersion(data.version); @@ -133,10 +132,10 @@ async function loadPlans(projectRoot, flow) { sources.set(w.source, { file: w.source, sha256: w.digest, key: w.key, version: w.version }); if (!w.attributes || !w.attributes.schema) continue; const master = resolveMaster(schemas, w.attributes.schema, config.domain); - if (master.type !== 'master') continue; + if (!master.attributes || master.attributes.type !== 'master') continue; hasMaster = true; if (!master.attributes || !master.attributes.schema) throw new Error(`Master JSON schema is missing in ${master.source}`); - fields.push(...fieldsFromSchema(master.attributes.schema, master.type)); + fields.push(...fieldsFromSchema(master.attributes.schema, master.attributes.type)); sources.set(master.source, { file: master.source, sha256: master.digest, key: master.key, version: master.version }); } if (!hasMaster) return null; diff --git a/test/indexes.test.js b/test/indexes.test.js index dc7027a..17228cd 100644 --- a/test/indexes.test.js +++ b/test/indexes.test.js @@ -13,7 +13,7 @@ function workspace() { const root=fs.mkdtempSync(path.join(os.tmpdir(),'vnext-indexes-')); const write=(file, data)=>{ const target=path.join(root,file);fs.mkdirSync(path.dirname(target),{recursive:true});fs.writeFileSync(target,JSON.stringify(data)); }; write('vnext.config.json',{domain:'test', paths:{componentsRoot:'components',workflows:'Workflows',schemas:'Schemas'}}); - const master={type:'master',key:'master',version:'1.0.0',domain:'test',flow:'sys-schemas',attributes:{type:'workflow',schema:{type:'object',properties:{amount:{type:'number','x-indexed':true,'x-filterOperators':['gt']}}}}}; + const master={key:'master',version:'1.0.0',domain:'test',flow:'sys-schemas',attributes:{type:'master',schema:{type:'object',properties:{amount:{type:'number','x-indexed':true,'x-filterOperators':['gt']}}}}}; const flow={key:'orders',version:'1.0.0',domain:'test',flow:'sys-flows',attributes:{schema:{key:'master',domain:'test',flow:'sys-schemas',version:'latest'}}}; write('components/Schemas/master.json',master);write('components/Workflows/orders.json',flow); return {root,write,master,flow}; @@ -69,7 +69,7 @@ test('generation is offline, immutable, scoped, and writes manifests matching SQ test('all local workflow versions are merged; duplicate identities and schema collisions fail',async()=>{ const w=workspace(); try { - w.write('components/Schemas/master2.json',{...w.master,version:'2.0.0',attributes:{schema:{properties:{name:{type:'string','x-indexed':true}}}}}); + w.write('components/Schemas/master2.json',{...w.master,version:'2.0.0',attributes:{type:'master',schema:{properties:{name:{type:'string','x-indexed':true}}}}}); w.write('components/Workflows/orders.json',{...w.flow,attributes:{schema:{...w.flow.attributes.schema,version:'1.0.0'}}}); w.write('components/Workflows/orders2.json',{...w.flow,version:'2.0.0'}); assert.deepEqual((await loadPlans(w.root))[0].projections.map(p=>p.path),['amount','amount','name']); @@ -104,63 +104,49 @@ test('conditional indexed nodes and ancestors are rejected without rejecting unr }); -test('only referenced master schemas produce SQL plans; skip non-master latest versions without falling back',async()=>{ +test('only referenced master schemas produce SQL plans; skip non-master latest without fallback',async()=>{ const w=workspace(); try { - for(const type of ['transition','view','function']) { - w.write('components/Schemas/master.json',{...w.master,type,attributes:{schema:{type:'object',properties:{name:{type:'string'}}}}}); + for(const type of ['transition','view','function','workflow','task','headers','json-schema','custom-schema','MASTER']) { + w.write('components/Schemas/master.json',{...w.master,attributes:{type,schema:{type:'object'}}}); assert.deepEqual(await loadPlans(w.root),[]); - await assert.rejects(generate({output:'ignored-output'},w.root),/No workflows referencing type: master/); + await assert.rejects(generate({output:'ignored-output'},w.root),/No workflows referencing attributes.type: master/); assert.equal(fs.existsSync(path.join(w.root,'ignored-output')),false); } w.write('components/Schemas/master.json',w.master); - w.write('components/Schemas/next.json',{...w.master,version:'2.0.0',type:'transition',attributes:{schema:{type:'object'}}}); + w.write('components/Schemas/next.json',{...w.master,version:'2.0.0',attributes:{type:'custom-schema',schema:{type:'object'}}}); assert.deepEqual(await loadPlans(w.root),[]); w.write('components/Workflows/orders.json',{...w.flow,attributes:{schema:{...w.flow.attributes.schema,version:'1.0.0'}}}); assert.equal((await loadPlans(w.root)).length,1); } finally {fs.rmSync(w.root,{recursive:true,force:true});} }); -test('non-master x-indexed metadata and invalid schema purposes fail validation',async()=>{ +test('only exact attributes.type master allows x-indexed including false',async()=>{ const w=workspace(); try { - for(const type of ['transition','view','function']) { + for(const type of ['transition','view','function','custom-schema','MASTER',undefined,null,'',' ']) { for(const indexed of [true,false]) { - w.write('components/Schemas/master.json',{...w.master,type,attributes:{schema:{properties:{nested:{properties:{value:{type:'number','x-indexed':indexed}}}}}}}); - await assert.rejects(loadPlans(w.root),/x-indexed is only allowed.*master/); + w.write('components/Schemas/master.json',{...w.master,type:'master',attributes:{type,schema:{properties:{nested:{properties:{value:{type:'number','x-indexed':indexed}}}}}}}); + await assert.rejects(loadPlans(w.root),/x-indexed is only allowed when attributes.type is 'master'/); } } - for(const type of ['workflow','json-schema','MASTER',42,{},[]]) { - w.write('components/Schemas/master.json',{...w.master,type,attributes:{schema:{type:'object'}}}); - await assert.rejects(loadPlans(w.root),/Schema type must be one of/); + for(const type of [42,{},[]]) { + w.write('components/Schemas/master.json',{...w.master,attributes:{type,schema:{type:'object'}}}); + await assert.rejects(loadPlans(w.root),/attributes.type must be a string/); } } finally {fs.rmSync(w.root,{recursive:true,force:true});} }); - -test('root type controls indexing while existing attributes.type remains independent',async()=>{ +test('root type is ignored and missing attributes.type never defaults to master',async()=>{ const w=workspace(); try { - for(const attributeType of ['workflow','task','function','view','schema','extension','headers','json-schema']) { - w.write('components/Schemas/master.json',{...w.master,attributes:{...w.master.attributes,type:attributeType}}); + for(const type of ['view','custom',undefined,null]) { + w.write('components/Schemas/master.json',{...w.master,type}); assert.equal((await loadPlans(w.root)).length,1); } - w.write('components/Schemas/master.json',{...w.master,type:'view',attributes:{type:'master',schema:{type:'object'}}}); - assert.deepEqual(await loadPlans(w.root),[]); - w.write('components/Schemas/master.json',{...w.master,type:undefined,attributes:{...w.master.attributes,type:'master'}}); - await assert.rejects(loadPlans(w.root),/x-indexed is only allowed/); - } finally {fs.rmSync(w.root,{recursive:true,force:true});} -}); - - -test('optional root purpose has no default and never opts into indexing',async()=>{ - const w=workspace(); - try { for(const type of [undefined,null,'',' ']) { - w.write('components/Schemas/master.json',{...w.master,type,attributes:{type:'master',schema:{type:'object'}}}); + w.write('components/Schemas/master.json',{...w.master,type:'master',attributes:{type,schema:{type:'object'}}}); assert.deepEqual(await loadPlans(w.root),[]); - w.write('components/Schemas/master.json',{...w.master,type}); - await assert.rejects(loadPlans(w.root),/x-indexed is only allowed/); } } finally {fs.rmSync(w.root,{recursive:true,force:true});} });