brainy/src/db/backup.ts

/**
 * @module db/backup
 * @description Portable graph backup/restore engine for Brainy 8.0 — the
 * `BackupData v1` format plus the `export`/`import` engines that power
 * `db.export()` / `brain.export()` (read, at a pinned generation) and
 * `brain.import(backup)` (write, applied as one atomic transaction).
 *
 * The wire format is identical to the 7.x line (`formatVersion: 1`), so a
 * document written by 7.x imports cleanly here and vice-versa. This is the
 * **portable** round-trip (human-readable, partial-or-whole, cross-version),
 * distinct from `db.persist()`/`Brainy.load()` (the native whole-brain snapshot,
 * which preserves generation history) and from `brain.import(file)` (foreign-file
 * ingestion).
 *
 * **Reserved-field split:** each `BackupEntity` 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 } 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 portable backup document. */
export const BACKUP_FORMAT = 'brainy-backup'
/** Current portable-format version (import gates on this for cross-version migration). */
export const BACKUP_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
  /** Which edges to include (default: `'induced'`). */
  edges?: 'induced' | 'incident' | 'none'
}

/** Controls how a `BackupData` 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 `BackupData`. Standard fields top-level; `metadata` is custom-only. */
export interface BackupEntity {
  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 `BackupData`. */
export interface BackupRelation {
  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 BackupData {
  format: typeof BACKUP_FORMAT
  formatVersion: number
  brainyVersion: string
  createdAt: string
  embedding: { model: string; dimensions: number }
  selector?: ExportSelector
  entities: BackupEntity[]
  relations: BackupRelation[]
  blobs?: Record<string, string>
  danglingIds?: string[]
  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 BackupReader<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 BackupWriter<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 `BackupData` document (vs a file/foreign object)? */
export function isBackupData(value: unknown): value is BackupData {
  return (
    typeof value === 'object' &&
    value !== null &&
    (value as any).format === BACKUP_FORMAT &&
    Array.isArray((value as any).entities)
  )
}

// ============================================================================
// EXPORT
// ============================================================================

/**
 * @description Serialize part or all of a graph (read through `reader` at its pinned
 * generation) into a portable `BackupData` document.
 * @param reader - Generation-correct read surface (`Db` or `Brainy`).
 * @param storage - Storage adapter (used only for VFS blob bytes when `includeContent`).
 * @param selector - WHAT to export (omit for the whole brain).
 * @param options - HOW to export (vectors / file bytes / edge policy).
 * @param dimensions - Embedding dimensionality for the manifest.
 */
export async function exportGraph<T = any>(
  reader: BackupReader<T>,
  storage: StorageAdapter | undefined,
  selector: ExportSelector,
  options: ExportOptions
): Promise<BackupData> {
  const {
    includeVectors = false,
    includeContent = false,
    includeSystem = false,
    edges = 'induced'
  } = options

  // 1. Resolve the node-id set.
  const idSet = await resolveSelector(reader, selector, includeSystem)

  // 2. Read canonical entities (reserved fields top-level), applying any predicate filter.
  const usePredicate = hasPredicate(selector)
  const entityMap = new Map<string, Entity<T>>()
  const entities: BackupEntity[] = []
  for (const id of idSet) {
    const e = await reader.get(id, { includeVectors })
    if (!e) continue
    if (!includeSystem && (e as any).visibility === 'system') continue
    if (usePredicate && !matchesPredicate(e, selector)) continue
    entityMap.set(id, e)
    entities.push(toBackupEntity(e, includeVectors))
  }
  const keptIds = new Set(entityMap.keys())

  // 3. Edges per policy.
  const { relations, danglingIds } = await collectEdges(reader, keptIds, edges)

  // 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: BACKUP_FORMAT,
    formatVersion: BACKUP_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 } : {}),
    stats: {
      entityCount: entities.length,
      relationCount: relations.length,
      blobCount,
      vectorDimensions: dimensions
    }
  }
}

// ============================================================================
// IMPORT
// ============================================================================

/**
 * @description Apply a `BackupData` 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 `BackupData` document.
 * @param options - Conflict / vector / id-remap policy.
 * @throws If `data` is not a `BackupData`, or its `formatVersion` is newer than supported.
 */
export async function importGraph<T = any>(
  writer: BackupWriter<T>,
  storage: StorageAdapter | undefined,
  data: BackupData,
  options: ImportOptions = {}
): Promise<ImportResult> {
  if (!isBackupData(data)) {
    throw new Error(
      `import() expects a BackupData document (format:'${BACKUP_FORMAT}'). ` +
        `For file ingestion (CSV/PDF/Excel/JSON), pass a file/buffer instead.`
    )
  }
  if (typeof data.formatVersion === 'number' && data.formatVersion > BACKUP_FORMAT_VERSION) {
    throw new Error(
      `Backup formatVersion ${data.formatVersion} is newer than this Brainy supports ` +
        `(max ${BACKUP_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 backup, 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
  )
}

async function resolveSelector<T>(
  reader: BackupReader<T>,
  s: ExportSelector,
  includeSystem: boolean
): Promise<Set<string>> {
  let idSet: Set<string>
  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 {
    idSet = await enumerateAll(reader, s)
  }
  if (!includeSystem) idSet.delete(VFS_ROOT_ID)
  return idSet
}

/** Whole-brain / predicate enumeration via generation-correct paginated `find()`. */
async function enumerateAll<T>(reader: BackupReader<T>, s: ExportSelector): 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
  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
}

async function resolveCollectionSubtree<T>(
  reader: BackupReader<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: BackupReader<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: BackupReader<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: BackupReader<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: BackupReader<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 toBackupEntity<T>(e: Entity<T>, includeVectors: boolean): BackupEntity {
  const be: BackupEntity = { 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 toBackupRelation<T>(r: Relation<T>): BackupRelation {
  const br: BackupRelation = { 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
}

async function collectEdges<T>(
  reader: BackupReader<T>,
  idSet: Set<string>,
  edges: 'induced' | 'incident' | 'none'
): Promise<{ relations: BackupRelation[]; danglingIds?: string[] }> {
  if (edges === 'none') return { relations: [] }

  const relations: BackupRelation[] = []
  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 })
    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(toBackupRelation(r))
    }
  }

  if (edges === 'incident') {
    for (const id of idSet) {
      const rels = await reader.related({ to: id, limit: RELATION_FETCH_LIMIT })
      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(toBackupRelation(r))
        }
      }
    }
  }

  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: BackupEntity): 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: BackupEntity): 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
}