2025-08-26 12:32:21 -07:00
/ * *
* Metadata Index System
* Maintains inverted indexes for fast metadata filtering
* Automatically updates indexes when data changes
* /
import { StorageAdapter } from '../coreTypes.js'
import { MetadataIndexCache , MetadataIndexCacheConfig } from './metadataIndexCache.js'
import { prodLog } from './logger.js'
import { getGlobalCache , UnifiedCache } from './unifiedCache.js'
export interface MetadataIndexEntry {
field : string
value : string | number | boolean
ids : Set < string >
lastUpdated : number
}
export interface FieldIndexData {
// Maps value -> count for quick filter discovery
values : Record < string , number >
lastUpdated : number
}
export interface MetadataIndexStats {
totalEntries : number
totalIds : number
fieldsIndexed : string [ ]
lastRebuild : number
indexSize : number // in bytes
}
export interface MetadataIndexConfig {
maxIndexSize? : number // Max number of entries per field value (default: 10000)
rebuildThreshold? : number // Rebuild if index is this % stale (default: 0.1)
autoOptimize? : boolean // Auto-cleanup unused entries (default: true)
indexedFields? : string [ ] // Only index these fields (default: all)
excludeFields? : string [ ] // Never index these fields
}
/ * *
* Manages metadata indexes for fast filtering
* Maintains inverted indexes : field + value - > list of IDs
* /
// Sorted index for range queries
interface SortedFieldIndex {
values : Array < [ value : any , ids : Set < string > ] >
isDirty : boolean
fieldType : 'number' | 'string' | 'date' | 'mixed'
}
export class MetadataIndexManager {
private storage : StorageAdapter
private config : Required < MetadataIndexConfig >
private indexCache = new Map < string , MetadataIndexEntry > ( )
private dirtyEntries = new Set < string > ( )
private isRebuilding = false
private metadataCache : MetadataIndexCache
private fieldIndexes = new Map < string , FieldIndexData > ( )
private dirtyFields = new Set < string > ( )
private lastFlushTime = Date . now ( )
private autoFlushThreshold = 10 // Start with 10 for more frequent non-blocking flushes
// Sorted indices for range queries (only for numeric/date fields)
private sortedIndices = new Map < string , SortedFieldIndex > ( )
private numericFields = new Set < string > ( ) // Track which fields are numeric
// Unified cache for coordinated memory management
private unifiedCache : UnifiedCache
constructor ( storage : StorageAdapter , config : MetadataIndexConfig = { } ) {
this . storage = storage
this . config = {
maxIndexSize : config.maxIndexSize ? ? 10000 ,
rebuildThreshold : config.rebuildThreshold ? ? 0.1 ,
autoOptimize : config.autoOptimize ? ? true ,
indexedFields : config.indexedFields ? ? [ ] ,
excludeFields : config.excludeFields ? ? [ 'id' , 'createdAt' , 'updatedAt' , 'embedding' , 'vector' , 'embeddings' , 'vectors' ]
}
// Initialize metadata cache with similar config to search cache
this . metadataCache = new MetadataIndexCache ( {
maxAge : 5 * 60 * 1000 , // 5 minutes
maxSize : 500 , // 500 entries (field indexes + value chunks)
enabled : true
} )
// Get global unified cache for coordinated memory management
this . unifiedCache = getGlobalCache ( )
}
/ * *
* Get index key for field and value
* /
private getIndexKey ( field : string , value : any ) : string {
const normalizedValue = this . normalizeValue ( value )
return ` ${ field } : ${ normalizedValue } `
}
/ * *
* Ensure sorted index exists for a field ( for range queries )
* /
private async ensureSortedIndex ( field : string ) : Promise < void > {
if ( ! this . sortedIndices . has ( field ) ) {
// Try to load from storage first
const loaded = await this . loadSortedIndex ( field )
if ( loaded ) {
this . sortedIndices . set ( field , loaded )
} else {
// Create new sorted index
this . sortedIndices . set ( field , {
values : [ ] ,
isDirty : true ,
fieldType : 'mixed'
} )
}
}
}
/ * *
* Build sorted index for a field from hash index
* /
private async buildSortedIndex ( field : string ) : Promise < void > {
const sortedIndex = this . sortedIndices . get ( field )
if ( ! sortedIndex || ! sortedIndex . isDirty ) return
// Collect all values for this field from hash index
const valueMap = new Map < any , Set < string > > ( )
for ( const [ key , entry ] of this . indexCache . entries ( ) ) {
if ( entry . field === field ) {
const existing = valueMap . get ( entry . value )
if ( existing ) {
// Merge ID sets
entry . ids . forEach ( id = > existing . add ( id ) )
} else {
valueMap . set ( entry . value , new Set ( entry . ids ) )
}
}
}
// Convert to sorted array
const sorted = Array . from ( valueMap . entries ( ) )
// Detect field type and sort accordingly
if ( sorted . length > 0 ) {
const sampleValue = sorted [ 0 ] [ 0 ]
if ( typeof sampleValue === 'number' ) {
sortedIndex . fieldType = 'number'
sorted . sort ( ( a , b ) = > a [ 0 ] - b [ 0 ] )
} else if ( sampleValue instanceof Date ) {
sortedIndex . fieldType = 'date'
sorted . sort ( ( a , b ) = > a [ 0 ] . getTime ( ) - b [ 0 ] . getTime ( ) )
} else {
sortedIndex . fieldType = 'string'
sorted . sort ( ( a , b ) = > {
const aVal = String ( a [ 0 ] )
const bVal = String ( b [ 0 ] )
return aVal < bVal ? - 1 : aVal > bVal ? 1 : 0
} )
}
}
sortedIndex . values = sorted
sortedIndex . isDirty = false
}
/ * *
* Binary search for range start ( inclusive or exclusive )
* /
private binarySearchStart ( sorted : Array < [ any , Set < string > ] > , target : any , inclusive : boolean ) : number {
let left = 0
let right = sorted . length - 1
let result = sorted . length
while ( left <= right ) {
const mid = Math . floor ( ( left + right ) / 2 )
const midVal = sorted [ mid ] [ 0 ]
if ( inclusive ? midVal >= target : midVal > target ) {
result = mid
right = mid - 1
} else {
left = mid + 1
}
}
return result
}
/ * *
* Binary search for range end ( inclusive or exclusive )
* /
private binarySearchEnd ( sorted : Array < [ any , Set < string > ] > , target : any , inclusive : boolean ) : number {
let left = 0
let right = sorted . length - 1
let result = - 1
while ( left <= right ) {
const mid = Math . floor ( ( left + right ) / 2 )
const midVal = sorted [ mid ] [ 0 ]
if ( inclusive ? midVal <= target : midVal < target ) {
result = mid
left = mid + 1
} else {
right = mid - 1
}
}
return result
}
/ * *
* Get IDs matching a range query
* /
private async getIdsForRange (
field : string ,
min? : any ,
max? : any ,
includeMin : boolean = true ,
includeMax : boolean = true
) : Promise < string [ ] > {
// Ensure sorted index exists and is up to date
await this . ensureSortedIndex ( field )
await this . buildSortedIndex ( field )
const sortedIndex = this . sortedIndices . get ( field )
if ( ! sortedIndex || sortedIndex . values . length === 0 ) return [ ]
const sorted = sortedIndex . values
const resultSet = new Set < string > ( )
// Find range boundaries
let start = 0
let end = sorted . length - 1
if ( min !== undefined ) {
start = this . binarySearchStart ( sorted , min , includeMin )
}
if ( max !== undefined ) {
end = this . binarySearchEnd ( sorted , max , includeMax )
}
// Collect all IDs in range
for ( let i = start ; i <= end && i < sorted . length ; i ++ ) {
const [ , ids ] = sorted [ i ]
ids . forEach ( id = > resultSet . add ( id ) )
}
return Array . from ( resultSet )
}
/ * *
* Generate field index filename for filter discovery
* /
private getFieldIndexFilename ( field : string ) : string {
return ` field_ ${ field } `
}
/ * *
* Generate value chunk filename for scalable storage
* /
private getValueChunkFilename ( field : string , value : any , chunkIndex : number = 0 ) : string {
const normalizedValue = this . normalizeValue ( value )
const safeValue = this . makeSafeFilename ( normalizedValue )
return ` ${ field } _ ${ safeValue } _chunk ${ chunkIndex } `
}
/ * *
* Make a value safe for use in filenames
* /
private makeSafeFilename ( value : string ) : string {
// Replace unsafe characters and limit length
return value
. replace ( /[^a-zA-Z0-9-_]/g , '_' )
. substring ( 0 , 50 )
. toLowerCase ( )
}
/ * *
* Normalize value for consistent indexing
* /
private normalizeValue ( value : any ) : string {
if ( value === null || value === undefined ) return '__NULL__'
if ( typeof value === 'boolean' ) return value ? '__TRUE__' : '__FALSE__'
if ( typeof value === 'number' ) return value . toString ( )
if ( Array . isArray ( value ) ) {
const joined = value . map ( v = > this . normalizeValue ( v ) ) . join ( ',' )
// Hash very long array values to avoid filesystem limits
if ( joined . length > 100 ) {
return this . hashValue ( joined )
}
return joined
}
const stringValue = String ( value ) . toLowerCase ( ) . trim ( )
// Hash very long string values to avoid filesystem limits
if ( stringValue . length > 100 ) {
return this . hashValue ( stringValue )
}
return stringValue
}
/ * *
* Create a short hash for long values to avoid filesystem filename limits
* /
private hashValue ( value : string ) : string {
// Simple hash function to create shorter keys
let hash = 0
for ( let i = 0 ; i < value . length ; i ++ ) {
const char = value . charCodeAt ( i )
hash = ( ( hash << 5 ) - hash ) + char
hash = hash & hash // Convert to 32-bit integer
}
return ` __HASH_ ${ Math . abs ( hash ) . toString ( 36 ) } `
}
/ * *
* Check if field should be indexed
* /
private shouldIndexField ( field : string ) : boolean {
if ( this . config . excludeFields . includes ( field ) ) return false
if ( this . config . indexedFields . length > 0 ) {
return this . config . indexedFields . includes ( field )
}
return true
}
/ * *
* Extract indexable field - value pairs from metadata
* /
private extractIndexableFields ( metadata : any ) : Array < { field : string , value : any } > {
const fields : Array < { field : string , value : any } > = [ ]
const extract = ( obj : any , prefix = '' ) : void = > {
for ( const [ key , value ] of Object . entries ( obj ) ) {
const fullKey = prefix ? ` ${ prefix } . ${ key } ` : key
if ( ! this . shouldIndexField ( fullKey ) ) continue
if ( value && typeof value === 'object' && ! Array . isArray ( value ) ) {
// Recurse into nested objects
extract ( value , fullKey )
} else {
// Index this field
fields . push ( { field : fullKey , value } )
// If it's an array, also index each element
if ( Array . isArray ( value ) ) {
for ( const item of value ) {
fields . push ( { field : fullKey , value : item } )
}
}
}
}
}
if ( metadata && typeof metadata === 'object' ) {
extract ( metadata )
}
return fields
}
/ * *
* Add item to metadata indexes
* /
async addToIndex ( id : string , metadata : any , skipFlush : boolean = false ) : Promise < void > {
const fields = this . extractIndexableFields ( metadata )
// Mark sorted indices as dirty when adding new data
for ( const { field } of fields ) {
const sortedIndex = this . sortedIndices . get ( field )
if ( sortedIndex ) {
sortedIndex . isDirty = true
}
}
for ( let i = 0 ; i < fields . length ; i ++ ) {
const { field , value } = fields [ i ]
const key = this . getIndexKey ( field , value )
// Get or create index entry
let entry = this . indexCache . get ( key )
if ( ! entry ) {
const loadedEntry = await this . loadIndexEntry ( key )
entry = loadedEntry ? ? {
field ,
value : this.normalizeValue ( value ) ,
ids : new Set < string > ( ) ,
lastUpdated : Date.now ( )
}
this . indexCache . set ( key , entry )
}
// Add ID to entry
entry . ids . add ( id )
entry . lastUpdated = Date . now ( )
this . dirtyEntries . add ( key )
// Update field index
await this . updateFieldIndex ( field , value , 1 )
// Yield to event loop every 5 fields to prevent blocking
if ( i % 5 === 4 ) {
await this . yieldToEventLoop ( )
}
}
// Adaptive auto-flush based on usage patterns
if ( ! skipFlush ) {
const timeSinceLastFlush = Date . now ( ) - this . lastFlushTime
const shouldAutoFlush =
this . dirtyEntries . size >= this . autoFlushThreshold || // Size threshold
( this . dirtyEntries . size > 10 && timeSinceLastFlush > 5000 ) // Time threshold (5 seconds)
if ( shouldAutoFlush ) {
const startTime = Date . now ( )
await this . flush ( )
const flushTime = Date . now ( ) - startTime
// Adapt threshold based on flush performance
if ( flushTime < 50 ) {
// Fast flush, can handle more entries
this . autoFlushThreshold = Math . min ( 200 , this . autoFlushThreshold * 1.2 )
} else if ( flushTime > 200 ) {
// Slow flush, reduce batch size
this . autoFlushThreshold = Math . max ( 20 , this . autoFlushThreshold * 0.8 )
}
// Yield to event loop after flush to prevent blocking
await this . yieldToEventLoop ( )
}
}
// Invalidate cache for these fields
for ( const { field } of fields ) {
this . metadataCache . invalidatePattern ( ` field_values_ ${ field } ` )
}
}
/ * *
* Update field index with value count
* /
private async updateFieldIndex ( field : string , value : any , delta : number ) : Promise < void > {
let fieldIndex = this . fieldIndexes . get ( field )
if ( ! fieldIndex ) {
// Load from storage if not in memory
fieldIndex = await this . loadFieldIndex ( field ) ? ? {
values : { } ,
lastUpdated : Date.now ( )
}
this . fieldIndexes . set ( field , fieldIndex )
}
const normalizedValue = this . normalizeValue ( value )
fieldIndex . values [ normalizedValue ] = ( fieldIndex . values [ normalizedValue ] || 0 ) + delta
// Remove if count drops to 0
if ( fieldIndex . values [ normalizedValue ] <= 0 ) {
delete fieldIndex . values [ normalizedValue ]
}
fieldIndex . lastUpdated = Date . now ( )
this . dirtyFields . add ( field )
}
/ * *
* Remove item from metadata indexes
* /
async removeFromIndex ( id : string , metadata? : any ) : Promise < void > {
if ( metadata ) {
// Remove from specific field indexes
const fields = this . extractIndexableFields ( metadata )
for ( const { field , value } of fields ) {
const key = this . getIndexKey ( field , value )
let entry = this . indexCache . get ( key )
if ( ! entry ) {
const loadedEntry = await this . loadIndexEntry ( key )
entry = loadedEntry ? ? undefined
}
if ( entry ) {
entry . ids . delete ( id )
entry . lastUpdated = Date . now ( )
this . dirtyEntries . add ( key )
// Update field index
await this . updateFieldIndex ( field , value , - 1 )
// If no IDs left, mark for cleanup
if ( entry . ids . size === 0 ) {
this . indexCache . delete ( key )
await this . deleteIndexEntry ( key )
}
}
// Invalidate cache
this . metadataCache . invalidatePattern ( ` field_values_ ${ field } ` )
}
} else {
// Remove from all indexes (slower, requires scanning)
for ( const [ key , entry ] of this . indexCache . entries ( ) ) {
if ( entry . ids . has ( id ) ) {
entry . ids . delete ( id )
entry . lastUpdated = Date . now ( )
this . dirtyEntries . add ( key )
if ( entry . ids . size === 0 ) {
this . indexCache . delete ( key )
await this . deleteIndexEntry ( key )
}
}
}
}
}
2025-08-27 15:38:48 -07:00
/ * *
* Get all IDs in the index
* /
async getAllIds ( ) : Promise < string [ ] > {
// Collect all unique IDs from all index entries
const allIds = new Set < string > ( )
// First, add all IDs from the in-memory cache
for ( const entry of this . indexCache . values ( ) ) {
entry . ids . forEach ( id = > allIds . add ( id ) )
}
// If storage has a method to get all nouns, use it as the source of truth
// This ensures we include items that might not be indexed yet
if ( this . storage && typeof ( this . storage as any ) . getNouns === 'function' ) {
try {
const result = await ( this . storage as any ) . getNouns ( {
pagination : { limit : 100000 }
} )
if ( result && result . items ) {
result . items . forEach ( ( item : any ) = > {
if ( item . id ) allIds . add ( item . id )
} )
}
} catch ( e ) {
// Fall back to using only indexed IDs
}
}
return Array . from ( allIds )
}
2025-08-26 12:32:21 -07:00
/ * *
* Get IDs for a specific field - value combination with caching
* /
async getIds ( field : string , value : any ) : Promise < string [ ] > {
const key = this . getIndexKey ( field , value )
// Check metadata cache first
const cacheKey = ` ids_ ${ key } `
const cachedIds = this . metadataCache . get ( cacheKey )
if ( cachedIds ) {
return cachedIds
}
// Try in-memory cache
let entry = this . indexCache . get ( key )
// Load from storage if not cached
if ( ! entry ) {
const loadedEntry = await this . loadIndexEntry ( key )
if ( loadedEntry ) {
entry = loadedEntry
this . indexCache . set ( key , entry )
}
}
const ids = entry ? Array . from ( entry . ids ) : [ ]
// Cache the result
this . metadataCache . set ( cacheKey , ids )
return ids
}
/ * *
* Get all available values for a field ( for filter discovery )
* /
async getFilterValues ( field : string ) : Promise < string [ ] > {
// Check cache first
const cacheKey = ` field_values_ ${ field } `
const cachedValues = this . metadataCache . get ( cacheKey )
if ( cachedValues ) {
return cachedValues
}
// Check in-memory field indexes first
let fieldIndex = this . fieldIndexes . get ( field )
// If not in memory, load from storage
if ( ! fieldIndex ) {
const loaded = await this . loadFieldIndex ( field )
if ( loaded ) {
fieldIndex = loaded
this . fieldIndexes . set ( field , loaded )
}
}
if ( ! fieldIndex ) {
return [ ]
}
const values = Object . keys ( fieldIndex . values )
// Cache the result
this . metadataCache . set ( cacheKey , values )
return values
}
/ * *
* Get all indexed fields ( for filter discovery )
* /
async getFilterFields ( ) : Promise < string [ ] > {
// Check cache first
const cacheKey = 'all_filter_fields'
const cachedFields = this . metadataCache . get ( cacheKey )
if ( cachedFields ) {
return cachedFields
}
// Get fields from in-memory indexes and storage
const fields = new Set < string > ( this . fieldIndexes . keys ( ) )
// Also scan storage for persisted field indexes (in case not loaded)
// This would require a new storage method to list field indexes
// For now, just use in-memory fields
const fieldsArray = Array . from ( fields )
// Cache the result
this . metadataCache . set ( cacheKey , fieldsArray )
return fieldsArray
}
/ * *
* Convert Brainy Field Operator filter to simple field - value criteria for indexing
* /
private convertFilterToCriteria ( filter : any ) : Array < { field : string , values : any [ ] } > {
const criteria : Array < { field : string , values : any [ ] } > = [ ]
if ( ! filter || typeof filter !== 'object' ) {
return criteria
}
for ( const [ key , value ] of Object . entries ( filter ) ) {
// Skip logical operators for now - handle them separately
if ( key === 'allOf' || key === 'anyOf' || key === 'not' ) continue
if ( value && typeof value === 'object' && ! Array . isArray ( value ) ) {
// Handle Brainy Field Operators
for ( const [ op , operand ] of Object . entries ( value ) ) {
switch ( op ) {
case 'oneOf' :
if ( Array . isArray ( operand ) ) {
criteria . push ( { field : key , values : operand } )
}
break
case 'equals' :
case 'is' :
case 'eq' :
criteria . push ( { field : key , values : [ operand ] } )
break
case 'contains' :
// For contains, the operand is the value we're looking for in an array field
criteria . push ( { field : key , values : [ operand ] } )
break
case 'greaterThan' :
case 'lessThan' :
case 'greaterEqual' :
case 'lessEqual' :
case 'between' :
// Range queries will be handled separately
// Sorted index will be created/loaded when needed in getIdsForRange
break
default :
break
}
}
} else {
// Direct value or array
const values = Array . isArray ( value ) ? value : [ value ]
criteria . push ( { field : key , values } )
}
}
return criteria
}
/ * *
* Get IDs matching Brainy Field Operator metadata filter using indexes where possible
* /
async getIdsForFilter ( filter : any ) : Promise < string [ ] > {
if ( ! filter || Object . keys ( filter ) . length === 0 ) {
return [ ]
}
// Handle logical operators
if ( filter . allOf && Array . isArray ( filter . allOf ) ) {
// For allOf, we need intersection of all sub-filters
const allIds : string [ ] [ ] = [ ]
for ( const subFilter of filter . allOf ) {
const subIds = await this . getIdsForFilter ( subFilter )
allIds . push ( subIds )
}
if ( allIds . length === 0 ) return [ ]
if ( allIds . length === 1 ) return allIds [ 0 ]
// Intersection of all sets
return allIds . reduce ( ( intersection , currentSet ) = >
intersection . filter ( id = > currentSet . includes ( id ) )
)
}
if ( filter . anyOf && Array . isArray ( filter . anyOf ) ) {
// For anyOf, we need union of all sub-filters
const unionIds = new Set < string > ( )
for ( const subFilter of filter . anyOf ) {
const subIds = await this . getIdsForFilter ( subFilter )
subIds . forEach ( id = > unionIds . add ( id ) )
}
return Array . from ( unionIds )
}
// Process field filters with range support
const idSets : string [ ] [ ] = [ ]
for ( const [ field , condition ] of Object . entries ( filter ) ) {
// Skip logical operators
if ( field === 'allOf' || field === 'anyOf' || field === 'not' ) continue
let fieldResults : string [ ] = [ ]
if ( condition && typeof condition === 'object' && ! Array . isArray ( condition ) ) {
// Handle Brainy Field Operators
for ( const [ op , operand ] of Object . entries ( condition ) ) {
switch ( op ) {
// Exact match operators
case 'equals' :
case 'is' :
case 'eq' :
fieldResults = await this . getIds ( field , operand )
break
// Multiple value operators
case 'oneOf' :
case 'in' :
if ( Array . isArray ( operand ) ) {
const unionIds = new Set < string > ( )
for ( const value of operand ) {
const ids = await this . getIds ( field , value )
ids . forEach ( id = > unionIds . add ( id ) )
}
fieldResults = Array . from ( unionIds )
}
break
// Range operators
case 'greaterThan' :
case 'gt' :
fieldResults = await this . getIdsForRange ( field , operand , undefined , false , true )
break
case 'greaterEqual' :
case 'gte' :
case 'greaterThanOrEqual' :
fieldResults = await this . getIdsForRange ( field , operand , undefined , true , true )
break
case 'lessThan' :
case 'lt' :
fieldResults = await this . getIdsForRange ( field , undefined , operand , true , false )
break
case 'lessEqual' :
case 'lte' :
case 'lessThanOrEqual' :
fieldResults = await this . getIdsForRange ( field , undefined , operand , true , true )
break
case 'between' :
if ( Array . isArray ( operand ) && operand . length === 2 ) {
fieldResults = await this . getIdsForRange ( field , operand [ 0 ] , operand [ 1 ] , true , true )
}
break
// Array contains operator
case 'contains' :
fieldResults = await this . getIds ( field , operand )
break
// Existence operator
case 'exists' :
if ( operand ) {
// Get all IDs that have this field (any value)
const allIds = new Set < string > ( )
for ( const [ key , entry ] of this . indexCache . entries ( ) ) {
if ( entry . field === field ) {
entry . ids . forEach ( id = > allIds . add ( id ) )
}
}
fieldResults = Array . from ( allIds )
}
break
2025-08-27 15:38:48 -07:00
// Negation operators
case 'notEquals' :
case 'isNot' :
case 'ne' :
// For notEquals, we need all IDs EXCEPT those matching the value
// This is especially important for soft delete: deleted !== true
// should include items without a deleted field
// First, get all IDs in the database
const allItemIds = await this . getAllIds ( )
// Then get IDs that match the value we want to exclude
const excludeIds = await this . getIds ( field , operand )
const excludeSet = new Set ( excludeIds )
// Return all IDs except those to exclude
fieldResults = allItemIds . filter ( id = > ! excludeSet . has ( id ) )
break
2025-08-26 12:32:21 -07:00
}
}
} else {
// Direct value match (shorthand for equals)
fieldResults = await this . getIds ( field , condition )
}
if ( fieldResults . length > 0 ) {
idSets . push ( fieldResults )
} else {
// If any field has no matches, intersection will be empty
return [ ]
}
}
if ( idSets . length === 0 ) return [ ]
if ( idSets . length === 1 ) return idSets [ 0 ]
// Intersection of all field criteria (implicit AND)
return idSets . reduce ( ( intersection , currentSet ) = >
intersection . filter ( id = > currentSet . includes ( id ) )
)
}
/ * *
* DEPRECATED - Old implementation for backward compatibility
* /
private async getIdsForFilterOld ( filter : any ) : Promise < string [ ] > {
if ( ! filter || Object . keys ( filter ) . length === 0 ) {
return [ ]
}
// Handle logical operators
if ( filter . allOf && Array . isArray ( filter . allOf ) ) {
// For allOf, we need intersection of all sub-filters
const allIds : string [ ] [ ] = [ ]
for ( const subFilter of filter . allOf ) {
const subIds = await this . getIdsForFilter ( subFilter )
allIds . push ( subIds )
}
if ( allIds . length === 0 ) return [ ]
if ( allIds . length === 1 ) return allIds [ 0 ]
// Intersection of all sets
return allIds . reduce ( ( intersection , currentSet ) = >
intersection . filter ( id = > currentSet . includes ( id ) )
)
}
if ( filter . anyOf && Array . isArray ( filter . anyOf ) ) {
// For anyOf, we need union of all sub-filters
const unionIds = new Set < string > ( )
for ( const subFilter of filter . anyOf ) {
const subIds = await this . getIdsForFilter ( subFilter )
subIds . forEach ( id = > unionIds . add ( id ) )
}
return Array . from ( unionIds )
}
// Handle regular field filters
const criteria = this . convertFilterToCriteria ( filter )
const idSets : string [ ] [ ] = [ ]
for ( const { field , values } of criteria ) {
const unionIds = new Set < string > ( )
for ( const value of values ) {
const ids = await this . getIds ( field , value )
ids . forEach ( id = > unionIds . add ( id ) )
}
idSets . push ( Array . from ( unionIds ) )
}
if ( idSets . length === 0 ) return [ ]
if ( idSets . length === 1 ) return idSets [ 0 ]
// Intersection of all field criteria (implicit $and)
return idSets . reduce ( ( intersection , currentSet ) = >
intersection . filter ( id = > currentSet . includes ( id ) )
)
}
/ * *
* Get IDs matching multiple criteria ( intersection ) - LEGACY METHOD
* @deprecated Use getIdsForFilter instead
* /
async getIdsForCriteria ( criteria : Record < string , any > ) : Promise < string [ ] > {
return this . getIdsForFilter ( criteria )
}
/ * *
* Flush dirty entries to storage ( non - blocking version )
* /
async flush ( ) : Promise < void > {
// Check if we have anything to flush (including sorted indices)
const hasDirtySortedIndices = Array . from ( this . sortedIndices . values ( ) ) . some ( idx = > idx . isDirty )
if ( this . dirtyEntries . size === 0 && this . dirtyFields . size === 0 && ! hasDirtySortedIndices ) {
return // Nothing to flush
}
// Process in smaller batches to avoid blocking
const BATCH_SIZE = 20
const allPromises : Promise < void > [ ] = [ ]
// Flush value entries in batches
const dirtyEntriesArray = Array . from ( this . dirtyEntries )
for ( let i = 0 ; i < dirtyEntriesArray . length ; i += BATCH_SIZE ) {
const batch = dirtyEntriesArray . slice ( i , i + BATCH_SIZE )
const batchPromises = batch . map ( key = > {
const entry = this . indexCache . get ( key )
return entry ? this . saveIndexEntry ( key , entry ) : Promise . resolve ( )
} )
allPromises . push ( . . . batchPromises )
// Yield to event loop between batches
if ( i + BATCH_SIZE < dirtyEntriesArray . length ) {
await this . yieldToEventLoop ( )
}
}
// Flush field indexes in batches
const dirtyFieldsArray = Array . from ( this . dirtyFields )
for ( let i = 0 ; i < dirtyFieldsArray . length ; i += BATCH_SIZE ) {
const batch = dirtyFieldsArray . slice ( i , i + BATCH_SIZE )
const batchPromises = batch . map ( field = > {
const fieldIndex = this . fieldIndexes . get ( field )
return fieldIndex ? this . saveFieldIndex ( field , fieldIndex ) : Promise . resolve ( )
} )
allPromises . push ( . . . batchPromises )
// Yield to event loop between batches
if ( i + BATCH_SIZE < dirtyFieldsArray . length ) {
await this . yieldToEventLoop ( )
}
}
// Flush sorted indices (for range queries)
for ( const [ field , sortedIndex ] of this . sortedIndices . entries ( ) ) {
if ( sortedIndex . isDirty ) {
allPromises . push ( this . saveSortedIndex ( field , sortedIndex ) )
}
}
// Wait for all operations to complete
await Promise . all ( allPromises )
this . dirtyEntries . clear ( )
this . dirtyFields . clear ( )
this . lastFlushTime = Date . now ( )
}
/ * *
* Yield control back to the Node . js event loop
* Prevents blocking during long - running operations
* /
private async yieldToEventLoop ( ) : Promise < void > {
return new Promise ( resolve = > setImmediate ( resolve ) )
}
/ * *
* Load field index from storage
* /
private async loadFieldIndex ( field : string ) : Promise < FieldIndexData | null > {
const filename = this . getFieldIndexFilename ( field )
const unifiedKey = ` metadata:field: ${ filename } `
// Check unified cache first with loader function
return await this . unifiedCache . get ( unifiedKey , async ( ) = > {
try {
const cacheKey = ` field_index_ ${ filename } `
// Check old cache for migration
const cached = this . metadataCache . get ( cacheKey )
if ( cached ) {
// Add to unified cache
const size = JSON . stringify ( cached ) . length
this . unifiedCache . set ( unifiedKey , cached , 'metadata' , size , 1 ) // Low rebuild cost
return cached
}
// Load from storage
const indexId = ` __metadata_field_index__ ${ filename } `
const data = await this . storage . getMetadata ( indexId )
if ( data ) {
const fieldIndex = {
values : data.values || { } ,
lastUpdated : data.lastUpdated || Date . now ( )
}
// Add to unified cache
const size = JSON . stringify ( fieldIndex ) . length
this . unifiedCache . set ( unifiedKey , fieldIndex , 'metadata' , size , 1 )
// Also keep in old cache for now (transition period)
this . metadataCache . set ( cacheKey , fieldIndex )
return fieldIndex
}
} catch ( error ) {
// Field index doesn't exist yet
}
return null
} )
}
/ * *
* Save field index to storage
* /
private async saveFieldIndex ( field : string , fieldIndex : FieldIndexData ) : Promise < void > {
const filename = this . getFieldIndexFilename ( field )
const indexId = ` __metadata_field_index__ ${ filename } `
const unifiedKey = ` metadata:field: ${ filename } `
await this . storage . saveMetadata ( indexId , {
values : fieldIndex.values ,
lastUpdated : fieldIndex.lastUpdated
} )
// Update unified cache
const size = JSON . stringify ( fieldIndex ) . length
this . unifiedCache . set ( unifiedKey , fieldIndex , 'metadata' , size , 1 )
// Invalidate old cache
this . metadataCache . invalidatePattern ( ` field_index_ ${ filename } ` )
}
/ * *
* Save sorted index to storage for range queries
* /
private async saveSortedIndex ( field : string , sortedIndex : SortedFieldIndex ) : Promise < void > {
const filename = ` sorted_ ${ field } `
const indexId = ` __metadata_sorted_index__ ${ filename } `
const unifiedKey = ` metadata:sorted: ${ field } `
// Convert Set to Array for serialization
const serializable = {
values : sortedIndex.values.map ( ( [ value , ids ] ) = > [ value , Array . from ( ids ) ] ) ,
fieldType : sortedIndex.fieldType ,
lastUpdated : Date.now ( )
}
await this . storage . saveMetadata ( indexId , serializable )
// Mark as clean
sortedIndex . isDirty = false
// Update unified cache (sorted indices are expensive to rebuild)
const size = JSON . stringify ( serializable ) . length
this . unifiedCache . set ( unifiedKey , sortedIndex , 'metadata' , size , 100 ) // Higher rebuild cost
}
/ * *
* Load sorted index from storage
* /
private async loadSortedIndex ( field : string ) : Promise < SortedFieldIndex | null > {
const filename = ` sorted_ ${ field } `
const indexId = ` __metadata_sorted_index__ ${ filename } `
const unifiedKey = ` metadata:sorted: ${ field } `
// Check unified cache first
const cached = await this . unifiedCache . get ( unifiedKey , async ( ) = > {
try {
const data = await this . storage . getMetadata ( indexId )
if ( data ) {
// Convert Arrays back to Sets
const sortedIndex : SortedFieldIndex = {
values : data.values.map ( ( [ value , ids ] : [ any , string [ ] ] ) = > [ value , new Set ( ids ) ] ) ,
fieldType : data.fieldType || 'mixed' ,
isDirty : false
}
// Add to unified cache
const size = JSON . stringify ( data ) . length
this . unifiedCache . set ( unifiedKey , sortedIndex , 'metadata' , size , 100 )
return sortedIndex
}
} catch ( error ) {
// Sorted index doesn't exist yet
}
return null
} )
return cached
}
/ * *
* Get index statistics
* /
async getStats ( ) : Promise < MetadataIndexStats > {
const fields = new Set < string > ( )
let totalEntries = 0
let totalIds = 0
for ( const entry of this . indexCache . values ( ) ) {
fields . add ( entry . field )
totalEntries ++
totalIds += entry . ids . size
}
return {
totalEntries ,
totalIds ,
fieldsIndexed : Array.from ( fields ) ,
lastRebuild : 0 , // TODO: track rebuild timestamp
indexSize : totalEntries * 100 // rough estimate
}
}
/ * *
* Rebuild entire index from scratch using pagination
* Non - blocking version that yields control back to event loop
* /
async rebuild ( ) : Promise < void > {
if ( this . isRebuilding ) return
this . isRebuilding = true
try {
prodLog . info ( '🔄 Starting non-blocking metadata index rebuild with batch processing to prevent socket exhaustion...' )
prodLog . info ( ` 📊 Storage adapter: ${ this . storage . constructor . name } ` )
prodLog . info ( ` 🔧 Batch processing available: ${ ! ! this . storage . getMetadataBatch } ` )
// Clear existing indexes
this . indexCache . clear ( )
this . dirtyEntries . clear ( )
this . fieldIndexes . clear ( )
this . dirtyFields . clear ( )
// Rebuild noun metadata indexes using pagination
let nounOffset = 0
const nounLimit = 25 // Even smaller batches during initialization to prevent socket exhaustion
let hasMoreNouns = true
let totalNounsProcessed = 0
while ( hasMoreNouns ) {
const result = await this . storage . getNouns ( {
pagination : { offset : nounOffset , limit : nounLimit }
} )
// CRITICAL FIX: Use batch metadata reading to prevent socket exhaustion
const nounIds = result . items . map ( noun = > noun . id )
let metadataBatch : Map < string , any >
if ( this . storage . getMetadataBatch ) {
// Use batch reading if available (prevents socket exhaustion)
prodLog . info ( ` 📦 Processing metadata batch ${ Math . floor ( totalNounsProcessed / nounLimit ) + 1 } ( ${ nounIds . length } items)... ` )
metadataBatch = await this . storage . getMetadataBatch ( nounIds )
const successRate = ( ( metadataBatch . size / nounIds . length ) * 100 ) . toFixed ( 1 )
prodLog . info ( ` ✅ Batch loaded ${ metadataBatch . size } / ${ nounIds . length } metadata objects ( ${ successRate } % success) ` )
} else {
// Fallback to individual calls with strict concurrency control
prodLog . warn ( ` ⚠️ FALLBACK: Storage adapter missing getMetadataBatch - using individual calls with concurrency limit ` )
metadataBatch = new Map ( )
const CONCURRENCY_LIMIT = 3 // Very conservative limit
for ( let i = 0 ; i < nounIds . length ; i += CONCURRENCY_LIMIT ) {
const batch = nounIds . slice ( i , i + CONCURRENCY_LIMIT )
const batchPromises = batch . map ( async ( id ) = > {
try {
const metadata = await this . storage . getMetadata ( id )
return { id , metadata }
} catch ( error ) {
prodLog . debug ( ` Failed to read metadata for ${ id } : ` , error )
return { id , metadata : null }
}
} )
const batchResults = await Promise . all ( batchPromises )
for ( const { id , metadata } of batchResults ) {
if ( metadata ) {
metadataBatch . set ( id , metadata )
}
}
// Yield between batches to prevent socket exhaustion
await this . yieldToEventLoop ( )
}
}
// Process the metadata batch
for ( const noun of result . items ) {
const metadata = metadataBatch . get ( noun . id )
if ( metadata ) {
// Skip flush during rebuild for performance
await this . addToIndex ( noun . id , metadata , true )
}
}
// Yield after processing the entire batch
await this . yieldToEventLoop ( )
totalNounsProcessed += result . items . length
hasMoreNouns = result . hasMore
nounOffset += nounLimit
// Progress logging and event loop yield after each batch
if ( totalNounsProcessed % 100 === 0 || ! hasMoreNouns ) {
prodLog . debug ( ` 📊 Indexed ${ totalNounsProcessed } nouns... ` )
}
await this . yieldToEventLoop ( )
}
// Rebuild verb metadata indexes using pagination
let verbOffset = 0
const verbLimit = 25 // Even smaller batches during initialization to prevent socket exhaustion
let hasMoreVerbs = true
let totalVerbsProcessed = 0
while ( hasMoreVerbs ) {
const result = await this . storage . getVerbs ( {
pagination : { offset : verbOffset , limit : verbLimit }
} )
// CRITICAL FIX: Use batch verb metadata reading to prevent socket exhaustion
const verbIds = result . items . map ( verb = > verb . id )
let verbMetadataBatch : Map < string , any >
if ( ( this . storage as any ) . getVerbMetadataBatch ) {
// Use batch reading if available (prevents socket exhaustion)
verbMetadataBatch = await ( this . storage as any ) . getVerbMetadataBatch ( verbIds )
prodLog . debug ( ` 📦 Batch loaded ${ verbMetadataBatch . size } / ${ verbIds . length } verb metadata objects ` )
} else {
// Fallback to individual calls with strict concurrency control
verbMetadataBatch = new Map ( )
const CONCURRENCY_LIMIT = 3 // Very conservative limit to prevent socket exhaustion
for ( let i = 0 ; i < verbIds . length ; i += CONCURRENCY_LIMIT ) {
const batch = verbIds . slice ( i , i + CONCURRENCY_LIMIT )
const batchPromises = batch . map ( async ( id ) = > {
try {
const metadata = await this . storage . getVerbMetadata ( id )
return { id , metadata }
} catch ( error ) {
prodLog . debug ( ` Failed to read verb metadata for ${ id } : ` , error )
return { id , metadata : null }
}
} )
const batchResults = await Promise . all ( batchPromises )
for ( const { id , metadata } of batchResults ) {
if ( metadata ) {
verbMetadataBatch . set ( id , metadata )
}
}
// Yield between batches to prevent socket exhaustion
await this . yieldToEventLoop ( )
}
}
// Process the verb metadata batch
for ( const verb of result . items ) {
const metadata = verbMetadataBatch . get ( verb . id )
if ( metadata ) {
// Skip flush during rebuild for performance
await this . addToIndex ( verb . id , metadata , true )
}
}
// Yield after processing the entire batch
await this . yieldToEventLoop ( )
totalVerbsProcessed += result . items . length
hasMoreVerbs = result . hasMore
verbOffset += verbLimit
// Progress logging and event loop yield after each batch
if ( totalVerbsProcessed % 100 === 0 || ! hasMoreVerbs ) {
prodLog . debug ( ` 🔗 Indexed ${ totalVerbsProcessed } verbs... ` )
}
await this . yieldToEventLoop ( )
}
// Flush to storage with final yield
prodLog . debug ( '💾 Flushing metadata index to storage...' )
await this . flush ( )
await this . yieldToEventLoop ( )
prodLog . info ( ` ✅ Metadata index rebuild completed! Processed ${ totalNounsProcessed } nouns and ${ totalVerbsProcessed } verbs ` )
prodLog . info ( ` 🎯 Initial indexing may show minor socket timeouts - this is expected and doesn't affect data processing ` )
} finally {
this . isRebuilding = false
}
}
/ * *
* Load index entry from storage using safe filenames
* /
private async loadIndexEntry ( key : string ) : Promise < MetadataIndexEntry | null > {
const unifiedKey = ` metadata:entry: ${ key } `
// Use unified cache with loader function
return await this . unifiedCache . get ( unifiedKey , async ( ) = > {
try {
// Extract field and value from key
const [ field , value ] = key . split ( ':' , 2 )
const filename = this . getValueChunkFilename ( field , value )
// Load from metadata indexes directory with safe filename
const indexId = ` __metadata_index__ ${ filename } `
const data = await this . storage . getMetadata ( indexId )
if ( data ) {
const entry = {
field : data.field ,
value : data.value ,
ids : new Set ( data . ids || [ ] ) ,
lastUpdated : data.lastUpdated || Date . now ( )
}
// Add to unified cache (metadata entries are cheap to rebuild)
const size = JSON . stringify ( Array . from ( entry . ids ) ) . length + 100
this . unifiedCache . set ( unifiedKey , entry , 'metadata' , size , 1 )
return entry
}
} catch ( error ) {
// Index entry doesn't exist yet
}
return null
} )
}
/ * *
* Save index entry to storage using safe filenames
* /
private async saveIndexEntry ( key : string , entry : MetadataIndexEntry ) : Promise < void > {
const unifiedKey = ` metadata:entry: ${ key } `
const data = {
field : entry.field ,
value : entry.value ,
ids : Array.from ( entry . ids ) ,
lastUpdated : entry.lastUpdated
}
// Extract field and value from key for safe filename generation
const [ field , value ] = key . split ( ':' , 2 )
const filename = this . getValueChunkFilename ( field , value )
// Store metadata indexes with safe filename
const indexId = ` __metadata_index__ ${ filename } `
await this . storage . saveMetadata ( indexId , data )
// Update unified cache
const size = JSON . stringify ( data . ids ) . length + 100
this . unifiedCache . set ( unifiedKey , entry , 'metadata' , size , 1 )
}
/ * *
* Delete index entry from storage using safe filenames
* /
private async deleteIndexEntry ( key : string ) : Promise < void > {
const unifiedKey = ` metadata:entry: ${ key } `
try {
const [ field , value ] = key . split ( ':' , 2 )
const filename = this . getValueChunkFilename ( field , value )
const indexId = ` __metadata_index__ ${ filename } `
await this . storage . saveMetadata ( indexId , null )
// Remove from unified cache
this . unifiedCache . delete ( unifiedKey )
} catch ( error ) {
// Entry might not exist
}
}
}