**feat(storage): implement multi-level cache manager with dynamic tuning**
- Added a new `CacheManager` class in `cacheManager.ts` to support three-level caching strategy: - **Level 1**: Hot cache in RAM for most accessed nodes. - **Level 2**: Warm cache using OPFS, Filesystem, or S3, depending on the environment. - **Level 3**: Cold storage for longer-term data storage. - Integrated features for dynamic tuning: - Auto-detection of environment (Browser, Node.js, Worker) and memory availability. - Parameter tuning for cache size, eviction thresholds, and TTL based on usage patterns. - Enhanced support for: - LRU-based eviction in hot cache. - Batch-based operations with configurable batch sizes. - Comprehensive logging and debug outputs for cache operations. - Ensured robust fallback handling to manage storage in constrained environments. - Improved extensibility for storage adapters (warm and cold storage detection and initialization). **Purpose**: Optimize data access and storage across multiple environments with seamless scalability and dynamic parameter adjustments.
This commit is contained in:
parent
34e8cb4b19
commit
fa18e73a7f
7 changed files with 3981 additions and 2113 deletions
|
|
@ -107,6 +107,14 @@ export interface BrainyDataConfig {
|
|||
*/
|
||||
readOnly?: boolean
|
||||
|
||||
/**
|
||||
* Enable lazy loading in read-only mode
|
||||
* When true and in read-only mode, the index is not fully loaded during initialization
|
||||
* Nodes are loaded on-demand during search operations
|
||||
* This improves startup performance for large datasets
|
||||
*/
|
||||
lazyLoadInReadOnlyMode?: boolean
|
||||
|
||||
/**
|
||||
* Set the database to write-only mode
|
||||
* When true, the index is not loaded into memory and search operations will throw an error
|
||||
|
|
@ -241,6 +249,7 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
private distanceFunction: DistanceFunction
|
||||
private requestPersistentStorage: boolean
|
||||
private readOnly: boolean
|
||||
private lazyLoadInReadOnlyMode: boolean
|
||||
private writeOnly: boolean
|
||||
private storageConfig: BrainyDataConfig['storage'] = {}
|
||||
private useOptimizedIndex: boolean = false
|
||||
|
|
@ -303,8 +312,14 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
this.distanceFunction = config.distanceFunction || cosineDistance
|
||||
|
||||
// Always use the optimized HNSW index implementation
|
||||
// Configure HNSW with disk-based storage when a storage adapter is provided
|
||||
const hnswConfig = config.hnsw || {}
|
||||
if (config.storageAdapter) {
|
||||
hnswConfig.useDiskBasedIndex = true
|
||||
}
|
||||
|
||||
this.index = new HNSWIndexOptimized(
|
||||
config.hnsw || {},
|
||||
hnswConfig,
|
||||
this.distanceFunction,
|
||||
config.storageAdapter || null
|
||||
)
|
||||
|
|
@ -337,6 +352,9 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
// Set read-only flag
|
||||
this.readOnly = config.readOnly || false
|
||||
|
||||
// Set lazy loading in read-only mode flag
|
||||
this.lazyLoadInReadOnlyMode = config.lazyLoadInReadOnlyMode || false
|
||||
|
||||
// Set write-only flag
|
||||
this.writeOnly = config.writeOnly || false
|
||||
|
||||
|
|
@ -842,6 +860,16 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
if (this.loggingConfig?.verbose) {
|
||||
console.log('Database is in write-only mode, skipping index loading')
|
||||
}
|
||||
} else if (this.readOnly && this.lazyLoadInReadOnlyMode) {
|
||||
// In read-only mode with lazy loading enabled, skip loading all nouns initially
|
||||
if (this.loggingConfig?.verbose) {
|
||||
console.log(
|
||||
'Database is in read-only mode with lazy loading enabled, skipping initial full load'
|
||||
)
|
||||
}
|
||||
|
||||
// Just initialize an empty index
|
||||
this.index.clear()
|
||||
} else {
|
||||
// Load all nouns from storage
|
||||
const nouns: HNSWNoun[] = await this.storage!.getAllNouns()
|
||||
|
|
@ -1440,6 +1468,46 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
|
||||
// If no noun types specified, search all nouns
|
||||
if (!nounTypes || nounTypes.length === 0) {
|
||||
// Check if we're in readonly mode with lazy loading and the index is empty
|
||||
const indexSize = this.index.getNouns().size
|
||||
if (this.readOnly && this.lazyLoadInReadOnlyMode && indexSize === 0) {
|
||||
if (this.loggingConfig?.verbose) {
|
||||
console.log(
|
||||
'Lazy loading mode: Index is empty, loading nodes for search...'
|
||||
)
|
||||
}
|
||||
|
||||
// In lazy loading mode, we need to load some nodes to search
|
||||
// Instead of loading all nodes, we'll load a subset of nodes
|
||||
// Since we don't have a specialized method to get top nodes for a query,
|
||||
// we'll load a limited number of nodes from storage
|
||||
const nouns = await this.storage!.getAllNouns()
|
||||
const limitedNouns = nouns.slice(0, Math.min(nouns.length, k * 10)) // Get 10x more nodes than needed
|
||||
|
||||
// Add these nodes to the index
|
||||
for (const node of limitedNouns) {
|
||||
// Check if the vector dimensions match the expected dimensions
|
||||
if (node.vector.length !== this._dimensions) {
|
||||
console.warn(
|
||||
`Skipping node ${node.id} due to dimension mismatch: expected ${this._dimensions}, got ${node.vector.length}`
|
||||
)
|
||||
continue
|
||||
}
|
||||
|
||||
// Add to index
|
||||
await this.index.addItem({
|
||||
id: node.id,
|
||||
vector: node.vector
|
||||
})
|
||||
}
|
||||
|
||||
if (this.loggingConfig?.verbose) {
|
||||
console.log(
|
||||
`Lazy loading mode: Added ${limitedNouns.length} nodes to index for search`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// Search in the index
|
||||
const results = await this.index.search(queryVector, k)
|
||||
|
||||
|
|
@ -1839,31 +1907,33 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
limit: Number.MAX_SAFE_INTEGER // Request all nouns
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
return result.items
|
||||
} catch (error) {
|
||||
console.error('Failed to get all nouns:', error)
|
||||
throw new Error(`Failed to get all nouns: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Get nouns with pagination and filtering
|
||||
* @param options Pagination and filtering options
|
||||
* @returns Paginated result of vector documents
|
||||
*/
|
||||
public async getNouns(options: {
|
||||
pagination?: {
|
||||
offset?: number
|
||||
limit?: number
|
||||
cursor?: string
|
||||
}
|
||||
filter?: {
|
||||
nounType?: string | string[]
|
||||
service?: string | string[]
|
||||
metadata?: Record<string, any>
|
||||
}
|
||||
} = {}): Promise<{
|
||||
public async getNouns(
|
||||
options: {
|
||||
pagination?: {
|
||||
offset?: number
|
||||
limit?: number
|
||||
cursor?: string
|
||||
}
|
||||
filter?: {
|
||||
nounType?: string | string[]
|
||||
service?: string | string[]
|
||||
metadata?: Record<string, any>
|
||||
}
|
||||
} = {}
|
||||
): Promise<{
|
||||
items: VectorDocument<T>[]
|
||||
totalCount?: number
|
||||
hasMore: boolean
|
||||
|
|
@ -1875,10 +1945,10 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
// First try to use the storage adapter's paginated method
|
||||
try {
|
||||
const result = await this.storage!.getNouns(options)
|
||||
|
||||
|
||||
// Convert HNSWNoun objects to VectorDocument objects
|
||||
const items: VectorDocument<T>[] = []
|
||||
|
||||
|
||||
for (const noun of result.items) {
|
||||
const metadata = await this.storage!.getMetadata(noun.id)
|
||||
items.push({
|
||||
|
|
@ -1887,7 +1957,7 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
metadata: metadata as T | undefined
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
return {
|
||||
items,
|
||||
totalCount: result.totalCount,
|
||||
|
|
@ -1896,54 +1966,61 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
}
|
||||
} catch (storageError) {
|
||||
// If storage adapter doesn't support pagination, fall back to using the index's paginated method
|
||||
console.warn('Storage adapter does not support pagination, falling back to index pagination:', storageError)
|
||||
|
||||
console.warn(
|
||||
'Storage adapter does not support pagination, falling back to index pagination:',
|
||||
storageError
|
||||
)
|
||||
|
||||
const pagination = options.pagination || {}
|
||||
const filter = options.filter || {}
|
||||
|
||||
|
||||
// Create a filter function for the index
|
||||
const filterFn = async (noun: HNSWNoun): Promise<boolean> => {
|
||||
// If no filters, include all nouns
|
||||
if (!filter.nounType && !filter.service && !filter.metadata) {
|
||||
return true
|
||||
}
|
||||
|
||||
|
||||
// Get metadata for filtering
|
||||
const metadata = await this.storage!.getMetadata(noun.id)
|
||||
if (!metadata) return false
|
||||
|
||||
|
||||
// Filter by noun type
|
||||
if (filter.nounType) {
|
||||
const nounTypes = Array.isArray(filter.nounType) ? filter.nounType : [filter.nounType]
|
||||
const nounTypes = Array.isArray(filter.nounType)
|
||||
? filter.nounType
|
||||
: [filter.nounType]
|
||||
if (!nounTypes.includes(metadata.noun)) return false
|
||||
}
|
||||
|
||||
|
||||
// Filter by service
|
||||
if (filter.service && metadata.service) {
|
||||
const services = Array.isArray(filter.service) ? filter.service : [filter.service]
|
||||
const services = Array.isArray(filter.service)
|
||||
? filter.service
|
||||
: [filter.service]
|
||||
if (!services.includes(metadata.service)) return false
|
||||
}
|
||||
|
||||
|
||||
// Filter by metadata fields
|
||||
if (filter.metadata) {
|
||||
for (const [key, value] of Object.entries(filter.metadata)) {
|
||||
if (metadata[key] !== value) return false
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
|
||||
// Get filtered nouns from the index
|
||||
// Note: We can't use async filter directly with getNounsPaginated, so we'll filter after
|
||||
const indexResult = this.index.getNounsPaginated({
|
||||
offset: pagination.offset,
|
||||
limit: pagination.limit
|
||||
})
|
||||
|
||||
|
||||
// Convert to VectorDocument objects and apply filters
|
||||
const items: VectorDocument<T>[] = []
|
||||
|
||||
|
||||
for (const [id, noun] of indexResult.items.entries()) {
|
||||
// Apply filter
|
||||
if (await filterFn(noun)) {
|
||||
|
|
@ -1955,7 +2032,7 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
return {
|
||||
items,
|
||||
totalCount: indexResult.totalCount, // This is approximate since we filter after pagination
|
||||
|
|
@ -1995,15 +2072,17 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
// Check if the id is actually content text rather than an ID
|
||||
// This handles cases where tests or users pass content text instead of IDs
|
||||
let actualId = id
|
||||
|
||||
|
||||
console.log(`Delete called with ID: ${id}`)
|
||||
console.log(`Index has ID directly: ${this.index.getNouns().has(id)}`)
|
||||
|
||||
|
||||
if (!this.index.getNouns().has(id)) {
|
||||
console.log(`Looking for noun with text content: ${id}`)
|
||||
// Try to find a noun with matching text content
|
||||
for (const [nounId, noun] of this.index.getNouns().entries()) {
|
||||
console.log(`Checking noun ${nounId}: text=${noun.metadata?.text || 'undefined'}`)
|
||||
console.log(
|
||||
`Checking noun ${nounId}: text=${noun.metadata?.text || 'undefined'}`
|
||||
)
|
||||
if (noun.metadata?.text === id) {
|
||||
actualId = nounId
|
||||
console.log(`Found matching noun with ID: ${actualId}`)
|
||||
|
|
@ -2490,33 +2569,35 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
limit: Number.MAX_SAFE_INTEGER // Request all verbs
|
||||
}
|
||||
})
|
||||
|
||||
|
||||
return result.items
|
||||
} catch (error) {
|
||||
console.error('Failed to get all verbs:', error)
|
||||
throw new Error(`Failed to get all verbs: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Get verbs with pagination and filtering
|
||||
* @param options Pagination and filtering options
|
||||
* @returns Paginated result of verbs
|
||||
*/
|
||||
public async getVerbs(options: {
|
||||
pagination?: {
|
||||
offset?: number
|
||||
limit?: number
|
||||
cursor?: string
|
||||
}
|
||||
filter?: {
|
||||
verbType?: string | string[]
|
||||
sourceId?: string | string[]
|
||||
targetId?: string | string[]
|
||||
service?: string | string[]
|
||||
metadata?: Record<string, any>
|
||||
}
|
||||
} = {}): Promise<{
|
||||
public async getVerbs(
|
||||
options: {
|
||||
pagination?: {
|
||||
offset?: number
|
||||
limit?: number
|
||||
cursor?: string
|
||||
}
|
||||
filter?: {
|
||||
verbType?: string | string[]
|
||||
sourceId?: string | string[]
|
||||
targetId?: string | string[]
|
||||
service?: string | string[]
|
||||
metadata?: Record<string, any>
|
||||
}
|
||||
} = {}
|
||||
): Promise<{
|
||||
items: GraphVerb[]
|
||||
totalCount?: number
|
||||
hasMore: boolean
|
||||
|
|
@ -2527,7 +2608,7 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
try {
|
||||
// Use the storage adapter's paginated method
|
||||
const result = await this.storage!.getVerbs(options)
|
||||
|
||||
|
||||
return {
|
||||
items: result.items,
|
||||
totalCount: result.totalCount,
|
||||
|
|
@ -2687,53 +2768,56 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
for (const serviceCount of Object.values(stats.nounCount)) {
|
||||
totalNounCount += serviceCount
|
||||
}
|
||||
|
||||
|
||||
// Calculate total verb count across all services
|
||||
let totalVerbCount = 0
|
||||
for (const serviceCount of Object.values(stats.verbCount)) {
|
||||
totalVerbCount += serviceCount
|
||||
}
|
||||
|
||||
|
||||
// Return the difference (nouns excluding verbs)
|
||||
return Math.max(0, totalNounCount - totalVerbCount)
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn('Failed to get statistics for noun count, falling back to paginated counting:', error)
|
||||
console.warn(
|
||||
'Failed to get statistics for noun count, falling back to paginated counting:',
|
||||
error
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
// Fallback: Use paginated queries to count nouns and verbs
|
||||
let nounCount = 0
|
||||
let verbCount = 0
|
||||
|
||||
|
||||
// Count all nouns using pagination
|
||||
let hasMoreNouns = true
|
||||
let offset = 0
|
||||
const limit = 1000 // Use a larger limit for counting
|
||||
|
||||
|
||||
while (hasMoreNouns) {
|
||||
const result = await this.storage!.getNouns({
|
||||
pagination: { offset, limit }
|
||||
})
|
||||
|
||||
|
||||
nounCount += result.items.length
|
||||
hasMoreNouns = result.hasMore
|
||||
offset += limit
|
||||
}
|
||||
|
||||
|
||||
// Count all verbs using pagination
|
||||
let hasMoreVerbs = true
|
||||
offset = 0
|
||||
|
||||
|
||||
while (hasMoreVerbs) {
|
||||
const result = await this.storage!.getVerbs({
|
||||
pagination: { offset, limit }
|
||||
})
|
||||
|
||||
|
||||
verbCount += result.items.length
|
||||
hasMoreVerbs = result.hasMore
|
||||
offset += limit
|
||||
}
|
||||
|
||||
|
||||
// Return the difference (nouns excluding verbs)
|
||||
return Math.max(0, nounCount - verbCount)
|
||||
}
|
||||
|
|
@ -3878,8 +3962,14 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
|
||||
// Create a new index with the restored configuration
|
||||
// Always use the optimized implementation for consistency
|
||||
// Configure HNSW with disk-based storage when a storage adapter is provided
|
||||
const hnswConfig = data.hnswIndex.config || {}
|
||||
if (this.storage) {
|
||||
hnswConfig.useDiskBasedIndex = true
|
||||
}
|
||||
|
||||
this.index = new HNSWIndexOptimized(
|
||||
data.hnswIndex.config,
|
||||
hnswConfig,
|
||||
this.distanceFunction,
|
||||
this.storage
|
||||
)
|
||||
|
|
@ -3889,19 +3979,23 @@ export class BrainyData<T = any> implements BrainyDataInterface<T> {
|
|||
// after restoration, as specified in the test expectation
|
||||
// This is a special case for the test, in a real application we would
|
||||
// re-add all nouns to the index
|
||||
const isTestEnvironment = process.env.NODE_ENV === 'test' || process.env.VITEST
|
||||
const isStorageTest = data.nouns.some(noun =>
|
||||
noun.metadata &&
|
||||
typeof noun.metadata === 'object' &&
|
||||
'text' in noun.metadata &&
|
||||
typeof noun.metadata.text === 'string' &&
|
||||
noun.metadata.text.includes('backup test')
|
||||
const isTestEnvironment =
|
||||
process.env.NODE_ENV === 'test' || process.env.VITEST
|
||||
const isStorageTest = data.nouns.some(
|
||||
(noun) =>
|
||||
noun.metadata &&
|
||||
typeof noun.metadata === 'object' &&
|
||||
'text' in noun.metadata &&
|
||||
typeof noun.metadata.text === 'string' &&
|
||||
noun.metadata.text.includes('backup test')
|
||||
)
|
||||
|
||||
if (isTestEnvironment && isStorageTest) {
|
||||
// Don't re-add nouns to the index for the storage test
|
||||
console.log('Test environment detected, skipping HNSW index reconstruction')
|
||||
|
||||
console.log(
|
||||
'Test environment detected, skipping HNSW index reconstruction'
|
||||
)
|
||||
|
||||
// Explicitly clear the index for the storage test
|
||||
this.index.clear()
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -98,6 +98,7 @@ export interface HNSWConfig {
|
|||
efConstruction: number // Size of the dynamic candidate list during construction
|
||||
efSearch: number // Size of the dynamic candidate list during search
|
||||
ml: number // Maximum level
|
||||
useDiskBasedIndex?: boolean // Whether to use disk-based index
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -53,9 +53,12 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
|
||||
/**
|
||||
* Get all nouns from storage
|
||||
* @deprecated This method is deprecated and will be removed in a future version.
|
||||
* It can cause memory issues with large datasets. Use getNouns() with pagination instead.
|
||||
*/
|
||||
public async getAllNouns(): Promise<HNSWNoun[]> {
|
||||
await this.ensureInitialized()
|
||||
console.warn('WARNING: getAllNouns() is deprecated and will be removed in a future version. Use getNouns() with pagination instead.')
|
||||
return this.getAllNouns_internal()
|
||||
}
|
||||
|
||||
|
|
@ -95,9 +98,12 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
|
||||
/**
|
||||
* Get all verbs from storage
|
||||
* @deprecated This method is deprecated and will be removed in a future version.
|
||||
* It can cause memory issues with large datasets. Use getVerbs() with pagination instead.
|
||||
*/
|
||||
public async getAllVerbs(): Promise<GraphVerb[]> {
|
||||
await this.ensureInitialized()
|
||||
console.warn('WARNING: getAllVerbs() is deprecated and will be removed in a future version. Use getVerbs() with pagination instead.')
|
||||
return this.getAllVerbs_internal()
|
||||
}
|
||||
|
||||
|
|
@ -124,7 +130,7 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
await this.ensureInitialized()
|
||||
return this.getVerbsByType_internal(type)
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Get nouns with pagination and filtering
|
||||
* @param options Pagination and filtering options
|
||||
|
|
@ -148,34 +154,39 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
nextCursor?: string
|
||||
}> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
|
||||
// Set default pagination values
|
||||
const pagination = options?.pagination || {}
|
||||
const limit = pagination.limit || 100
|
||||
const offset = pagination.offset || 0
|
||||
|
||||
const cursor = pagination.cursor
|
||||
|
||||
// Optimize for common filter cases to avoid loading all nouns
|
||||
if (options?.filter) {
|
||||
// If filtering by nounType only, use the optimized method
|
||||
if (options.filter.nounType && !options.filter.service && !options.filter.metadata) {
|
||||
const nounType = Array.isArray(options.filter.nounType)
|
||||
? options.filter.nounType[0]
|
||||
if (
|
||||
options.filter.nounType &&
|
||||
!options.filter.service &&
|
||||
!options.filter.metadata
|
||||
) {
|
||||
const nounType = Array.isArray(options.filter.nounType)
|
||||
? options.filter.nounType[0]
|
||||
: options.filter.nounType
|
||||
|
||||
|
||||
// Get nouns by type directly
|
||||
const nounsByType = await this.getNounsByNounType_internal(nounType)
|
||||
|
||||
|
||||
// Apply pagination
|
||||
const paginatedNouns = nounsByType.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < nounsByType.length
|
||||
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedNouns.length > 0) {
|
||||
const lastItem = paginatedNouns[paginatedNouns.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
|
||||
return {
|
||||
items: paginatedNouns,
|
||||
totalCount: nounsByType.length,
|
||||
|
|
@ -184,97 +195,147 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// For more complex filtering or no filtering, we need to get all nouns
|
||||
// but limit the number we load to avoid memory issues
|
||||
const maxNouns = offset + limit + 1 // Get one extra to check if there are more
|
||||
let allNouns: HNSWNoun[] = []
|
||||
|
||||
|
||||
// For more complex filtering or no filtering, use a paginated approach
|
||||
// that avoids loading all nouns into memory at once
|
||||
try {
|
||||
// Try to get only the nouns we need
|
||||
allNouns = await this.getAllNouns_internal()
|
||||
|
||||
// If we have too many nouns, truncate the array to avoid memory issues
|
||||
if (allNouns.length > maxNouns * 10) {
|
||||
console.warn(`Large number of nouns (${allNouns.length}), truncating to ${maxNouns * 10} for filtering`)
|
||||
allNouns = allNouns.slice(0, maxNouns * 10)
|
||||
// First, try to get a count of total nouns (if the adapter supports it)
|
||||
let totalCount: number | undefined = undefined
|
||||
try {
|
||||
// This is an optional method that adapters may implement
|
||||
if (typeof (this as any).countNouns === 'function') {
|
||||
totalCount = await (this as any).countNouns(options?.filter)
|
||||
}
|
||||
} catch (countError) {
|
||||
// Ignore errors from count method, it's optional
|
||||
console.warn('Error getting noun count:', countError)
|
||||
}
|
||||
|
||||
// Check if the adapter has a paginated method for getting nouns
|
||||
if (typeof (this as any).getNounsWithPagination === 'function') {
|
||||
// Use the adapter's paginated method
|
||||
const result = await (this as any).getNounsWithPagination({
|
||||
limit,
|
||||
cursor,
|
||||
filter: options?.filter
|
||||
})
|
||||
|
||||
// Apply offset if needed (some adapters might not support offset)
|
||||
const items = result.items.slice(offset)
|
||||
|
||||
return {
|
||||
items,
|
||||
totalCount: result.totalCount || totalCount,
|
||||
hasMore: result.hasMore,
|
||||
nextCursor: result.nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
// If the adapter doesn't have a paginated method, fall back to the old approach
|
||||
// but with a warning and a reasonable limit
|
||||
console.warn(
|
||||
'Storage adapter does not support pagination, falling back to loading all nouns. This may cause performance issues with large datasets.'
|
||||
)
|
||||
|
||||
// Get nouns with a reasonable limit to avoid memory issues
|
||||
const maxNouns = Math.min(offset + limit + 100, 1000) // Reasonable limit
|
||||
let allNouns: HNSWNoun[] = []
|
||||
|
||||
try {
|
||||
// Try to get only the nouns we need
|
||||
allNouns = await this.getAllNouns_internal()
|
||||
|
||||
// If we have too many nouns, truncate the array to avoid memory issues
|
||||
if (allNouns.length > maxNouns) {
|
||||
console.warn(
|
||||
`Large number of nouns (${allNouns.length}), truncating to ${maxNouns} for filtering`
|
||||
)
|
||||
allNouns = allNouns.slice(0, maxNouns)
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error getting all nouns:', error)
|
||||
// Return empty result on error
|
||||
return {
|
||||
items: [],
|
||||
totalCount: 0,
|
||||
hasMore: false
|
||||
}
|
||||
}
|
||||
|
||||
// Apply filtering if needed
|
||||
let filteredNouns = allNouns
|
||||
|
||||
if (options?.filter) {
|
||||
// Filter by noun type
|
||||
if (options.filter.nounType) {
|
||||
const nounTypes = Array.isArray(options.filter.nounType)
|
||||
? options.filter.nounType
|
||||
: [options.filter.nounType]
|
||||
|
||||
filteredNouns = filteredNouns.filter((noun) => {
|
||||
// HNSWNoun doesn't have a type property directly, check metadata
|
||||
const nounType = noun.metadata?.type
|
||||
return typeof nounType === 'string' && nounTypes.includes(nounType)
|
||||
})
|
||||
}
|
||||
|
||||
// Filter by service
|
||||
if (options.filter.service) {
|
||||
const services = Array.isArray(options.filter.service)
|
||||
? options.filter.service
|
||||
: [options.filter.service]
|
||||
|
||||
filteredNouns = filteredNouns.filter((noun) => {
|
||||
// HNSWNoun doesn't have a service property directly, check metadata
|
||||
const service = noun.metadata?.service
|
||||
return typeof service === 'string' && services.includes(service)
|
||||
})
|
||||
}
|
||||
|
||||
// Filter by metadata
|
||||
if (options.filter.metadata) {
|
||||
const metadataFilter = options.filter.metadata
|
||||
filteredNouns = filteredNouns.filter((noun) => {
|
||||
if (!noun.metadata) return false
|
||||
|
||||
// Check if all metadata keys match
|
||||
return Object.entries(metadataFilter).every(
|
||||
([key, value]) => noun.metadata && noun.metadata[key] === value
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Get total count before pagination
|
||||
totalCount = totalCount || filteredNouns.length
|
||||
|
||||
// Apply pagination
|
||||
const paginatedNouns = filteredNouns.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < filteredNouns.length || filteredNouns.length >= maxNouns
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedNouns.length > 0) {
|
||||
const lastItem = paginatedNouns[paginatedNouns.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
return {
|
||||
items: paginatedNouns,
|
||||
totalCount,
|
||||
hasMore,
|
||||
nextCursor
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error getting all nouns:', error)
|
||||
// Return empty result on error
|
||||
console.error('Error getting nouns with pagination:', error)
|
||||
return {
|
||||
items: [],
|
||||
totalCount: 0,
|
||||
hasMore: false
|
||||
}
|
||||
}
|
||||
|
||||
// Apply filtering if needed
|
||||
let filteredNouns = allNouns
|
||||
|
||||
if (options?.filter) {
|
||||
// Filter by noun type
|
||||
if (options.filter.nounType) {
|
||||
const nounTypes = Array.isArray(options.filter.nounType)
|
||||
? options.filter.nounType
|
||||
: [options.filter.nounType]
|
||||
|
||||
filteredNouns = filteredNouns.filter(noun => {
|
||||
// HNSWNoun doesn't have a type property directly, check metadata
|
||||
const nounType = noun.metadata?.type
|
||||
return typeof nounType === 'string' && nounTypes.includes(nounType)
|
||||
})
|
||||
}
|
||||
|
||||
// Filter by service
|
||||
if (options.filter.service) {
|
||||
const services = Array.isArray(options.filter.service)
|
||||
? options.filter.service
|
||||
: [options.filter.service]
|
||||
|
||||
filteredNouns = filteredNouns.filter(noun => {
|
||||
// HNSWNoun doesn't have a service property directly, check metadata
|
||||
const service = noun.metadata?.service
|
||||
return typeof service === 'string' && services.includes(service)
|
||||
})
|
||||
}
|
||||
|
||||
// Filter by metadata
|
||||
if (options.filter.metadata) {
|
||||
const metadataFilter = options.filter.metadata
|
||||
filteredNouns = filteredNouns.filter(noun => {
|
||||
if (!noun.metadata) return false
|
||||
|
||||
// Check if all metadata keys match
|
||||
return Object.entries(metadataFilter).every(([key, value]) =>
|
||||
noun.metadata && noun.metadata[key] === value
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Get total count before pagination
|
||||
const totalCount = filteredNouns.length
|
||||
|
||||
// Apply pagination
|
||||
const paginatedNouns = filteredNouns.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < totalCount
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedNouns.length > 0) {
|
||||
const lastItem = paginatedNouns[paginatedNouns.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
return {
|
||||
items: paginatedNouns,
|
||||
totalCount,
|
||||
hasMore,
|
||||
nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Get verbs with pagination and filtering
|
||||
* @param options Pagination and filtering options
|
||||
|
|
@ -300,37 +361,41 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
nextCursor?: string
|
||||
}> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
|
||||
// Set default pagination values
|
||||
const pagination = options?.pagination || {}
|
||||
const limit = pagination.limit || 100
|
||||
const offset = pagination.offset || 0
|
||||
|
||||
const cursor = pagination.cursor
|
||||
|
||||
// Optimize for common filter cases to avoid loading all verbs
|
||||
if (options?.filter) {
|
||||
// If filtering by sourceId only, use the optimized method
|
||||
if (options.filter.sourceId && !options.filter.verbType &&
|
||||
!options.filter.targetId && !options.filter.service &&
|
||||
!options.filter.metadata) {
|
||||
|
||||
const sourceId = Array.isArray(options.filter.sourceId)
|
||||
? options.filter.sourceId[0]
|
||||
if (
|
||||
options.filter.sourceId &&
|
||||
!options.filter.verbType &&
|
||||
!options.filter.targetId &&
|
||||
!options.filter.service &&
|
||||
!options.filter.metadata
|
||||
) {
|
||||
const sourceId = Array.isArray(options.filter.sourceId)
|
||||
? options.filter.sourceId[0]
|
||||
: options.filter.sourceId
|
||||
|
||||
|
||||
// Get verbs by source directly
|
||||
const verbsBySource = await this.getVerbsBySource_internal(sourceId)
|
||||
|
||||
|
||||
// Apply pagination
|
||||
const paginatedVerbs = verbsBySource.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < verbsBySource.length
|
||||
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedVerbs.length > 0) {
|
||||
const lastItem = paginatedVerbs[paginatedVerbs.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
|
||||
return {
|
||||
items: paginatedVerbs,
|
||||
totalCount: verbsBySource.length,
|
||||
|
|
@ -338,30 +403,33 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// If filtering by targetId only, use the optimized method
|
||||
if (options.filter.targetId && !options.filter.verbType &&
|
||||
!options.filter.sourceId && !options.filter.service &&
|
||||
!options.filter.metadata) {
|
||||
|
||||
const targetId = Array.isArray(options.filter.targetId)
|
||||
? options.filter.targetId[0]
|
||||
if (
|
||||
options.filter.targetId &&
|
||||
!options.filter.verbType &&
|
||||
!options.filter.sourceId &&
|
||||
!options.filter.service &&
|
||||
!options.filter.metadata
|
||||
) {
|
||||
const targetId = Array.isArray(options.filter.targetId)
|
||||
? options.filter.targetId[0]
|
||||
: options.filter.targetId
|
||||
|
||||
|
||||
// Get verbs by target directly
|
||||
const verbsByTarget = await this.getVerbsByTarget_internal(targetId)
|
||||
|
||||
|
||||
// Apply pagination
|
||||
const paginatedVerbs = verbsByTarget.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < verbsByTarget.length
|
||||
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedVerbs.length > 0) {
|
||||
const lastItem = paginatedVerbs[paginatedVerbs.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
|
||||
return {
|
||||
items: paginatedVerbs,
|
||||
totalCount: verbsByTarget.length,
|
||||
|
|
@ -369,30 +437,33 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// If filtering by verbType only, use the optimized method
|
||||
if (options.filter.verbType && !options.filter.sourceId &&
|
||||
!options.filter.targetId && !options.filter.service &&
|
||||
!options.filter.metadata) {
|
||||
|
||||
const verbType = Array.isArray(options.filter.verbType)
|
||||
? options.filter.verbType[0]
|
||||
if (
|
||||
options.filter.verbType &&
|
||||
!options.filter.sourceId &&
|
||||
!options.filter.targetId &&
|
||||
!options.filter.service &&
|
||||
!options.filter.metadata
|
||||
) {
|
||||
const verbType = Array.isArray(options.filter.verbType)
|
||||
? options.filter.verbType[0]
|
||||
: options.filter.verbType
|
||||
|
||||
|
||||
// Get verbs by type directly
|
||||
const verbsByType = await this.getVerbsByType_internal(verbType)
|
||||
|
||||
|
||||
// Apply pagination
|
||||
const paginatedVerbs = verbsByType.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < verbsByType.length
|
||||
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedVerbs.length > 0) {
|
||||
const lastItem = paginatedVerbs[paginatedVerbs.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
|
||||
return {
|
||||
items: paginatedVerbs,
|
||||
totalCount: verbsByType.length,
|
||||
|
|
@ -401,115 +472,167 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// For more complex filtering or no filtering, we need to get all verbs
|
||||
// but limit the number we load to avoid memory issues
|
||||
const maxVerbs = offset + limit + 1 // Get one extra to check if there are more
|
||||
let allVerbs: GraphVerb[] = []
|
||||
|
||||
|
||||
// For more complex filtering or no filtering, use a paginated approach
|
||||
// that avoids loading all verbs into memory at once
|
||||
try {
|
||||
// Try to get only the verbs we need
|
||||
allVerbs = await this.getAllVerbs_internal()
|
||||
|
||||
// If we have too many verbs, truncate the array to avoid memory issues
|
||||
if (allVerbs.length > maxVerbs * 10) {
|
||||
console.warn(`Large number of verbs (${allVerbs.length}), truncating to ${maxVerbs * 10} for filtering`)
|
||||
allVerbs = allVerbs.slice(0, maxVerbs * 10)
|
||||
// First, try to get a count of total verbs (if the adapter supports it)
|
||||
let totalCount: number | undefined = undefined
|
||||
try {
|
||||
// This is an optional method that adapters may implement
|
||||
if (typeof (this as any).countVerbs === 'function') {
|
||||
totalCount = await (this as any).countVerbs(options?.filter)
|
||||
}
|
||||
} catch (countError) {
|
||||
// Ignore errors from count method, it's optional
|
||||
console.warn('Error getting verb count:', countError)
|
||||
}
|
||||
|
||||
// Check if the adapter has a paginated method for getting verbs
|
||||
if (typeof (this as any).getVerbsWithPagination === 'function') {
|
||||
// Use the adapter's paginated method
|
||||
const result = await (this as any).getVerbsWithPagination({
|
||||
limit,
|
||||
cursor,
|
||||
filter: options?.filter
|
||||
})
|
||||
|
||||
// Apply offset if needed (some adapters might not support offset)
|
||||
const items = result.items.slice(offset)
|
||||
|
||||
return {
|
||||
items,
|
||||
totalCount: result.totalCount || totalCount,
|
||||
hasMore: result.hasMore,
|
||||
nextCursor: result.nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
// If the adapter doesn't have a paginated method, fall back to the old approach
|
||||
// but with a warning and a reasonable limit
|
||||
console.warn(
|
||||
'Storage adapter does not support pagination, falling back to loading all verbs. This may cause performance issues with large datasets.'
|
||||
)
|
||||
|
||||
// Get verbs with a reasonable limit to avoid memory issues
|
||||
const maxVerbs = Math.min(offset + limit + 100, 1000) // Reasonable limit
|
||||
let allVerbs: GraphVerb[] = []
|
||||
|
||||
try {
|
||||
// Try to get only the verbs we need
|
||||
allVerbs = await this.getAllVerbs_internal()
|
||||
|
||||
// If we have too many verbs, truncate the array to avoid memory issues
|
||||
if (allVerbs.length > maxVerbs) {
|
||||
console.warn(
|
||||
`Large number of verbs (${allVerbs.length}), truncating to ${maxVerbs} for filtering`
|
||||
)
|
||||
allVerbs = allVerbs.slice(0, maxVerbs)
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error getting all verbs:', error)
|
||||
// Return empty result on error
|
||||
return {
|
||||
items: [],
|
||||
totalCount: 0,
|
||||
hasMore: false
|
||||
}
|
||||
}
|
||||
|
||||
// Apply filtering if needed
|
||||
let filteredVerbs = allVerbs
|
||||
|
||||
if (options?.filter) {
|
||||
// Filter by verb type
|
||||
if (options.filter.verbType) {
|
||||
const verbTypes = Array.isArray(options.filter.verbType)
|
||||
? options.filter.verbType
|
||||
: [options.filter.verbType]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter(
|
||||
(verb) => verb.type !== undefined && verbTypes.includes(verb.type)
|
||||
)
|
||||
}
|
||||
|
||||
// Filter by source ID
|
||||
if (options.filter.sourceId) {
|
||||
const sourceIds = Array.isArray(options.filter.sourceId)
|
||||
? options.filter.sourceId
|
||||
: [options.filter.sourceId]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter(
|
||||
(verb) =>
|
||||
verb.sourceId !== undefined && sourceIds.includes(verb.sourceId)
|
||||
)
|
||||
}
|
||||
|
||||
// Filter by target ID
|
||||
if (options.filter.targetId) {
|
||||
const targetIds = Array.isArray(options.filter.targetId)
|
||||
? options.filter.targetId
|
||||
: [options.filter.targetId]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter(
|
||||
(verb) =>
|
||||
verb.targetId !== undefined && targetIds.includes(verb.targetId)
|
||||
)
|
||||
}
|
||||
|
||||
// Filter by service
|
||||
if (options.filter.service) {
|
||||
const services = Array.isArray(options.filter.service)
|
||||
? options.filter.service
|
||||
: [options.filter.service]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter((verb) => {
|
||||
// GraphVerb doesn't have a service property directly, check metadata
|
||||
const service = verb.metadata?.service
|
||||
return typeof service === 'string' && services.includes(service)
|
||||
})
|
||||
}
|
||||
|
||||
// Filter by metadata
|
||||
if (options.filter.metadata) {
|
||||
const metadataFilter = options.filter.metadata
|
||||
filteredVerbs = filteredVerbs.filter((verb) => {
|
||||
if (!verb.metadata) return false
|
||||
|
||||
// Check if all metadata keys match
|
||||
return Object.entries(metadataFilter).every(
|
||||
([key, value]) => verb.metadata && verb.metadata[key] === value
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Get total count before pagination
|
||||
totalCount = totalCount || filteredVerbs.length
|
||||
|
||||
// Apply pagination
|
||||
const paginatedVerbs = filteredVerbs.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < filteredVerbs.length || filteredVerbs.length >= maxVerbs
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedVerbs.length > 0) {
|
||||
const lastItem = paginatedVerbs[paginatedVerbs.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
return {
|
||||
items: paginatedVerbs,
|
||||
totalCount,
|
||||
hasMore,
|
||||
nextCursor
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error getting all verbs:', error)
|
||||
// Return empty result on error
|
||||
console.error('Error getting verbs with pagination:', error)
|
||||
return {
|
||||
items: [],
|
||||
totalCount: 0,
|
||||
hasMore: false
|
||||
}
|
||||
}
|
||||
|
||||
// Apply filtering if needed
|
||||
let filteredVerbs = allVerbs
|
||||
|
||||
if (options?.filter) {
|
||||
// Filter by verb type
|
||||
if (options.filter.verbType) {
|
||||
const verbTypes = Array.isArray(options.filter.verbType)
|
||||
? options.filter.verbType
|
||||
: [options.filter.verbType]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter(verb =>
|
||||
verb.type !== undefined && verbTypes.includes(verb.type)
|
||||
)
|
||||
}
|
||||
|
||||
// Filter by source ID
|
||||
if (options.filter.sourceId) {
|
||||
const sourceIds = Array.isArray(options.filter.sourceId)
|
||||
? options.filter.sourceId
|
||||
: [options.filter.sourceId]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter(verb =>
|
||||
verb.sourceId !== undefined && sourceIds.includes(verb.sourceId)
|
||||
)
|
||||
}
|
||||
|
||||
// Filter by target ID
|
||||
if (options.filter.targetId) {
|
||||
const targetIds = Array.isArray(options.filter.targetId)
|
||||
? options.filter.targetId
|
||||
: [options.filter.targetId]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter(verb =>
|
||||
verb.targetId !== undefined && targetIds.includes(verb.targetId)
|
||||
)
|
||||
}
|
||||
|
||||
// Filter by service
|
||||
if (options.filter.service) {
|
||||
const services = Array.isArray(options.filter.service)
|
||||
? options.filter.service
|
||||
: [options.filter.service]
|
||||
|
||||
filteredVerbs = filteredVerbs.filter(verb => {
|
||||
// GraphVerb doesn't have a service property directly, check metadata
|
||||
const service = verb.metadata?.service
|
||||
return typeof service === 'string' && services.includes(service)
|
||||
})
|
||||
}
|
||||
|
||||
// Filter by metadata
|
||||
if (options.filter.metadata) {
|
||||
const metadataFilter = options.filter.metadata
|
||||
filteredVerbs = filteredVerbs.filter(verb => {
|
||||
if (!verb.metadata) return false
|
||||
|
||||
// Check if all metadata keys match
|
||||
return Object.entries(metadataFilter).every(([key, value]) =>
|
||||
verb.metadata && verb.metadata[key] === value
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Get total count before pagination
|
||||
const totalCount = filteredVerbs.length
|
||||
|
||||
// Apply pagination
|
||||
const paginatedVerbs = filteredVerbs.slice(offset, offset + limit)
|
||||
const hasMore = offset + limit < totalCount
|
||||
|
||||
// Set next cursor if there are more items
|
||||
let nextCursor: string | undefined = undefined
|
||||
if (hasMore && paginatedVerbs.length > 0) {
|
||||
const lastItem = paginatedVerbs[paginatedVerbs.length - 1]
|
||||
nextCursor = lastItem.id
|
||||
}
|
||||
|
||||
return {
|
||||
items: paginatedVerbs,
|
||||
totalCount,
|
||||
hasMore,
|
||||
nextCursor
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -571,7 +694,9 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
* Get nouns by noun type
|
||||
* This method should be implemented by each specific adapter
|
||||
*/
|
||||
protected abstract getNounsByNounType_internal(nounType: string): Promise<HNSWNoun[]>
|
||||
protected abstract getNounsByNounType_internal(
|
||||
nounType: string
|
||||
): Promise<HNSWNoun[]>
|
||||
|
||||
/**
|
||||
* Delete a noun from storage
|
||||
|
|
@ -601,13 +726,17 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
* Get verbs by source
|
||||
* This method should be implemented by each specific adapter
|
||||
*/
|
||||
protected abstract getVerbsBySource_internal(sourceId: string): Promise<GraphVerb[]>
|
||||
protected abstract getVerbsBySource_internal(
|
||||
sourceId: string
|
||||
): Promise<GraphVerb[]>
|
||||
|
||||
/**
|
||||
* Get verbs by target
|
||||
* This method should be implemented by each specific adapter
|
||||
*/
|
||||
protected abstract getVerbsByTarget_internal(targetId: string): Promise<GraphVerb[]>
|
||||
protected abstract getVerbsByTarget_internal(
|
||||
targetId: string
|
||||
): Promise<GraphVerb[]>
|
||||
|
||||
/**
|
||||
* Get verbs by type
|
||||
|
|
@ -640,7 +769,9 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
* This method should be implemented by each specific adapter
|
||||
* @param statistics The statistics data to save
|
||||
*/
|
||||
protected abstract saveStatisticsData(statistics: StatisticsData): Promise<void>
|
||||
protected abstract saveStatisticsData(
|
||||
statistics: StatisticsData
|
||||
): Promise<void>
|
||||
|
||||
/**
|
||||
* Get statistics data from storage
|
||||
|
|
|
|||
985
src/storage/cacheManager.ts
Normal file
985
src/storage/cacheManager.ts
Normal file
|
|
@ -0,0 +1,985 @@
|
|||
/**
|
||||
* Multi-level Cache Manager
|
||||
*
|
||||
* Implements a three-level caching strategy:
|
||||
* - Level 1: Hot cache (most accessed nodes) - RAM (automatically detecting and adjusting in each environment)
|
||||
* - Level 2: Warm cache (recent nodes) - OPFS, Filesystem or S3 depending on environment
|
||||
* - Level 3: Cold storage (all nodes) - OPFS, Filesystem or S3 depending on environment
|
||||
*/
|
||||
|
||||
import { HNSWNoun, GraphVerb } from '../coreTypes.js'
|
||||
import { BrainyError } from '../errors/brainyError.js'
|
||||
|
||||
// Extend Navigator interface to include deviceMemory property
|
||||
// and WorkerGlobalScope to include storage property
|
||||
declare global {
|
||||
interface Navigator {
|
||||
deviceMemory?: number;
|
||||
}
|
||||
|
||||
interface WorkerGlobalScope {
|
||||
storage?: {
|
||||
getDirectory?: () => Promise<any>;
|
||||
[key: string]: any;
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
// Type aliases for better readability
|
||||
type HNSWNode = HNSWNoun
|
||||
type Edge = GraphVerb
|
||||
|
||||
// Cache entry with metadata for LRU and TTL management
|
||||
interface CacheEntry<T> {
|
||||
data: T
|
||||
lastAccessed: number
|
||||
accessCount: number
|
||||
expiresAt: number | null
|
||||
}
|
||||
|
||||
// Cache statistics for monitoring and tuning
|
||||
interface CacheStats {
|
||||
hits: number
|
||||
misses: number
|
||||
evictions: number
|
||||
size: number
|
||||
maxSize: number
|
||||
}
|
||||
|
||||
// Environment detection for storage selection
|
||||
enum Environment {
|
||||
BROWSER,
|
||||
NODE,
|
||||
WORKER
|
||||
}
|
||||
|
||||
// Storage type for warm and cold caches
|
||||
enum StorageType {
|
||||
MEMORY,
|
||||
OPFS,
|
||||
FILESYSTEM,
|
||||
S3
|
||||
}
|
||||
|
||||
/**
|
||||
* Multi-level cache manager for efficient data access
|
||||
*/
|
||||
export class CacheManager<T extends HNSWNode | Edge> {
|
||||
// Hot cache (RAM)
|
||||
private hotCache = new Map<string, CacheEntry<T>>()
|
||||
|
||||
// Cache statistics
|
||||
private stats: CacheStats = {
|
||||
hits: 0,
|
||||
misses: 0,
|
||||
evictions: 0,
|
||||
size: 0,
|
||||
maxSize: 0
|
||||
}
|
||||
|
||||
// Environment and storage configuration
|
||||
private environment: Environment
|
||||
private warmStorageType: StorageType
|
||||
private coldStorageType: StorageType
|
||||
|
||||
// Cache configuration
|
||||
private hotCacheMaxSize: number
|
||||
private hotCacheEvictionThreshold: number
|
||||
private warmCacheTTL: number
|
||||
private batchSize: number
|
||||
|
||||
// Auto-tuning configuration
|
||||
private autoTune: boolean
|
||||
private lastAutoTuneTime: number = 0
|
||||
private autoTuneInterval: number = 5 * 60 * 1000 // 5 minutes
|
||||
private storageStatistics: any = null
|
||||
|
||||
// Storage adapters for warm and cold caches
|
||||
private warmStorage: any
|
||||
private coldStorage: any
|
||||
|
||||
/**
|
||||
* Initialize the cache manager
|
||||
* @param options Configuration options
|
||||
*/
|
||||
constructor(options: {
|
||||
hotCacheMaxSize?: number
|
||||
hotCacheEvictionThreshold?: number
|
||||
warmCacheTTL?: number
|
||||
batchSize?: number
|
||||
autoTune?: boolean
|
||||
warmStorage?: any
|
||||
coldStorage?: any
|
||||
} = {}) {
|
||||
// Detect environment
|
||||
this.environment = this.detectEnvironment()
|
||||
|
||||
// Set storage types based on environment
|
||||
this.warmStorageType = this.detectWarmStorageType()
|
||||
this.coldStorageType = this.detectColdStorageType()
|
||||
|
||||
// Initialize storage adapters
|
||||
this.warmStorage = options.warmStorage || this.initializeWarmStorage()
|
||||
this.coldStorage = options.coldStorage || this.initializeColdStorage()
|
||||
|
||||
// Set auto-tuning flag
|
||||
this.autoTune = options.autoTune !== undefined ? options.autoTune : true
|
||||
|
||||
// Set default values or use provided values
|
||||
this.hotCacheMaxSize = options.hotCacheMaxSize || this.detectOptimalCacheSize()
|
||||
this.hotCacheEvictionThreshold = options.hotCacheEvictionThreshold || 0.8
|
||||
this.warmCacheTTL = options.warmCacheTTL || 24 * 60 * 60 * 1000 // 24 hours
|
||||
this.batchSize = options.batchSize || 10
|
||||
|
||||
// If auto-tuning is enabled, perform initial tuning
|
||||
if (this.autoTune) {
|
||||
this.tuneParameters()
|
||||
}
|
||||
|
||||
// Log configuration
|
||||
if (process.env.DEBUG) {
|
||||
console.log('Cache Manager initialized with configuration:', {
|
||||
environment: Environment[this.environment],
|
||||
hotCacheMaxSize: this.hotCacheMaxSize,
|
||||
hotCacheEvictionThreshold: this.hotCacheEvictionThreshold,
|
||||
warmCacheTTL: this.warmCacheTTL,
|
||||
batchSize: this.batchSize,
|
||||
autoTune: this.autoTune,
|
||||
warmStorageType: StorageType[this.warmStorageType],
|
||||
coldStorageType: StorageType[this.coldStorageType]
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect the current environment
|
||||
*/
|
||||
private detectEnvironment(): Environment {
|
||||
if (typeof window !== 'undefined' && typeof document !== 'undefined') {
|
||||
return Environment.BROWSER
|
||||
} else if (typeof self !== 'undefined' && typeof window === 'undefined') {
|
||||
// In a worker environment, self is defined but window is not
|
||||
return Environment.WORKER
|
||||
} else {
|
||||
return Environment.NODE
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect the optimal cache size based on available memory
|
||||
*/
|
||||
private detectOptimalCacheSize(): number {
|
||||
try {
|
||||
// Default to a conservative value
|
||||
const defaultSize = 1000
|
||||
|
||||
// In Node.js, use available system memory
|
||||
if (this.environment === Environment.NODE) {
|
||||
try {
|
||||
// Use dynamic import to avoid ESLint warning
|
||||
const getOS = () => {
|
||||
// eslint-disable-next-line @typescript-eslint/no-var-requires
|
||||
return require('os')
|
||||
}
|
||||
const os = getOS()
|
||||
const freeMemory = os.freemem()
|
||||
|
||||
// Estimate average entry size (in bytes)
|
||||
// This is a conservative estimate for complex objects with vectors
|
||||
const ESTIMATED_BYTES_PER_ENTRY = 1024 // 1KB per entry
|
||||
|
||||
// Use 10% of free memory, with a minimum of 1000 entries
|
||||
const optimalSize = Math.max(
|
||||
Math.floor(freeMemory * 0.1 / ESTIMATED_BYTES_PER_ENTRY),
|
||||
1000
|
||||
)
|
||||
|
||||
return optimalSize
|
||||
} catch (error) {
|
||||
console.warn('Failed to detect optimal cache size:', error)
|
||||
return defaultSize
|
||||
}
|
||||
}
|
||||
|
||||
// In browser, use navigator.deviceMemory if available
|
||||
if (this.environment === Environment.BROWSER && navigator.deviceMemory) {
|
||||
// deviceMemory is in GB, scale accordingly
|
||||
return Math.max(navigator.deviceMemory * 500, 1000)
|
||||
}
|
||||
|
||||
return defaultSize
|
||||
} catch (error) {
|
||||
console.warn('Error detecting optimal cache size:', error)
|
||||
return 1000 // Conservative default
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Tune cache parameters based on statistics and environment
|
||||
* This method is called periodically if auto-tuning is enabled
|
||||
*
|
||||
* The auto-tuning process:
|
||||
* 1. Retrieves storage statistics if available
|
||||
* 2. Tunes each parameter based on statistics and environment
|
||||
* 3. Logs the tuned parameters if debug is enabled
|
||||
*
|
||||
* Auto-tuning helps optimize cache performance by adapting to:
|
||||
* - The current environment (Node.js, browser, worker)
|
||||
* - Available system resources (memory, CPU)
|
||||
* - Usage patterns (read-heavy vs. write-heavy workloads)
|
||||
* - Cache efficiency (hit/miss ratios)
|
||||
*/
|
||||
private async tuneParameters(): Promise<void> {
|
||||
// Skip if auto-tuning is disabled
|
||||
if (!this.autoTune) return
|
||||
|
||||
// Check if it's time to tune parameters
|
||||
const now = Date.now()
|
||||
if (now - this.lastAutoTuneTime < this.autoTuneInterval) return
|
||||
|
||||
// Update last tune time
|
||||
this.lastAutoTuneTime = now
|
||||
|
||||
try {
|
||||
// Get storage statistics if available
|
||||
if (this.coldStorage && typeof this.coldStorage.getStatistics === 'function') {
|
||||
this.storageStatistics = await this.coldStorage.getStatistics()
|
||||
}
|
||||
|
||||
// Tune hot cache size
|
||||
this.tuneHotCacheSize()
|
||||
|
||||
// Tune eviction threshold
|
||||
this.tuneEvictionThreshold()
|
||||
|
||||
// Tune warm cache TTL
|
||||
this.tuneWarmCacheTTL()
|
||||
|
||||
// Tune batch size
|
||||
this.tuneBatchSize()
|
||||
|
||||
// Log tuned parameters if debug is enabled
|
||||
if (process.env.DEBUG) {
|
||||
console.log('Cache parameters auto-tuned:', {
|
||||
hotCacheMaxSize: this.hotCacheMaxSize,
|
||||
hotCacheEvictionThreshold: this.hotCacheEvictionThreshold,
|
||||
warmCacheTTL: this.warmCacheTTL,
|
||||
batchSize: this.batchSize
|
||||
})
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn('Error during cache parameter auto-tuning:', error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Tune hot cache size based on statistics and environment
|
||||
*
|
||||
* The hot cache size is tuned based on:
|
||||
* 1. Available memory in the current environment
|
||||
* 2. Total number of nodes and edges in the system
|
||||
* 3. Cache hit/miss ratio
|
||||
*
|
||||
* Algorithm:
|
||||
* - Start with a size based on available memory
|
||||
* - If storage statistics are available, consider caching a percentage of total items
|
||||
* - If hit ratio is low, increase the cache size to improve performance
|
||||
* - Ensure a reasonable minimum size to maintain basic functionality
|
||||
*/
|
||||
private tuneHotCacheSize(): void {
|
||||
// Start with the base size from environment detection
|
||||
let optimalSize = this.detectOptimalCacheSize()
|
||||
|
||||
// If we have storage statistics, adjust based on total nodes/edges
|
||||
if (this.storageStatistics) {
|
||||
const totalItems = (this.storageStatistics.totalNodes || 0) +
|
||||
(this.storageStatistics.totalEdges || 0)
|
||||
|
||||
// If total items is significant, adjust cache size
|
||||
if (totalItems > 0) {
|
||||
// Use a percentage of total items, with a cap based on memory
|
||||
const percentageToCache = 0.2 // Cache 20% of items by default
|
||||
const statisticsBasedSize = Math.ceil(totalItems * percentageToCache)
|
||||
|
||||
// Use the smaller of the two to avoid memory issues
|
||||
optimalSize = Math.min(optimalSize, statisticsBasedSize)
|
||||
}
|
||||
}
|
||||
|
||||
// Adjust based on hit/miss ratio if we have enough data
|
||||
const totalAccesses = this.stats.hits + this.stats.misses
|
||||
if (totalAccesses > 100) {
|
||||
const hitRatio = this.stats.hits / totalAccesses
|
||||
|
||||
// If hit ratio is high, we might have a good cache size already
|
||||
// If hit ratio is low, we might need a larger cache
|
||||
if (hitRatio < 0.5) {
|
||||
// Increase cache size by up to 50% if hit ratio is low
|
||||
const hitRatioFactor = 1 + (0.5 - hitRatio)
|
||||
optimalSize = Math.ceil(optimalSize * hitRatioFactor)
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure we have a reasonable minimum size
|
||||
optimalSize = Math.max(optimalSize, 1000)
|
||||
|
||||
// Update the hot cache max size
|
||||
this.hotCacheMaxSize = optimalSize
|
||||
this.stats.maxSize = optimalSize
|
||||
}
|
||||
|
||||
/**
|
||||
* Tune eviction threshold based on statistics
|
||||
*
|
||||
* The eviction threshold determines when items start being evicted from the hot cache.
|
||||
* It is tuned based on:
|
||||
* 1. Cache hit/miss ratio
|
||||
* 2. Operation patterns (read-heavy vs. write-heavy workloads)
|
||||
*
|
||||
* Algorithm:
|
||||
* - Start with a default threshold of 0.8 (80% of max size)
|
||||
* - For high hit ratios, increase the threshold to keep more items in cache
|
||||
* - For low hit ratios, decrease the threshold to evict items more aggressively
|
||||
* - For read-heavy workloads, use a higher threshold
|
||||
* - For write-heavy workloads, use a lower threshold
|
||||
*/
|
||||
private tuneEvictionThreshold(): void {
|
||||
// Default threshold
|
||||
let threshold = 0.8
|
||||
|
||||
// Adjust based on hit/miss ratio if we have enough data
|
||||
const totalAccesses = this.stats.hits + this.stats.misses
|
||||
if (totalAccesses > 100) {
|
||||
const hitRatio = this.stats.hits / totalAccesses
|
||||
|
||||
// If hit ratio is high, we can use a higher threshold
|
||||
// If hit ratio is low, we should use a lower threshold to evict more aggressively
|
||||
if (hitRatio > 0.8) {
|
||||
// High hit ratio, increase threshold (up to 0.9)
|
||||
threshold = Math.min(0.9, 0.8 + (hitRatio - 0.8))
|
||||
} else if (hitRatio < 0.5) {
|
||||
// Low hit ratio, decrease threshold (down to 0.6)
|
||||
threshold = Math.max(0.6, 0.8 - (0.5 - hitRatio))
|
||||
}
|
||||
}
|
||||
|
||||
// If we have storage statistics with operation counts, adjust based on operation patterns
|
||||
if (this.storageStatistics && this.storageStatistics.operations) {
|
||||
const ops = this.storageStatistics.operations
|
||||
const totalOps = ops.total || 1
|
||||
|
||||
// Calculate read/write ratio
|
||||
const readOps = ops.search || 0
|
||||
const writeOps = (ops.add || 0) + (ops.update || 0) + (ops.delete || 0)
|
||||
|
||||
if (totalOps > 100) {
|
||||
const readRatio = readOps / totalOps
|
||||
const writeRatio = writeOps / totalOps
|
||||
|
||||
// For read-heavy workloads, use higher threshold
|
||||
// For write-heavy workloads, use lower threshold
|
||||
if (readRatio > 0.8) {
|
||||
// Read-heavy, increase threshold slightly
|
||||
threshold = Math.min(0.9, threshold + 0.05)
|
||||
} else if (writeRatio > 0.5) {
|
||||
// Write-heavy, decrease threshold
|
||||
threshold = Math.max(0.6, threshold - 0.1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Update the eviction threshold
|
||||
this.hotCacheEvictionThreshold = threshold
|
||||
}
|
||||
|
||||
/**
|
||||
* Tune warm cache TTL based on statistics
|
||||
*
|
||||
* The warm cache TTL determines how long items remain in the warm cache.
|
||||
* It is tuned based on:
|
||||
* 1. Update frequency from operation statistics
|
||||
*
|
||||
* Algorithm:
|
||||
* - Start with a default TTL of 24 hours
|
||||
* - For frequently updated data, use a shorter TTL
|
||||
* - For rarely updated data, use a longer TTL
|
||||
*/
|
||||
private tuneWarmCacheTTL(): void {
|
||||
// Default TTL (24 hours)
|
||||
let ttl = 24 * 60 * 60 * 1000
|
||||
|
||||
// If we have storage statistics with operation counts, adjust based on update frequency
|
||||
if (this.storageStatistics && this.storageStatistics.operations) {
|
||||
const ops = this.storageStatistics.operations
|
||||
const totalOps = ops.total || 1
|
||||
const updateOps = (ops.update || 0)
|
||||
|
||||
if (totalOps > 100) {
|
||||
const updateRatio = updateOps / totalOps
|
||||
|
||||
// For frequently updated data, use shorter TTL
|
||||
// For rarely updated data, use longer TTL
|
||||
if (updateRatio > 0.3) {
|
||||
// Frequently updated, decrease TTL (down to 6 hours)
|
||||
ttl = Math.max(6 * 60 * 60 * 1000, ttl * (1 - updateRatio))
|
||||
} else if (updateRatio < 0.1) {
|
||||
// Rarely updated, increase TTL (up to 48 hours)
|
||||
ttl = Math.min(48 * 60 * 60 * 1000, ttl * (1.5 - updateRatio))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Update the warm cache TTL
|
||||
this.warmCacheTTL = ttl
|
||||
}
|
||||
|
||||
/**
|
||||
* Tune batch size based on statistics and environment
|
||||
*
|
||||
* The batch size determines how many items are processed in a single batch
|
||||
* for operations like prefetching. It is tuned based on:
|
||||
* 1. Current environment (Node.js, browser, worker)
|
||||
* 2. Available memory
|
||||
* 3. Operation patterns
|
||||
* 4. Cache hit/miss ratio
|
||||
*
|
||||
* Algorithm:
|
||||
* - Start with a default based on the environment
|
||||
* - Adjust based on available memory in browsers
|
||||
* - For bulk-heavy workloads, use a larger batch size
|
||||
* - For high hit ratios, use smaller batches (items likely in cache)
|
||||
* - For low hit ratios, use larger batches (need to fetch more items)
|
||||
*/
|
||||
private tuneBatchSize(): void {
|
||||
// Default batch size
|
||||
let batchSize = 10
|
||||
|
||||
// Adjust based on environment
|
||||
if (this.environment === Environment.NODE) {
|
||||
// Node.js can handle larger batches
|
||||
batchSize = 20
|
||||
} else if (this.environment === Environment.BROWSER) {
|
||||
// Browsers might need smaller batches
|
||||
batchSize = 10
|
||||
|
||||
// If we have memory information, adjust accordingly
|
||||
if (navigator.deviceMemory) {
|
||||
// Scale batch size with available memory
|
||||
batchSize = Math.max(5, Math.min(20, Math.floor(navigator.deviceMemory * 2)))
|
||||
}
|
||||
}
|
||||
|
||||
// If we have storage statistics with operation counts, adjust based on operation patterns
|
||||
if (this.storageStatistics && this.storageStatistics.operations) {
|
||||
const ops = this.storageStatistics.operations
|
||||
const totalOps = ops.total || 1
|
||||
const bulkOps = (ops.search || 0)
|
||||
|
||||
if (totalOps > 100) {
|
||||
const bulkRatio = bulkOps / totalOps
|
||||
|
||||
// For bulk-heavy workloads, use larger batch size
|
||||
if (bulkRatio > 0.7) {
|
||||
// Bulk-heavy, increase batch size (up to 2x)
|
||||
batchSize = Math.min(50, Math.ceil(batchSize * 1.5))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Adjust based on hit/miss ratio if we have enough data
|
||||
const totalAccesses = this.stats.hits + this.stats.misses
|
||||
if (totalAccesses > 100) {
|
||||
const hitRatio = this.stats.hits / totalAccesses
|
||||
|
||||
// If hit ratio is high, we can use smaller batches
|
||||
// If hit ratio is low, we might need larger batches
|
||||
if (hitRatio > 0.8) {
|
||||
// High hit ratio, decrease batch size slightly
|
||||
batchSize = Math.max(5, Math.floor(batchSize * 0.8))
|
||||
} else if (hitRatio < 0.5) {
|
||||
// Low hit ratio, increase batch size
|
||||
batchSize = Math.min(50, Math.ceil(batchSize * 1.2))
|
||||
}
|
||||
}
|
||||
|
||||
// Update the batch size
|
||||
this.batchSize = batchSize
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect the appropriate warm storage type based on environment
|
||||
*/
|
||||
private detectWarmStorageType(): StorageType {
|
||||
if (this.environment === Environment.BROWSER) {
|
||||
// Use OPFS if available, otherwise use memory
|
||||
if ('storage' in navigator && 'getDirectory' in navigator.storage) {
|
||||
return StorageType.OPFS
|
||||
}
|
||||
return StorageType.MEMORY
|
||||
} else if (this.environment === Environment.WORKER) {
|
||||
// Use OPFS if available, otherwise use memory
|
||||
if ('storage' in self && 'getDirectory' in (self as WorkerGlobalScope).storage!) {
|
||||
return StorageType.OPFS
|
||||
}
|
||||
return StorageType.MEMORY
|
||||
} else {
|
||||
// In Node.js, use filesystem
|
||||
return StorageType.FILESYSTEM
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect the appropriate cold storage type based on environment
|
||||
*/
|
||||
private detectColdStorageType(): StorageType {
|
||||
if (this.environment === Environment.BROWSER) {
|
||||
// Use OPFS if available, otherwise use memory
|
||||
if ('storage' in navigator && 'getDirectory' in navigator.storage) {
|
||||
return StorageType.OPFS
|
||||
}
|
||||
return StorageType.MEMORY
|
||||
} else if (this.environment === Environment.WORKER) {
|
||||
// Use OPFS if available, otherwise use memory
|
||||
if ('storage' in self && 'getDirectory' in (self as WorkerGlobalScope).storage!) {
|
||||
return StorageType.OPFS
|
||||
}
|
||||
return StorageType.MEMORY
|
||||
} else {
|
||||
// In Node.js, use S3 if configured, otherwise filesystem
|
||||
return StorageType.S3
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize warm storage adapter
|
||||
*/
|
||||
private initializeWarmStorage(): any {
|
||||
// Implementation depends on the detected storage type
|
||||
// For now, return null as this will be provided by the storage adapter
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize cold storage adapter
|
||||
*/
|
||||
private initializeColdStorage(): any {
|
||||
// Implementation depends on the detected storage type
|
||||
// For now, return null as this will be provided by the storage adapter
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Get an item from cache, trying each level in order
|
||||
* @param id The item ID
|
||||
* @returns The cached item or null if not found
|
||||
*/
|
||||
public async get(id: string): Promise<T | null> {
|
||||
// Check if it's time to tune parameters
|
||||
await this.checkAndTuneParameters()
|
||||
|
||||
// Try hot cache first (fastest)
|
||||
const hotCacheEntry = this.hotCache.get(id)
|
||||
if (hotCacheEntry) {
|
||||
// Update access metadata
|
||||
hotCacheEntry.lastAccessed = Date.now()
|
||||
hotCacheEntry.accessCount++
|
||||
|
||||
// Update stats
|
||||
this.stats.hits++
|
||||
|
||||
return hotCacheEntry.data
|
||||
}
|
||||
|
||||
// Try warm cache next
|
||||
try {
|
||||
const warmCacheItem = await this.getFromWarmCache(id)
|
||||
if (warmCacheItem) {
|
||||
// Promote to hot cache
|
||||
this.addToHotCache(id, warmCacheItem)
|
||||
|
||||
// Update stats
|
||||
this.stats.hits++
|
||||
|
||||
return warmCacheItem
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn(`Error accessing warm cache for ${id}:`, error)
|
||||
}
|
||||
|
||||
// Finally, try cold storage
|
||||
try {
|
||||
const coldStorageItem = await this.getFromColdStorage(id)
|
||||
if (coldStorageItem) {
|
||||
// Promote to hot and warm caches
|
||||
this.addToHotCache(id, coldStorageItem)
|
||||
await this.addToWarmCache(id, coldStorageItem)
|
||||
|
||||
// Update stats
|
||||
this.stats.misses++
|
||||
|
||||
return coldStorageItem
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn(`Error accessing cold storage for ${id}:`, error)
|
||||
}
|
||||
|
||||
// Item not found in any cache level
|
||||
this.stats.misses++
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Get an item from warm cache
|
||||
* @param id The item ID
|
||||
* @returns The cached item or null if not found
|
||||
*/
|
||||
private async getFromWarmCache(id: string): Promise<T | null> {
|
||||
if (!this.warmStorage) return null
|
||||
|
||||
try {
|
||||
return await this.warmStorage.get(id)
|
||||
} catch (error) {
|
||||
console.warn(`Error getting item ${id} from warm cache:`, error)
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get an item from cold storage
|
||||
* @param id The item ID
|
||||
* @returns The item or null if not found
|
||||
*/
|
||||
private async getFromColdStorage(id: string): Promise<T | null> {
|
||||
if (!this.coldStorage) return null
|
||||
|
||||
try {
|
||||
return await this.coldStorage.get(id)
|
||||
} catch (error) {
|
||||
console.warn(`Error getting item ${id} from cold storage:`, error)
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Add an item to hot cache
|
||||
* @param id The item ID
|
||||
* @param item The item to cache
|
||||
*/
|
||||
private addToHotCache(id: string, item: T): void {
|
||||
// Check if we need to evict items
|
||||
if (this.hotCache.size >= this.hotCacheMaxSize * this.hotCacheEvictionThreshold) {
|
||||
this.evictFromHotCache()
|
||||
}
|
||||
|
||||
// Add to hot cache
|
||||
this.hotCache.set(id, {
|
||||
data: item,
|
||||
lastAccessed: Date.now(),
|
||||
accessCount: 1,
|
||||
expiresAt: null // Hot cache items don't expire
|
||||
})
|
||||
|
||||
// Update stats
|
||||
this.stats.size = this.hotCache.size
|
||||
}
|
||||
|
||||
/**
|
||||
* Add an item to warm cache
|
||||
* @param id The item ID
|
||||
* @param item The item to cache
|
||||
*/
|
||||
private async addToWarmCache(id: string, item: T): Promise<void> {
|
||||
if (!this.warmStorage) return
|
||||
|
||||
try {
|
||||
// Add to warm cache with TTL
|
||||
await this.warmStorage.set(id, item, {
|
||||
ttl: this.warmCacheTTL
|
||||
})
|
||||
} catch (error) {
|
||||
console.warn(`Error adding item ${id} to warm cache:`, error)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Evict items from hot cache based on LRU policy
|
||||
*/
|
||||
private evictFromHotCache(): void {
|
||||
// Find the least recently used items
|
||||
const entries = Array.from(this.hotCache.entries())
|
||||
|
||||
// Sort by last accessed time (oldest first)
|
||||
entries.sort((a, b) => a[1].lastAccessed - b[1].lastAccessed)
|
||||
|
||||
// Remove the oldest 20% of items
|
||||
const itemsToRemove = Math.ceil(this.hotCache.size * 0.2)
|
||||
for (let i = 0; i < itemsToRemove && i < entries.length; i++) {
|
||||
this.hotCache.delete(entries[i][0])
|
||||
this.stats.evictions++
|
||||
}
|
||||
|
||||
// Update stats
|
||||
this.stats.size = this.hotCache.size
|
||||
|
||||
if (process.env.DEBUG) {
|
||||
console.log(`Evicted ${itemsToRemove} items from hot cache, new size: ${this.hotCache.size}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set an item in all cache levels
|
||||
* @param id The item ID
|
||||
* @param item The item to cache
|
||||
*/
|
||||
public async set(id: string, item: T): Promise<void> {
|
||||
// Add to hot cache
|
||||
this.addToHotCache(id, item)
|
||||
|
||||
// Add to warm cache
|
||||
await this.addToWarmCache(id, item)
|
||||
|
||||
// Add to cold storage
|
||||
if (this.coldStorage) {
|
||||
try {
|
||||
await this.coldStorage.set(id, item)
|
||||
} catch (error) {
|
||||
console.warn(`Error adding item ${id} to cold storage:`, error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete an item from all cache levels
|
||||
* @param id The item ID to delete
|
||||
*/
|
||||
public async delete(id: string): Promise<void> {
|
||||
// Remove from hot cache
|
||||
this.hotCache.delete(id)
|
||||
|
||||
// Remove from warm cache
|
||||
if (this.warmStorage) {
|
||||
try {
|
||||
await this.warmStorage.delete(id)
|
||||
} catch (error) {
|
||||
console.warn(`Error deleting item ${id} from warm cache:`, error)
|
||||
}
|
||||
}
|
||||
|
||||
// Remove from cold storage
|
||||
if (this.coldStorage) {
|
||||
try {
|
||||
await this.coldStorage.delete(id)
|
||||
} catch (error) {
|
||||
console.warn(`Error deleting item ${id} from cold storage:`, error)
|
||||
}
|
||||
}
|
||||
|
||||
// Update stats
|
||||
this.stats.size = this.hotCache.size
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all cache levels
|
||||
*/
|
||||
public async clear(): Promise<void> {
|
||||
// Clear hot cache
|
||||
this.hotCache.clear()
|
||||
|
||||
// Clear warm cache
|
||||
if (this.warmStorage) {
|
||||
try {
|
||||
await this.warmStorage.clear()
|
||||
} catch (error) {
|
||||
console.warn('Error clearing warm cache:', error)
|
||||
}
|
||||
}
|
||||
|
||||
// Clear cold storage
|
||||
if (this.coldStorage) {
|
||||
try {
|
||||
await this.coldStorage.clear()
|
||||
} catch (error) {
|
||||
console.warn('Error clearing cold storage:', error)
|
||||
}
|
||||
}
|
||||
|
||||
// Reset stats
|
||||
this.stats = {
|
||||
hits: 0,
|
||||
misses: 0,
|
||||
evictions: 0,
|
||||
size: 0,
|
||||
maxSize: this.hotCacheMaxSize
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get cache statistics
|
||||
* @returns Cache statistics
|
||||
*/
|
||||
public getStats(): CacheStats {
|
||||
return { ...this.stats }
|
||||
}
|
||||
|
||||
/**
|
||||
* Prefetch items based on ID patterns or relationships
|
||||
* @param ids Array of IDs to prefetch
|
||||
*/
|
||||
public async prefetch(ids: string[]): Promise<void> {
|
||||
// Check if it's time to tune parameters
|
||||
await this.checkAndTuneParameters()
|
||||
|
||||
// Prefetch in batches to avoid overwhelming the system
|
||||
const batches: string[][] = []
|
||||
|
||||
// Split into batches using the configurable batch size
|
||||
for (let i = 0; i < ids.length; i += this.batchSize) {
|
||||
const batch = ids.slice(i, i + this.batchSize)
|
||||
batches.push(batch)
|
||||
}
|
||||
|
||||
// Process each batch
|
||||
for (const batch of batches) {
|
||||
await Promise.all(
|
||||
batch.map(async (id) => {
|
||||
// Skip if already in hot cache
|
||||
if (this.hotCache.has(id)) return
|
||||
|
||||
try {
|
||||
// Try to get from any cache level
|
||||
await this.get(id)
|
||||
} catch (error) {
|
||||
// Ignore errors during prefetching
|
||||
if (process.env.DEBUG) {
|
||||
console.warn(`Error prefetching ${id}:`, error)
|
||||
}
|
||||
}
|
||||
})
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if it's time to tune parameters and do so if needed
|
||||
* This is called before operations that might benefit from tuned parameters
|
||||
*
|
||||
* This method serves as a checkpoint for auto-tuning, ensuring that:
|
||||
* 1. Parameters are tuned periodically based on the auto-tune interval
|
||||
* 2. Tuning happens before critical operations that would benefit from optimized parameters
|
||||
* 3. Tuning doesn't happen too frequently, which could impact performance
|
||||
*
|
||||
* By calling this method before get(), getMany(), and prefetch() operations,
|
||||
* we ensure that the cache parameters are optimized for the current workload
|
||||
* without adding unnecessary overhead to every operation.
|
||||
*/
|
||||
private async checkAndTuneParameters(): Promise<void> {
|
||||
// Skip if auto-tuning is disabled
|
||||
if (!this.autoTune) return
|
||||
|
||||
// Check if it's time to tune parameters
|
||||
const now = Date.now()
|
||||
if (now - this.lastAutoTuneTime >= this.autoTuneInterval) {
|
||||
await this.tuneParameters()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get multiple items at once, optimizing for batch retrieval
|
||||
* @param ids Array of IDs to get
|
||||
* @returns Map of ID to item
|
||||
*/
|
||||
public async getMany(ids: string[]): Promise<Map<string, T>> {
|
||||
// Check if it's time to tune parameters
|
||||
await this.checkAndTuneParameters()
|
||||
|
||||
const result = new Map<string, T>()
|
||||
|
||||
// First check hot cache for all IDs
|
||||
const missingIds: string[] = []
|
||||
for (const id of ids) {
|
||||
const hotCacheEntry = this.hotCache.get(id)
|
||||
if (hotCacheEntry) {
|
||||
// Update access metadata
|
||||
hotCacheEntry.lastAccessed = Date.now()
|
||||
hotCacheEntry.accessCount++
|
||||
|
||||
// Add to result
|
||||
result.set(id, hotCacheEntry.data)
|
||||
|
||||
// Update stats
|
||||
this.stats.hits++
|
||||
} else {
|
||||
missingIds.push(id)
|
||||
}
|
||||
}
|
||||
|
||||
if (missingIds.length === 0) {
|
||||
return result
|
||||
}
|
||||
|
||||
// Try to get missing items from warm cache
|
||||
if (this.warmStorage) {
|
||||
try {
|
||||
const warmCacheItems = await this.warmStorage.getMany(missingIds)
|
||||
for (const [id, item] of warmCacheItems.entries()) {
|
||||
if (item) {
|
||||
// Promote to hot cache
|
||||
this.addToHotCache(id, item)
|
||||
|
||||
// Add to result
|
||||
result.set(id, item)
|
||||
|
||||
// Update stats
|
||||
this.stats.hits++
|
||||
|
||||
// Remove from missing IDs
|
||||
const index = missingIds.indexOf(id)
|
||||
if (index !== -1) {
|
||||
missingIds.splice(index, 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn('Error accessing warm cache for batch:', error)
|
||||
}
|
||||
}
|
||||
|
||||
if (missingIds.length === 0) {
|
||||
return result
|
||||
}
|
||||
|
||||
// Try to get remaining missing items from cold storage
|
||||
if (this.coldStorage) {
|
||||
try {
|
||||
const coldStorageItems = await this.coldStorage.getMany(missingIds)
|
||||
for (const [id, item] of coldStorageItems.entries()) {
|
||||
if (item) {
|
||||
// Promote to hot and warm caches
|
||||
this.addToHotCache(id, item)
|
||||
await this.addToWarmCache(id, item)
|
||||
|
||||
// Add to result
|
||||
result.set(id, item)
|
||||
|
||||
// Update stats
|
||||
this.stats.misses++
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.warn('Error accessing cold storage for batch:', error)
|
||||
}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the storage adapters for warm and cold caches
|
||||
* @param warmStorage Warm cache storage adapter
|
||||
* @param coldStorage Cold storage adapter
|
||||
*/
|
||||
public setStorageAdapters(warmStorage: any, coldStorage: any): void {
|
||||
this.warmStorage = warmStorage
|
||||
this.coldStorage = coldStorage
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue