open-brainy/src/transaction/operations/IndexOperations.ts

/**
 * Index Operations with Rollback Support
 *
 * Provides transactional operations for all indexes:
 * - VectorIndexProvider (the JS HNSW fallback, or a native acceleration provider)
 * - MetadataIndexManager (roaring bitmap filtering)
 * - GraphAdjacencyIndex (LSM-tree graph storage)
 *
 * Each operation can be executed and rolled back atomically.
 */

import type { VectorIndexProvider, GraphIndexProvider } from '../../plugin.js'
import type { MetadataIndexManager } from '../../utils/metadataIndex.js'
import type { GraphVerb } from '../../coreTypes.js'
import type { Operation, RollbackAction } from '../types.js'

/**
 * Backend identity stamped into an operation's emitted `name` string (e.g.
 * `AddToVectorIndex(hnsw-js)`), resolved from the provider's own
 * {@link VectorIndexProvider.name} self-report.
 *
 * These operation classes are backend-neutral (the vector index they wrap may
 * be Brainy's own JS HNSW fallback OR a native acceleration provider), but
 * their names surface directly in consumer-visible transaction journals and
 * timings. `name` is REQUIRED at the TypeScript level — but a native provider
 * instance compiled against the previous (pre-required) contract can still
 * reach this function at runtime without it. That legacy case is tolerated,
 * never crashed on and never silently mislabeled: it resolves to
 * `'unknown-provider'` and emits ONE loud warning naming the missing contract
 * field, so the fix (implement `name`) is discoverable rather than a silent
 * fossil label in every journal line thereafter.
 */
const warnedMissingName = new WeakSet<VectorIndexProvider>()

function resolveVectorProviderId(index: VectorIndexProvider): string {
  const name = (index as { name?: unknown }).name
  if (typeof name === 'string') return name
  if (!warnedMissingName.has(index)) {
    warnedMissingName.add(index)
    console.warn(
      '[vector-index] provider is missing the required `name` field (VectorIndexProvider.name, ' +
        'required since 8.10.0) — stamping "unknown-provider" in transaction op names until the ' +
        'provider declares its own identity.'
    )
  }
  return 'unknown-provider'
}

/**
 * Add to the vector index with rollback support.
 *
 * Backend-neutral: `index` is whatever the `'vector'` provider factory
 * returns — Brainy's own JS HNSW fallback, or a native acceleration provider
 * (e.g. DiskANN). The emitted `name` stamps the active backend (see
 * {@link resolveVectorProviderId}) so operators reading a transaction journal
 * or timing trace see which engine actually ran, never a fossil name from
 * whichever engine happened to be active when this op class was written.
 *
 * Rollback strategy:
 * - Remove item from index
 */
export class AddToVectorIndexOperation implements Operation {
  readonly name: string

  constructor(
    private readonly index: VectorIndexProvider,
    private readonly id: string,
    private readonly vector: number[]
  ) {
    this.name = `AddToVectorIndex(${resolveVectorProviderId(index)})`
  }

  async execute(): Promise<RollbackAction> {
    // Check if item already exists (for rollback decision)
    const existed = await this.itemExists(this.id)

    // Add to index
    await this.index.addItem({ id: this.id, vector: this.vector })

    // Return rollback action
    return async () => {
      if (!existed) {
        // Remove newly added item
        await this.index.removeItem(this.id)
      }
      // If item existed before, we don't rollback (update is OK)
      // This prevents index corruption from removing pre-existing items
    }
  }

  /**
   * Check if item exists in index.
   *
   * `getItem` is an optional, feature-detected provider capability — see the
   * VectorIndexProvider docs; it is intentionally absent from the required
   * contract (Brainy's JS HNSW index omits it). When the capability is
   * missing the answer must be `false`, not `true`: treating unknowable
   * pre-existence as "existed" made every rollback skip removeItem, leaving
   * phantom entries in the index after a failed transaction. The safe default
   * is to remove what this operation added — update flows pair this op with a
   * RemoveFromVectorIndexOperation whose own rollback restores the prior
   * vector, so reverse-order rollback reconstructs the original state either
   * way.
   */
  private async itemExists(id: string): Promise<boolean> {
    const index = this.index as VectorIndexProvider & {
      getItem?: (id: string) => Promise<unknown>
    }
    if (typeof index.getItem !== 'function') return false
    try {
      const item = await index.getItem(id)
      return item !== undefined && item !== null
    } catch {
      return false
    }
  }
}

/**
 * Remove from the vector index with rollback support.
 *
 * Backend-neutral: see {@link AddToVectorIndexOperation} — `index` may be the
 * JS HNSW fallback or a native acceleration provider; the emitted `name`
 * stamps the active backend.
 *
 * Rollback strategy:
 * - Re-add item to index with original vector
 *
 * Note: Requires storing the vector for rollback
 */
export class RemoveFromVectorIndexOperation implements Operation {
  readonly name: string

  constructor(
    private readonly index: VectorIndexProvider,
    private readonly id: string,
    private readonly vector: number[] // Required for rollback
  ) {
    this.name = `RemoveFromVectorIndex(${resolveVectorProviderId(index)})`
  }

  async execute(): Promise<RollbackAction> {
    // Remove from index
    await this.index.removeItem(this.id)

    // Return rollback action
    return async () => {
      // Re-add item with original vector
      await this.index.addItem({ id: this.id, vector: this.vector })
    }
  }
}

/**
 * Replace an item's vector in the vector index as ONE atomic transaction leg —
 * the row is never absent from vector search during an update.
 *
 * Backend-neutral: see {@link AddToVectorIndexOperation} — `index` may be the
 * JS HNSW fallback or a native acceleration provider; the emitted `name`
 * stamps the active backend.
 *
 * Why this op exists: update flows historically staged a
 * {@link RemoveFromVectorIndexOperation} followed by an
 * {@link AddToVectorIndexOperation} as two separately-awaited operations.
 * Between them the row was in NEITHER index — dark to semantic recall while
 * perfectly visible to metadata reads (a transient-invisibility window that
 * stretched to seconds in a production deployment). The structural cure is a
 * single leg that never removes without simultaneously re-inserting.
 *
 * Execution strategy (feature-detected, in preference order):
 * 1. Provider exposes `updateItem` → ONE in-place call. The provider swaps
 *    the vector without the row ever leaving its index, and an element-wise
 *    UNCHANGED vector (the type-only-update production shape) is a pure
 *    no-op on its side.
 * 2. Provider without `updateItem` (a native provider that has not shipped
 *    it yet) → `removeItem` + `addItem` executed ADJACENT within this single
 *    op. Still strictly better than the historical pair: no other transaction
 *    operation can interleave between the two calls. This is a temporary
 *    seam — the native side of the pair is expected to ship its own
 *    `updateItem` so path 1 applies everywhere; when it does, this fallback
 *    becomes dead code that costs nothing.
 *
 * Rollback strategy (mirrors the execute branch that ran):
 * - `updateItem` path → `updateItem` back to `oldVector`.
 * - Fallback path → `removeItem` + `addItem` back to `oldVector`.
 *
 * Rollback semantics when the item did not exist at execute time: this op's
 * contract is that the caller read the entity and its CURRENT vector
 * (`oldVector`) before staging — update flows only stage it for existing
 * rows. If the item was somehow absent, execute() inserts it (`updateItem`
 * delegates to add; the fallback's remove is a no-op before its add), and
 * rollback restores `oldVector` rather than removing — the same posture as
 * {@link RemoveFromVectorIndexOperation}'s unconditional re-add: by
 * constructing the op with `oldVector` the caller DECLARED the before-state,
 * and rollback reconstructs that declared state instead of silently deciding
 * the row should vanish.
 */
export class ReplaceInVectorIndexOperation implements Operation {
  readonly name: string

  constructor(
    private readonly index: VectorIndexProvider,
    private readonly id: string,
    private readonly oldVector: number[], // Required for rollback
    private readonly newVector: number[]
  ) {
    this.name = `ReplaceInVectorIndex(${resolveVectorProviderId(index)})`
  }

  async execute(): Promise<RollbackAction> {
    // Feature-detect the in-place capability — optional on the provider
    // contract, like `getItem`/`setPersistMode` (Brainy's JS HNSW index
    // ships it; a native provider may not have yet).
    const index = this.index as VectorIndexProvider & {
      updateItem?: (item: { id: string; vector: number[] }) => Promise<void>
    }

    if (typeof index.updateItem === 'function') {
      // Atomic path: one in-place call, the row never leaves the index.
      await index.updateItem({ id: this.id, vector: this.newVector })

      return async () => {
        // Restore the declared before-state in place (see class JSDoc for
        // the item-did-not-exist posture).
        await index.updateItem!({ id: this.id, vector: this.oldVector })
      }
    }

    // Fallback seam: remove+add ADJACENT within this single op — no other
    // transaction operation can interleave between them (see class JSDoc).
    await this.index.removeItem(this.id)
    await this.index.addItem({ id: this.id, vector: this.newVector })

    return async () => {
      // updateItem-style restore via the same adjacent pair, back to the
      // declared before-state.
      await this.index.removeItem(this.id)
      await this.index.addItem({ id: this.id, vector: this.oldVector })
    }
  }
}

/**
 * Add to metadata index with rollback support
 *
 * Rollback strategy:
 * - Remove item from index
 */
export class AddToMetadataIndexOperation implements Operation {
  readonly name = 'AddToMetadataIndex'

  constructor(
    private readonly index: MetadataIndexManager,
    private readonly id: string,
    private readonly entity: any // Entity or metadata structure
  ) {}

  async execute(): Promise<RollbackAction> {
    // Add to metadata index (skipFlush=true for transaction atomicity)
    await this.index.addToIndex(this.id, this.entity, true)

    // Return rollback action
    return async () => {
      // Remove from metadata index
      await this.index.removeFromIndex(this.id, this.entity)
    }
  }
}

/**
 * Remove from metadata index with rollback support
 *
 * Rollback strategy:
 * - Re-add item to index with original metadata
 */
export class RemoveFromMetadataIndexOperation implements Operation {
  readonly name = 'RemoveFromMetadataIndex'

  constructor(
    private readonly index: MetadataIndexManager,
    private readonly id: string,
    private readonly entity: any // Required for rollback
  ) {}

  async execute(): Promise<RollbackAction> {
    // Remove from metadata index
    await this.index.removeFromIndex(this.id, this.entity)

    // Return rollback action
    return async () => {
      // Re-add with original metadata (skipFlush=true)
      await this.index.addToIndex(this.id, this.entity, true)
    }
  }
}

/**
 * Add verb to graph index with rollback support
 *
 * Rollback strategy:
 * - Remove verb from graph index
 *
 * 8.0 u64 contract: the coordinator resolves both endpoint ints via the
 * shared idMapper (`getOrAssign`) and passes them alongside the verb; the
 * provider returns the interned verb int, which is surfaced through the
 * optional `onVerbInt` callback so the coordinator can feed its warm cache.
 *
 * Generation: `generationFn` is resolved at execute time (not construction) so
 * the edge is stamped at the transaction's in-flight commit generation — which
 * the generation store only assigns once the batch begins executing. The same
 * generation is reused for the rollback removal, so an add and its undo
 * reference one watermark in a provider's per-generation edge chain.
 */
/**
 * A verb's interned endpoint ints — eager (already resolved), or a thunk
 * evaluated when the operation EXECUTES. The lazy form exists for
 * `transact()` forward references: a relate whose endpoint is added in the
 * SAME batch cannot resolve ints at plan time (the entity does not exist yet
 * — a strict native id mapper rightly refuses to assign, and even a permissive
 * one would leak the assignment if the batch is rejected at precommit).
 * Deferring to execute time resolves after the batch's add operations have
 * applied, mirroring how `generationFn` is already evaluated lazily.
 */
export type VerbEndpointInts =
  | { sourceInt: bigint; targetInt: bigint }
  | (() => { sourceInt: bigint; targetInt: bigint })

/** Resolve a {@link VerbEndpointInts} at execution time. */
function resolveEndpointInts(
  endpoints: VerbEndpointInts
): { sourceInt: bigint; targetInt: bigint } {
  return typeof endpoints === 'function' ? endpoints() : endpoints
}

export class AddToGraphIndexOperation implements Operation {
  readonly name = 'AddToGraphIndex'

  /**
   * @param index - The graph-index provider (JS baseline or native).
   * @param verb - The verb to index (`sourceInt`/`targetInt` mirrored on it).
   * @param endpointInts - The endpoints' interned ints — eager, or a thunk
   *   evaluated at execute time (REQUIRED for transact forward references;
   *   see {@link VerbEndpointInts}).
   * @param generationFn - Resolves the commit generation to stamp this edge at,
   *   evaluated when the operation executes (see class note).
   * @param onVerbInt - Optional hook invoked with the interned verb int
   *   returned by the provider (feeds the coordinator's verb-int warm cache).
   */
  constructor(
    private readonly index: GraphIndexProvider,
    private readonly verb: GraphVerb,
    private readonly endpointInts: VerbEndpointInts,
    private readonly generationFn: () => bigint,
    private readonly onVerbInt?: (verbInt: bigint) => void
  ) {}

  async execute(): Promise<RollbackAction> {
    // Stamp this edge at the in-flight commit generation; reuse it for the
    // rollback so add + undo reference the same watermark. Endpoint ints
    // resolve HERE — after any same-batch adds have applied.
    const generation = this.generationFn()
    const { sourceInt, targetInt } = resolveEndpointInts(this.endpointInts)
    const verbInt = await this.index.addVerb(this.verb, sourceInt, targetInt, generation)
    this.onVerbInt?.(verbInt)

    // Return rollback action
    return async () => {
      // Remove verb from graph index
      await this.index.removeVerb(this.verb.id, generation)
    }
  }
}

/**
 * Remove verb from graph index with rollback support
 *
 * Rollback strategy:
 * - Re-add verb to graph index
 *
 * 8.0 u64 contract: rollback re-adds through `addVerb(verb, sourceInt,
 * targetInt, generation)`, so the coordinator resolves the endpoint ints up
 * front (while the entity → int mappings are guaranteed to still exist). The
 * removal generation is resolved at execute time and reused for the rollback
 * re-add, so the round trip references one watermark.
 */
export class RemoveFromGraphIndexOperation implements Operation {
  readonly name = 'RemoveFromGraphIndex'

  /**
   * @param index - The graph-index provider (JS baseline or native).
   * @param verb - The verb being removed (required for rollback re-add).
   * @param endpointInts - The endpoints' interned ints for the rollback
   *   re-add — eager, or a thunk evaluated at execute time (required when the
   *   verb or its endpoints were created in the SAME transact batch; see
   *   {@link VerbEndpointInts}).
   * @param generationFn - Resolves the commit generation for this removal,
   *   evaluated when the operation executes.
   */
  constructor(
    private readonly index: GraphIndexProvider,
    private readonly verb: GraphVerb, // Required for rollback
    private readonly endpointInts: VerbEndpointInts,
    private readonly generationFn: () => bigint
  ) {}

  async execute(): Promise<RollbackAction> {
    // Resolve the removal generation once; reuse it for the rollback re-add.
    // Endpoint ints resolve HERE (after any same-batch adds applied) and are
    // captured for the rollback, whose re-add must use the same mappings.
    const generation = this.generationFn()
    const { sourceInt, targetInt } = resolveEndpointInts(this.endpointInts)
    await this.index.removeVerb(this.verb.id, generation)

    // Return rollback action
    return async () => {
      // Re-add verb with original data
      await this.index.addVerb(this.verb, sourceInt, targetInt, generation)
    }
  }
}

/**
 * Batch operation: Add multiple items to the vector index (backend-neutral —
 * see {@link AddToVectorIndexOperation}).
 *
 * Useful for bulk imports with transaction support.
 * Rolls back all items if any fail.
 */
export class BatchAddToVectorIndexOperation implements Operation {
  readonly name: string

  private operations: AddToVectorIndexOperation[]

  constructor(
    index: VectorIndexProvider,
    items: Array<{ id: string; vector: number[] }>
  ) {
    this.name = `BatchAddToVectorIndex(${resolveVectorProviderId(index)})`
    this.operations = items.map(
      item => new AddToVectorIndexOperation(index, item.id, item.vector)
    )
  }

  async execute(): Promise<RollbackAction> {
    const rollbackActions: RollbackAction[] = []

    // Execute all operations
    for (const op of this.operations) {
      const rollback = await op.execute()
      if (rollback) {
        rollbackActions.push(rollback)
      }
    }

    // Return combined rollback action
    return async () => {
      // Execute all rollbacks in reverse order
      for (let i = rollbackActions.length - 1; i >= 0; i--) {
        await rollbackActions[i]()
      }
    }
  }
}

/**
 * Batch operation: Add multiple entities to metadata index
 *
 * Useful for bulk imports with transaction support.
 */
export class BatchAddToMetadataIndexOperation implements Operation {
  readonly name = 'BatchAddToMetadataIndex'

  private operations: AddToMetadataIndexOperation[]

  constructor(
    index: MetadataIndexManager,
    items: Array<{ id: string; entity: any }>
  ) {
    this.operations = items.map(
      item => new AddToMetadataIndexOperation(index, item.id, item.entity)
    )
  }

  async execute(): Promise<RollbackAction> {
    const rollbackActions: RollbackAction[] = []

    // Execute all operations
    for (const op of this.operations) {
      const rollback = await op.execute()
      if (rollback) {
        rollbackActions.push(rollback)
      }
    }

    // Return combined rollback action
    return async () => {
      // Execute all rollbacks in reverse order
      for (let i = rollbackActions.length - 1; i >= 0; i--) {
        await rollbackActions[i]()
      }
    }
  }
}