brainy/src/db/portableGraph.ts

1090 lines
44 KiB
TypeScript

/**
* @module db/portableGraph
* @description Portable graph export/import engine for Brainy 8.0 — the
* `PortableGraph v1` document format plus the `export`/`import` engines that power
* `db.export()` / `brain.export()` (read, at a pinned generation) and
* `brain.import(graph)` (write, applied as one atomic transaction).
*
* A `PortableGraph` is a self-describing, versioned, **portable** representation
* of a graph (entities + relations, optionally vectors and file blobs) — the unit
* Brainy exports for interchange between instances, versions, and products, and the
* payload other artifacts embed (e.g. a workbench file's `graph` field). It is NOT
* a backup: that role belongs to `db.persist()`/`Brainy.load()`, the native
* whole-brain snapshot which preserves generation history. `PortableGraph` is the
* human-readable, partial-or-whole, cross-version round-trip; distinct also from
* `brain.import(file)` (foreign-file ingestion).
*
* Every document carries `format: 'brainy-portable-graph'` and `formatVersion: 1`;
* import gates on `formatVersion` for forward migration. The format is shared
* verbatim across the 7.x and 8.0 lines.
*
* **Reserved-field split:** each `PortableGraphEntity` carries Brainy's standard fields
* (`subtype`, `visibility`, `data`, `confidence`, `weight`, `service`, `createdBy`,
* `createdAt`) at the TOP LEVEL and `metadata` holds ONLY custom user fields —
* mirroring the in-memory `Entity`, so import maps each to its dedicated `add`/
* `relate` parameter rather than the metadata bag.
*/
import { Entity, Relation, Result } from '../types/brainy.types.js'
import { NounType, VerbType } from '../types/graphTypes.js'
import { StorageAdapter, HNSWVerbWithMetadata } from '../coreTypes.js'
import { getBrainyVersion } from '../utils/version.js'
import { TxOperation } from './types.js'
/** Fixed entity id of the VFS root collection (a `system` entity; excluded unless `includeSystem`). */
const VFS_ROOT_ID = '00000000-0000-0000-0000-000000000000'
/** Magic string identifying a Brainy `PortableGraph` document (the `format` tag). */
export const PORTABLE_GRAPH_FORMAT = 'brainy-portable-graph'
/** Current portable-format version (import gates on this for cross-version migration). */
export const PORTABLE_GRAPH_FORMAT_VERSION = 1
/** Default embedding model label (informational; `dimensions` is the real compat gate). */
const DEFAULT_EMBED_MODEL = 'all-MiniLM-L6-v2'
/** Page size for find-based enumeration (whole-brain / predicate selectors). */
const ENUM_PAGE = 1000
/** Per-node relation fetch ceiling (source/target-filtered, so no unfiltered-scan warning). */
const RELATION_FETCH_LIMIT = 100_000
// ============================================================================
// Public types — identical wire shape to the 7.x line
// ============================================================================
/**
* @description Selects WHICH part of the graph to export. Reuses `find()`'s grammar
* plus export-specific selectors. Omit (or pass `{}`) to export the whole brain.
* Structural selectors (`ids`/`collection`/`connected`/`vfsPath`) and predicate
* selectors (`type`/`subtype`/`where`/`service`/`visibility`) **compose**.
*/
export interface ExportSelector {
/** Exactly these entity ids. */
ids?: string[]
/** A collection id → the collection + its transitive `Contains` members. */
collection?: string
/** Alias for `collection`. */
memberOf?: string
/** An entity + its N-hop neighbourhood. */
connected?: {
from: string
depth?: number
verbs?: VerbType[]
direction?: 'out' | 'in' | 'both'
}
/** A VFS path → the directory/file + (for a directory) its `Contains` subtree. */
vfsPath?: string
/** For `vfsPath` directories: include the whole subtree (default: true). */
recursive?: boolean
/** For `collection`/`vfsPath`: cap traversal depth (default: unbounded). */
depth?: number
/** Predicate: entity type(s). */
type?: NounType | NounType[]
/** Predicate: entity subtype(s). */
subtype?: string | string[]
/** Predicate: exact-match metadata fields. */
where?: Record<string, any>
/** Predicate: multi-tenancy service id. */
service?: string
/** Predicate: visibility (`'public'` matches entities with no explicit visibility). */
visibility?: string
}
/** Controls HOW the selected graph is serialized. */
export interface ExportOptions {
/** Include embedding vectors verbatim (default: false → `import()` re-embeds from `data`). */
includeVectors?: boolean
/** Include VFS file bytes in `blobs` so files round-trip byte-identically (default: false). */
includeContent?: boolean
/** Include `visibility:'system'` entities (e.g. the VFS root) (default: false). */
includeSystem?: boolean
/**
* Admit BOTH hidden visibility tiers — `'internal'` AND `'system'` — into the
* whole-brain/predicate candidate set, in EITHER `enumeration` mode (default:
* false — today's behavior is byte-identical). `includeSystem` alone only ever
* reached `'system'` for a structural selector's per-entity gate; whole-brain/
* predicate enumeration never forwarded it into the candidate walk at all, so a
* hidden-tier row could never survive a whole-brain export regardless of any
* flag — the gap this option closes. `includeHidden: true` IMPLIES
* `includeSystem: true` (both tiers are admitted together; there is no
* "system but not internal" combination via this flag) — `includeSystem` on
* its own keeps its narrower, pre-existing meaning for back-compat.
*
* Migration-grade exports set `includeHidden: true` — a complete-canon export
* must carry every visibility tier; consumer-facing exports leave it off.
*/
includeHidden?: boolean
/** Which edges to include (default: `'induced'`). */
edges?: 'induced' | 'incident' | 'none'
/**
* How the whole-brain / predicate selector (no `ids`/`collection`/`connected`/
* `vfsPath`) resolves its candidate id set:
*
* - `'index'` (DEFAULT — unchanged behavior) — the generation-correct
* paginated `find()` walk. Fast (O(matches), not O(N)) but rides the
* metadata index as an acceleration structure: a lost/stale index posting
* can silently OMIT a canonical record, and a stale posting pointing at a
* record that no longer matches can silently produce a phantom (dropped
* later by the same predicate re-check `'canonical'` mode also runs, so
* phantoms never reach `entities` — but they ARE lost silently unless
* {@link reportIndexDrift} is set).
* - `'canonical'` — walks every live noun/verb directly off the storage
* adapter's canonical shard layout (`storage.getNouns()`/`getVerbs()` —
* the same primitive `repairIndex()`'s recount and every index-heal walk
* use), then applies the selector as a plain predicate over the walked
* records. This GUARANTEES canon-completeness — the metadata/graph
* indexes are never consulted, so their corruption cannot hide a live
* record — at the cost of an O(N) walk regardless of selector
* selectivity (unlike the index path's O(matches)). Structural selectors
* (`ids`/`collection`/`connected`/`vfsPath`) resolve their node set
* exactly as in `'index'` mode either way (they never rode the metadata
* index); `'canonical'` additionally walks relations canonically for
* EVERY selector, since a lost adjacency-index posting can hide a
* relation regardless of how the node set was produced. Requires a
* storage adapter and the CURRENT generation — throws on a historical
* `asOf()` view or a speculative `with()` overlay (see
* {@link CanonicalEnumerationUnavailableError}), because the canonical
* walk has no notion of "as of a past generation."
*/
enumeration?: 'index' | 'canonical'
/**
* Only meaningful with `enumeration:'canonical'` (ignored otherwise): ALSO
* run the `'index'` enumeration in parallel and diff it against the
* canonical ground truth, attaching the result as {@link PortableGraph.drift}.
* Never auto-heals anything — this is migration-audit evidence, reported
* loudly (`console.warn` with the counts) whenever either list is non-empty,
* never silently. Default: false.
*/
reportIndexDrift?: boolean
}
/**
* @description Index-vs-canonical drift for one `export({ enumeration: 'canonical',
* reportIndexDrift: true })` call. Only populated for the whole-brain / predicate
* selector (structural selectors never consulted the metadata index for their node
* set, so there is nothing to diff — both lists are empty for those).
*/
export interface ExportIndexDrift {
/**
* Ids the canonical storage walk confirmed (live, selector-matching) that the
* index-based `find()` enumeration did NOT return — canonical records the
* metadata index has lost track of.
*/
canonicalOnly: string[]
/**
* Ids the index-based `find()` enumeration returned for this selector that
* canonical ground truth (the storage walk + the same predicate check) does
* NOT support — phantom index rows (stale or cross-bucket postings).
*/
indexOnly: string[]
}
/** Controls how a `PortableGraph` is applied on `import()`. */
export interface ImportOptions {
/** Conflict policy for an existing id (default: `'merge'` — dedup-by-id). */
onConflict?: 'merge' | 'replace' | 'skip'
/** Vector policy (default: `'auto'` — re-embed from `data` when no vector is carried). */
reembed?: 'auto' | 'never'
/** Rewrite every id on the way in (e.g. to clone a template subgraph under fresh ids). */
remapIds?: (id: string) => string
/** Transaction metadata, recorded in the tx-log alongside the new generation. */
meta?: Record<string, unknown>
}
/** One entity in a `PortableGraph`. Standard fields top-level; `metadata` is custom-only. */
export interface PortableGraphEntity {
id: string
type: string
subtype?: string
visibility?: string
data?: any
confidence?: number
weight?: number
service?: string
createdBy?: any
createdAt?: number
vector?: number[]
metadata?: any
}
/** One relation (edge) in a `PortableGraph`. */
export interface PortableGraphRelation {
id: string
from: string
to: string
type: string
subtype?: string
visibility?: string
weight?: number
confidence?: number
metadata?: any
}
/** A self-describing, versioned, portable graph document (identical on 7.x and 8.0). */
export interface PortableGraph {
format: typeof PORTABLE_GRAPH_FORMAT
formatVersion: number
brainyVersion: string
createdAt: string
embedding: { model: string; dimensions: number }
selector?: ExportSelector
entities: PortableGraphEntity[]
relations: PortableGraphRelation[]
blobs?: Record<string, string>
danglingIds?: string[]
/** Present only when `export()` was called with `reportIndexDrift: true`. */
drift?: ExportIndexDrift
stats: { entityCount: number; relationCount: number; blobCount: number; vectorDimensions?: number }
}
/** Outcome of an `import()`. */
export interface ImportResult {
imported: number
merged: number
skipped: number
reembedded: number
blobsWritten: number
errors: Array<{ id: string; error: string }>
}
/** The minimal read surface the export engine needs — satisfied by `Db` and `Brainy`. */
export interface PortableGraphReader<T = any> {
get(id: string, options?: { includeVectors?: boolean }): Promise<Entity<T> | null>
find(query: any): Promise<Result<T>[]>
related(paramsOrId?: any): Promise<Relation<T>[]>
}
/** The minimal write surface the import engine needs — satisfied by `Brainy`. */
export interface PortableGraphWriter<T = any> {
get(id: string, options?: { includeVectors?: boolean }): Promise<Entity<T> | null>
transact(ops: TxOperation<T>[], options?: { meta?: Record<string, unknown> }): Promise<unknown>
}
/** Type guard: is this value a Brainy `PortableGraph` document (vs a file/foreign object)? */
export function isPortableGraph(value: unknown): value is PortableGraph {
return (
typeof value === 'object' &&
value !== null &&
(value as any).format === PORTABLE_GRAPH_FORMAT &&
Array.isArray((value as any).entities)
)
}
/** Result of {@link validatePortableGraph}. `valid` is true iff `errors` is empty. */
export interface PortableGraphValidation {
valid: boolean
errors: string[]
warnings: string[]
}
/**
* @description Dry-run check a value before `import()`, without mutating the brain:
* structural validity, a supported `formatVersion`, entity-id uniqueness + required
* fields, and relation-endpoint coverage. `import()` throws on a non-`PortableGraph` or a
* too-new `formatVersion`; `validatePortableGraph()` lets a consumer (e.g. a serializer checking
* a `.wbench` graph payload) surface issues first.
* @param data - The value to validate (need not be a `PortableGraph`).
* @returns `{ valid, errors, warnings }`.
* @example
* const { valid, errors } = validatePortableGraph(payload)
* if (!valid) throw new Error(`invalid PortableGraph: ${errors.join('; ')}`)
* await brain.import(payload)
*/
export function validatePortableGraph(data: unknown): PortableGraphValidation {
const errors: string[] = []
const warnings: string[] = []
if (!isPortableGraph(data)) {
errors.push(`not a PortableGraph document (expected format: '${PORTABLE_GRAPH_FORMAT}')`)
return { valid: false, errors, warnings }
}
if (typeof data.formatVersion !== 'number') {
errors.push('formatVersion is missing or not a number')
} else if (data.formatVersion > PORTABLE_GRAPH_FORMAT_VERSION) {
errors.push(
`formatVersion ${data.formatVersion} is newer than supported (max ${PORTABLE_GRAPH_FORMAT_VERSION})`
)
}
const ids = new Set<string>()
for (const e of data.entities) {
if (!e || typeof e.id !== 'string' || e.id.length === 0) {
errors.push('an entity is missing an id')
continue
}
if (ids.has(e.id)) errors.push(`duplicate entity id: ${e.id}`)
ids.add(e.id)
if (e.type === undefined || e.type === null || (e.type as string) === '') {
errors.push(`entity ${e.id} is missing a type`)
}
}
const dangling = new Set(data.danglingIds ?? [])
for (const r of data.relations) {
if (!r || typeof r.id !== 'string') {
errors.push('a relation is missing an id')
continue
}
if (r.type === undefined || r.type === null || (r.type as string) === '') {
errors.push(`relation ${r.id} is missing a type`)
}
if (!r.from || !r.to) {
errors.push(`relation ${r.id} is missing from/to`)
continue
}
if (!ids.has(r.from) && !dangling.has(r.from)) {
warnings.push(`relation ${r.id}: from-endpoint ${r.from} is not in entities`)
}
if (!ids.has(r.to) && !dangling.has(r.to)) {
warnings.push(`relation ${r.id}: to-endpoint ${r.to} is not in entities`)
}
}
if (!data.embedding || typeof data.embedding.dimensions !== 'number') {
warnings.push('embedding manifest is missing or has no dimensions')
}
return { valid: errors.length === 0, errors, warnings }
}
// ============================================================================
// EXPORT
// ============================================================================
/**
* @description Serialize part or all of a graph (read through `reader` at its pinned
* generation) into a portable `PortableGraph` document.
*
* `enumeration:'canonical'` (see {@link ExportOptions.enumeration}) requires a
* storage adapter and the current generation: it throws
* {@link CanonicalEnumerationUnavailableError} if `storage` is absent, and the
* caller (`Db.export()`) throws the same error before this runs if the view is
* historical or a speculative overlay — the canonical storage walk has no
* generation parameter, so it can only ever answer "as of right now."
*
* @param reader - Generation-correct read surface (`Db` or `Brainy`).
* @param storage - Storage adapter (VFS blob bytes when `includeContent`; the
* canonical noun/verb walk when `enumeration:'canonical'`).
* @param selector - WHAT to export (omit for the whole brain).
* @param options - HOW to export (vectors / file bytes / edge policy / enumeration mode).
*/
export async function exportGraph<T = any>(
reader: PortableGraphReader<T>,
storage: StorageAdapter | undefined,
selector: ExportSelector,
options: ExportOptions
): Promise<PortableGraph> {
const {
includeVectors = false,
includeContent = false,
includeSystem = false,
includeHidden = false,
edges = 'induced',
enumeration = 'index',
reportIndexDrift = false
} = options
if (enumeration === 'canonical' && !storage) {
throw new Error(
`export(): enumeration:'canonical' requires a storage adapter, but none was supplied ` +
`to this reader. Use enumeration:'index' (the default), or export through a Db/Brainy ` +
`that carries its storage adapter.`
)
}
const wantDrift = enumeration === 'canonical' && reportIndexDrift
// includeHidden IMPLIES includeSystem (see ExportOptions.includeHidden's JSDoc) — every
// system-tier gate below reads THIS combined value, never the raw option, so
// `includeHidden` alone is always sufficient to see system-tier rows too.
const effectiveIncludeSystem = includeSystem || includeHidden
// 1. Resolve the node-id set (+ the index's raw candidate set, only when diffing it).
// Both `enumerateAllCanonical` and `enumerateAllIndexed` receive the SAME
// `effectiveIncludeSystem`/`includeHidden` pair below, so a drift diff can never
// contain tier-policy noise — only genuine index-vs-canonical disagreement.
const { idSet, indexCandidateIds } = await resolveSelector(
reader,
storage,
selector,
effectiveIncludeSystem,
enumeration,
wantDrift,
includeHidden
)
// 2. Read canonical entities (reserved fields top-level), applying any predicate filter.
// Identical for both enumeration modes: 'canonical' only changes WHICH ids reach this
// loop, never how a candidate is verified — so the two modes can disagree on candidacy,
// never on what counts as a match.
const usePredicate = hasPredicate(selector)
const entityMap = new Map<string, Entity<T>>()
const entities: PortableGraphEntity[] = []
for (const id of idSet) {
const e = await reader.get(id, { includeVectors })
if (!e) continue
if (!effectiveIncludeSystem && (e as any).visibility === 'system') continue
if (usePredicate && !matchesPredicate(e, selector)) continue
entityMap.set(id, e)
entities.push(toPortableGraphEntity(e, includeVectors))
}
const keptIds = new Set(entityMap.keys())
// 2b. Finalize the drift report now that ground truth (keptIds) is known.
let drift: ExportIndexDrift | undefined
if (wantDrift) {
const indexIds = indexCandidateIds ?? new Set<string>()
const canonicalOnly = [...keptIds].filter((id) => !indexIds.has(id))
const indexOnly = [...indexIds].filter((id) => !keptIds.has(id))
drift = { canonicalOnly, indexOnly }
if (canonicalOnly.length > 0 || indexOnly.length > 0) {
console.warn(
`[Brainy] export() index drift: ${canonicalOnly.length} canonical-only id(s) ` +
`(canon-present, the index-based enumeration missed them) and ${indexOnly.length} ` +
`index-only id(s) (index-visible, canon-absent — phantom rows). ` +
`See the returned PortableGraph's 'drift' field for the exact ids. Nothing was ` +
`auto-healed — run brain.repairIndex() to reconcile the metadata index.`
)
}
}
// 3. Edges per policy. Canonical mode ALSO walks verbs canonically for every
// selector (not just whole-brain) — a lost adjacency-index posting can hide a
// relation regardless of how the node set was produced.
const { relations, danglingIds } =
enumeration === 'canonical'
? await collectEdgesCanonical(storage!, keptIds, edges, effectiveIncludeSystem, includeHidden)
: await collectEdges(reader, keptIds, edges, includeHidden)
// 4. VFS blob bytes (only when requested).
let blobs: Record<string, string> | undefined
if (includeContent) blobs = await collectBlobs(storage, entityMap)
const blobCount = blobs ? Object.keys(blobs).length : 0
// Embedding dimensionality for the manifest: detect from a carried vector,
// else the default model's 384 (informational — `dimensions` is the compat gate).
const dimensions = entities.find((e) => e.vector && e.vector.length)?.vector?.length ?? 384
return {
format: PORTABLE_GRAPH_FORMAT,
formatVersion: PORTABLE_GRAPH_FORMAT_VERSION,
brainyVersion: getBrainyVersion(),
createdAt: new Date().toISOString(),
embedding: { model: DEFAULT_EMBED_MODEL, dimensions },
selector,
entities,
relations,
...(blobs && blobCount > 0 ? { blobs } : {}),
...(danglingIds && danglingIds.length > 0 ? { danglingIds } : {}),
...(drift ? { drift } : {}),
stats: {
entityCount: entities.length,
relationCount: relations.length,
blobCount,
vectorDimensions: dimensions
}
}
}
// ============================================================================
// IMPORT
// ============================================================================
/**
* @description Apply a `PortableGraph` to the brain as ONE atomic transaction (a single
* new generation). Dedup-by-id merge by default; re-embeds from `data` when no vector
* is carried. VFS blob bytes are written first (content-addressed, idempotent).
* @param writer - The brain (`get` + `transact`).
* @param storage - Storage adapter (for VFS blob bytes).
* @param data - A `PortableGraph` document.
* @param options - Conflict / vector / id-remap policy.
* @throws If `data` is not a `PortableGraph`, or its `formatVersion` is newer than supported.
*/
export async function importGraph<T = any>(
writer: PortableGraphWriter<T>,
storage: StorageAdapter | undefined,
data: PortableGraph,
options: ImportOptions = {}
): Promise<ImportResult> {
if (!isPortableGraph(data)) {
throw new Error(
`import() expects a PortableGraph document (format:'${PORTABLE_GRAPH_FORMAT}'). ` +
`For file ingestion (CSV/PDF/Excel/JSON), pass a file/buffer instead.`
)
}
if (typeof data.formatVersion === 'number' && data.formatVersion > PORTABLE_GRAPH_FORMAT_VERSION) {
throw new Error(
`PortableGraph formatVersion ${data.formatVersion} is newer than this Brainy supports ` +
`(max ${PORTABLE_GRAPH_FORMAT_VERSION}). Upgrade Brainy to import this document.`
)
}
const { onConflict = 'merge', reembed = 'auto', remapIds, meta } = options
const result: ImportResult = {
imported: 0,
merged: 0,
skipped: 0,
reembedded: 0,
blobsWritten: 0,
errors: []
}
const mapId = (id: string) => (remapIds ? remapIds(id) : id)
// 1. Blob bytes first (content-addressed → idempotent), so file entities resolve content.
if (data.blobs && Object.keys(data.blobs).length > 0) {
const blobStorage = (storage as any)?.blobStorage
if (blobStorage?.write) {
for (const [hash, b64] of Object.entries(data.blobs)) {
try {
await blobStorage.write(Buffer.from(b64, 'base64'))
result.blobsWritten++
} catch (e) {
result.errors.push({ id: hash, error: `blob: ${(e as Error).message}` })
}
}
} else {
for (const hash of Object.keys(data.blobs)) {
result.errors.push({ id: hash, error: 'blob: storage does not support binary blobs' })
}
}
}
// 2. Build the operation batch (one atomic transaction → one generation).
const ops: TxOperation<T>[] = []
for (const be of data.entities || []) {
const id = mapId(be.id)
try {
const exists = (await writer.get(id)) !== null
if (exists) {
if (onConflict === 'skip') {
result.skipped++
continue
}
if (onConflict === 'merge') {
ops.push({ op: 'update', id, ...entityUpdateFields(be), merge: true } as TxOperation<T>)
result.merged++
continue
}
ops.push({ op: 'remove', id } as TxOperation<T>) // 'replace'
}
const useVector = Array.isArray(be.vector) && be.vector.length > 0
if (!useVector && reembed === 'never') {
result.errors.push({ id, error: 'no vector carried and reembed:never' })
continue
}
ops.push({
op: 'add',
id,
...entityAddFields(be),
...(useVector ? { vector: be.vector } : {})
} as TxOperation<T>)
result.imported++
if (!useVector) result.reembedded++
} catch (e) {
result.errors.push({ id, error: (e as Error).message })
}
}
for (const br of data.relations || []) {
ops.push({
op: 'relate',
from: mapId(br.from),
to: mapId(br.to),
type: br.type as VerbType,
...(br.subtype !== undefined ? { subtype: br.subtype } : {}),
...(br.weight !== undefined ? { weight: br.weight } : {}),
...(br.confidence !== undefined ? { confidence: br.confidence } : {}),
...(br.metadata !== undefined ? { metadata: br.metadata } : {})
} as TxOperation<T>)
}
// 3. Apply atomically — one generation for the whole graph, or none.
if (ops.length > 0) {
await writer.transact(ops, meta ? { meta } : undefined)
}
return result
}
// ============================================================================
// Selector resolution (private)
// ============================================================================
function hasStructural(s: ExportSelector): boolean {
return !!(s.ids || s.collection || s.memberOf || s.connected || s.vfsPath)
}
function hasPredicate(s: ExportSelector): boolean {
return (
s.type !== undefined ||
s.subtype !== undefined ||
s.where !== undefined ||
s.service !== undefined ||
s.visibility !== undefined
)
}
/**
* @param reader - Generation-correct read surface.
* @param storage - Storage adapter (only touched when `enumeration:'canonical'`
* resolves the whole-brain/predicate branch).
* @param s - The export selector.
* @param includeSystem - The ALREADY-COMBINED `includeSystem || includeHidden` value
* (see `exportGraph`'s `effectiveIncludeSystem`) — whether `visibility:'system'`
* entities are wanted.
* @param enumeration - `'index'` (default) or `'canonical'` — see {@link ExportOptions.enumeration}.
* Only affects the whole-brain/predicate branch (the `else` below): structural
* selectors (`ids`/`collection`/`connected`/`vfsPath`) never rode the metadata
* index for their node set, so they resolve identically either way.
* @param wantIndexCandidates - When true (only meaningful with `enumeration:'canonical'`
* on the whole-brain/predicate branch), ALSO run the index-based walk and return
* its raw candidate set as `indexCandidateIds`, for {@link ExportIndexDrift}.
* @param includeHidden - Whether `visibility:'internal'` entities are ALSO wanted
* (see {@link ExportOptions.includeHidden}). Threaded to BOTH enumeration
* functions identically so a drift diff never contains tier-policy noise.
*/
async function resolveSelector<T>(
reader: PortableGraphReader<T>,
storage: StorageAdapter | undefined,
s: ExportSelector,
includeSystem: boolean,
enumeration: 'index' | 'canonical',
wantIndexCandidates: boolean,
includeHidden: boolean
): Promise<{ idSet: Set<string>; indexCandidateIds?: Set<string> }> {
let idSet: Set<string>
let indexCandidateIds: Set<string> | undefined
if (s.ids && s.ids.length) {
idSet = new Set(s.ids)
} else if (s.collection ?? s.memberOf) {
idSet = await resolveCollectionSubtree(reader, (s.collection ?? s.memberOf)!, s.depth)
} else if (s.connected) {
idSet = await resolveConnected(reader, s.connected)
} else if (s.vfsPath) {
idSet = await resolveVfsPath(reader, s.vfsPath, s.recursive ?? true, s.depth)
} else if (enumeration === 'canonical') {
idSet = await enumerateAllCanonical(storage!, includeSystem, includeHidden)
if (wantIndexCandidates) indexCandidateIds = await enumerateAllIndexed(reader, s, includeHidden)
} else {
idSet = await enumerateAllIndexed(reader, s, includeHidden)
}
if (!includeSystem) idSet.delete(VFS_ROOT_ID)
return { idSet, indexCandidateIds }
}
/**
* @description Whole-brain / predicate enumeration via generation-correct
* paginated `find()`. The metadata index is an acceleration structure over
* this candidate set — see {@link enumerateAllCanonical} for the storage-level
* counterpart that never consults it.
*
* @param includeHidden - When true, forwards `includeInternal: true` AND
* `includeSystem: true` into the SAME `find()` call — `find()` supports both
* flags simultaneously (confirmed via `FindParams.includeInternal`/`includeSystem`
* and `Brainy`'s `resolveHiddenIds`/`excludedVisibilityTiers`), so ONE pass
* reaches both hidden tiers; no per-tier union pass is needed. When false
* (default), neither flag is forwarded — the pre-existing behavior, preserved
* byte-identically for back-compat (`ExportOptions.includeSystem` alone never
* reached this far; see {@link ExportOptions.includeHidden}'s JSDoc).
*/
async function enumerateAllIndexed<T>(
reader: PortableGraphReader<T>,
s: ExportSelector,
includeHidden = false
): Promise<Set<string>> {
const params: any = {}
if (s.type !== undefined) params.type = s.type
if (s.subtype !== undefined) params.subtype = s.subtype
if (s.where !== undefined) params.where = s.where
if (s.service !== undefined) params.service = s.service
if (includeHidden) {
params.includeInternal = true
params.includeSystem = true
}
const ids = new Set<string>()
let offset = 0
// eslint-disable-next-line no-constant-condition
while (true) {
const batch = await reader.find({ ...params, limit: ENUM_PAGE, offset })
for (const r of batch) ids.add(r.id)
if (batch.length < ENUM_PAGE) break
offset += ENUM_PAGE
}
return ids
}
/**
* @description Canonical (storage-level) counterpart of {@link enumerateAllIndexed}:
* walks every live noun directly off the storage adapter's canonical shard layout
* (`storage.getNouns()` — the same primitive `repairIndex()`'s recount and every
* index-heal walk use) instead of going through the metadata index. Guarantees
* canon-completeness — a lost or stale metadata-index posting cannot cause a
* canonical record to be silently missing from the returned set — at the cost of
* an O(N) walk regardless of selector selectivity (unlike the index path's
* O(matches)). Returns the RAW candidate id set; `exportGraph`'s caller applies
* `matchesPredicate` per-entity via `reader.get()` afterward, exactly as the index
* path does, so both paths share one predicate-evaluation code path and can only
* disagree on candidacy, never on what a match means.
*
* Mirrors `find()`'s hidden-tier policy given the SAME `includeSystem`/`includeHidden`
* pair (see {@link enumerateAllIndexed}) so the two enumeration modes produce
* identical id sets when the index is healthy, at ANY tier-visibility setting.
*
* @param includeSystem - The ALREADY-COMBINED `includeSystem || includeHidden` value.
* @param includeHidden - Whether `'internal'`-tier nouns are ALSO admitted.
*/
async function enumerateAllCanonical(
storage: StorageAdapter,
includeSystem = false,
includeHidden = false
): Promise<Set<string>> {
const ids = new Set<string>()
let offset = 0
let cursor: string | undefined
// eslint-disable-next-line no-constant-condition
while (true) {
const page = await storage.getNouns({ pagination: { limit: ENUM_PAGE, offset, cursor } })
for (const item of page.items) {
if (item.visibility === 'internal' && !includeHidden) continue
if (item.visibility === 'system' && !includeSystem) continue
ids.add(item.id)
}
if (!page.hasMore || page.items.length === 0) break
if (page.nextCursor !== undefined) {
cursor = page.nextCursor
} else {
offset += ENUM_PAGE
}
}
return ids
}
async function resolveCollectionSubtree<T>(
reader: PortableGraphReader<T>,
rootId: string,
depth?: number
): Promise<Set<string>> {
const set = new Set<string>([rootId])
const maxDepth = depth ?? Infinity
let frontier = [rootId]
let d = 0
while (frontier.length && d < maxDepth) {
const next: string[] = []
for (const id of frontier) {
const rels = await reader.related({ from: id, type: VerbType.Contains, limit: RELATION_FETCH_LIMIT })
for (const r of rels) {
if (!set.has(r.to)) {
set.add(r.to)
next.push(r.to)
}
}
}
frontier = next
d++
}
return set
}
async function resolveConnected<T>(
reader: PortableGraphReader<T>,
c: NonNullable<ExportSelector['connected']>
): Promise<Set<string>> {
const { from, depth = 1, verbs, direction = 'out' } = c
const set = new Set<string>([from])
let frontier = [from]
for (let d = 0; d < depth; d++) {
const next: string[] = []
for (const id of frontier) {
const neighbours = await neighboursOf(reader, id, direction, verbs)
for (const n of neighbours) {
if (!set.has(n)) {
set.add(n)
next.push(n)
}
}
}
frontier = next
if (!next.length) break
}
return set
}
async function neighboursOf<T>(
reader: PortableGraphReader<T>,
id: string,
direction: 'out' | 'in' | 'both',
verbs?: VerbType[]
): Promise<string[]> {
const out: string[] = []
if (direction === 'out' || direction === 'both') {
const rels = await reader.related({ from: id, limit: RELATION_FETCH_LIMIT })
for (const r of rels) if (!verbs || verbs.includes(r.type)) out.push(r.to)
}
if (direction === 'in' || direction === 'both') {
const rels = await reader.related({ to: id, limit: RELATION_FETCH_LIMIT })
for (const r of rels) if (!verbs || verbs.includes(r.type)) out.push(r.from)
}
return out
}
async function resolveVfsPath<T>(
reader: PortableGraphReader<T>,
path: string,
recursive: boolean,
depth?: number
): Promise<Set<string>> {
const dirId = await resolveVfsPathToId(reader, path)
if (!dirId) return new Set()
if (!recursive) {
const set = new Set<string>([dirId])
const rels = await reader.related({ from: dirId, type: VerbType.Contains, limit: RELATION_FETCH_LIMIT })
for (const r of rels) set.add(r.to)
return set
}
return resolveCollectionSubtree(reader, dirId, depth)
}
async function resolveVfsPathToId<T>(reader: PortableGraphReader<T>, path: string): Promise<string | null> {
const segments = path.split('/').filter(Boolean)
let currentId = VFS_ROOT_ID
for (const seg of segments) {
const rels = await reader.related({ from: currentId, type: VerbType.Contains, limit: RELATION_FETCH_LIMIT })
let found: string | null = null
for (const r of rels) {
const child = await reader.get(r.to)
if ((child?.metadata as any)?.name === seg) {
found = r.to
break
}
}
if (!found) return null
currentId = found
}
return currentId
}
function matchesPredicate<T>(e: Entity<T>, s: ExportSelector): boolean {
if (s.type !== undefined) {
const types = Array.isArray(s.type) ? s.type : [s.type]
if (!types.includes(e.type)) return false
}
if (s.subtype !== undefined) {
const subs = Array.isArray(s.subtype) ? s.subtype : [s.subtype]
if (e.subtype === undefined || !subs.includes(e.subtype)) return false
}
if (s.service !== undefined && e.service !== s.service) return false
if (s.visibility !== undefined) {
const vis = (e as any).visibility ?? 'public'
if (vis !== s.visibility) return false
}
if (s.where && !matchesWhere(e.metadata, s.where)) return false
return true
}
function matchesWhere(metadata: any, where: Record<string, any>): boolean {
if (!metadata) return false
for (const [key, value] of Object.entries(where)) {
if (metadata[key] !== value) return false
}
return true
}
// ============================================================================
// Serialization helpers (private)
// ============================================================================
function toPortableGraphEntity<T>(e: Entity<T>, includeVectors: boolean): PortableGraphEntity {
const be: PortableGraphEntity = { id: e.id, type: e.type as string }
if (e.subtype !== undefined) be.subtype = e.subtype
const vis = (e as any).visibility
if (vis !== undefined && vis !== 'public') be.visibility = vis
if (e.data !== undefined) be.data = e.data
if (e.confidence !== undefined) be.confidence = e.confidence
if (e.weight !== undefined) be.weight = e.weight
if (e.service !== undefined) be.service = e.service
if ((e as any).createdBy !== undefined) be.createdBy = (e as any).createdBy
if (e.createdAt !== undefined) be.createdAt = e.createdAt
if (includeVectors && e.vector && e.vector.length) be.vector = e.vector
if (e.metadata && Object.keys(e.metadata as any).length) be.metadata = e.metadata
return be
}
function toPortableGraphRelation<T>(r: Relation<T>): PortableGraphRelation {
const br: PortableGraphRelation = { id: r.id, from: r.from, to: r.to, type: r.type as string }
if (r.subtype !== undefined) br.subtype = r.subtype
const vis = (r as any).visibility
if (vis !== undefined && vis !== 'public') br.visibility = vis
if (r.weight !== undefined) br.weight = r.weight
if (r.confidence !== undefined) br.confidence = r.confidence
if (r.metadata && Object.keys(r.metadata as any).length) br.metadata = r.metadata
return br
}
/**
* @param includeHidden - When true, forwards `includeInternal`/`includeSystem` into
* every `related()` call so hidden-tier relations reach candidacy too — mirrors
* {@link enumerateAllIndexed}'s `includeHidden` handling, and preserves back-compat
* when false/omitted (the pre-existing, unconditional hidden-tier exclusion).
*/
async function collectEdges<T>(
reader: PortableGraphReader<T>,
idSet: Set<string>,
edges: 'induced' | 'incident' | 'none',
includeHidden = false
): Promise<{ relations: PortableGraphRelation[]; danglingIds?: string[] }> {
if (edges === 'none') return { relations: [] }
const tierOptIn = includeHidden ? { includeInternal: true, includeSystem: true } : {}
const relations: PortableGraphRelation[] = []
const dangling = new Set<string>()
const seen = new Set<string>()
for (const id of idSet) {
const rels = await reader.related({ from: id, limit: RELATION_FETCH_LIMIT, ...tierOptIn })
for (const r of rels) {
if (seen.has(r.id)) continue
const toIn = idSet.has(r.to)
if (edges === 'induced' && !toIn) continue
if (!toIn) dangling.add(r.to)
seen.add(r.id)
relations.push(toPortableGraphRelation(r))
}
}
if (edges === 'incident') {
for (const id of idSet) {
const rels = await reader.related({ to: id, limit: RELATION_FETCH_LIMIT, ...tierOptIn })
for (const r of rels) {
if (seen.has(r.id)) continue
if (!idSet.has(r.from)) {
dangling.add(r.from)
seen.add(r.id)
relations.push(toPortableGraphRelation(r))
}
}
}
}
return dangling.size > 0 ? { relations, danglingIds: Array.from(dangling) } : { relations }
}
/** Converts a canonical verb record (as returned by `storage.getVerbs()`) into the wire shape. */
function hnswVerbToPortableGraphRelation(v: HNSWVerbWithMetadata): PortableGraphRelation {
const br: PortableGraphRelation = { id: v.id, from: v.sourceId, to: v.targetId, type: v.verb as string }
if (v.subtype !== undefined) br.subtype = v.subtype
if (v.visibility !== undefined && v.visibility !== 'public') br.visibility = v.visibility
if (v.weight !== undefined) br.weight = v.weight
if (v.confidence !== undefined) br.confidence = v.confidence
if (v.metadata && Object.keys(v.metadata as any).length) br.metadata = v.metadata
return br
}
/**
* @description Canonical (storage-level) counterpart of {@link collectEdges}:
* walks every live verb directly off `storage.getVerbs()` — the same primitive
* `repairIndex()`'s recount and every index-heal walk use — instead of the graph
* adjacency index (`reader.related()`), so a lost/stale adjacency posting cannot
* cause a canonical relationship to be silently dropped from the export. Used for
* EVERY selector in `enumeration:'canonical'` mode, not just the whole-brain
* branch: relations can be blinded by adjacency-index corruption regardless of
* how `idSet` (the kept node ids) was produced.
*
* Mirrors `related()`'s default hidden-tier policy (hides `'internal'` and
* `'system'` unless `includeHidden`/`includeSystem` say otherwise) so the two
* enumeration modes produce identical relation sets when the index is healthy.
*
* @param includeSystem - Whether `'system'`-tier verbs are admitted (the
* caller passes the ALREADY-combined `includeSystem || includeHidden` value —
* see `exportGraph`'s `effectiveIncludeSystem`).
* @param includeHidden - Whether `'internal'`-tier verbs are ALSO admitted.
*/
async function collectEdgesCanonical(
storage: StorageAdapter,
idSet: Set<string>,
edges: 'induced' | 'incident' | 'none',
includeSystem = false,
includeHidden = false
): Promise<{ relations: PortableGraphRelation[]; danglingIds?: string[] }> {
if (edges === 'none') return { relations: [] }
const relations: PortableGraphRelation[] = []
const dangling = new Set<string>()
const seen = new Set<string>()
let offset = 0
let cursor: string | undefined
// eslint-disable-next-line no-constant-condition
while (true) {
const page = await storage.getVerbs({ pagination: { limit: ENUM_PAGE, offset, cursor } })
for (const v of page.items) {
if (seen.has(v.id)) continue
if (v.visibility === 'internal' && !includeHidden) continue
if (v.visibility === 'system' && !includeSystem) continue
const fromIn = idSet.has(v.sourceId)
const toIn = idSet.has(v.targetId)
if (edges === 'induced') {
if (!fromIn || !toIn) continue
} else if (!fromIn && !toIn) {
continue // 'incident': neither endpoint kept — irrelevant to this export
}
if (fromIn && !toIn) dangling.add(v.targetId)
if (toIn && !fromIn) dangling.add(v.sourceId)
seen.add(v.id)
relations.push(hnswVerbToPortableGraphRelation(v))
}
if (!page.hasMore || page.items.length === 0) break
if (page.nextCursor !== undefined) {
cursor = page.nextCursor
} else {
offset += ENUM_PAGE
}
}
return dangling.size > 0 ? { relations, danglingIds: Array.from(dangling) } : { relations }
}
async function collectBlobs<T>(
storage: StorageAdapter | undefined,
entityMap: Map<string, Entity<T>>
): Promise<Record<string, string>> {
const blobs: Record<string, string> = {}
const blobStorage = (storage as any)?.blobStorage
if (!blobStorage?.read) return blobs
for (const e of entityMap.values()) {
const storageMeta = (e.metadata as any)?.storage
const hash = storageMeta?.hash
if (storageMeta?.type === 'blob' && hash && !blobs[hash]) {
try {
const buf = await blobStorage.read(hash)
blobs[hash] = Buffer.from(buf).toString('base64')
} catch {
// Referenced blob bytes unreadable (storage drift): skip — file structure still
// travels, and stats.blobCount reflects what was actually captured.
}
}
}
return blobs
}
// ============================================================================
// add()/update() param mapping (private)
// ============================================================================
function entityAddFields(be: PortableGraphEntity): Record<string, any> {
const fields: Record<string, any> = { data: be.data, type: be.type as NounType }
if (be.subtype !== undefined) fields.subtype = be.subtype
if (be.visibility !== undefined) fields.visibility = be.visibility
if (be.service !== undefined) fields.service = be.service
if (be.confidence !== undefined) fields.confidence = be.confidence
if (be.weight !== undefined) fields.weight = be.weight
if (be.metadata !== undefined) fields.metadata = be.metadata
return fields
}
function entityUpdateFields(be: PortableGraphEntity): Record<string, any> {
const fields: Record<string, any> = {}
if (be.data !== undefined) fields.data = be.data
if (be.type !== undefined) fields.type = be.type as NounType
if (be.subtype !== undefined) fields.subtype = be.subtype
if (be.visibility !== undefined) fields.visibility = be.visibility
if (be.confidence !== undefined) fields.confidence = be.confidence
if (be.weight !== undefined) fields.weight = be.weight
if (be.metadata !== undefined) fields.metadata = be.metadata
if (Array.isArray(be.vector) && be.vector.length) fields.vector = be.vector
return fields
}