fix: move metadata routing to base class, fix GCS/S3 system key crashes
Critical fix for GCS/S3 native storage adapters crashing on metadata index keys. Problem: - GCS and S3 adapters crashed with "Invalid UUID format" errors - System keys like __metadata_field_index__status are NOT UUIDs - Adapters incorrectly tried to shard all metadata keys as UUIDs Solution: - Move sharding/routing logic from adapters to BaseStorage class - Add analyzeKey() method to detect system keys vs entity UUIDs - System keys route to _system/ directory (no sharding) - Entity UUIDs route to sharded directories (256 shards) - All adapters now implement 4 primitive operations: * writeObjectToPath(path, data) * readObjectFromPath(path) * deleteObjectFromPath(path) * listObjectsUnderPath(prefix) Benefits: - Impossible for future adapters to repeat this mistake - Zero breaking changes, full backward compatibility - No data migration required - Cleaner architecture with better separation of concerns Updated adapters: GcsStorage, S3CompatibleStorage, OPFSStorage, FileSystemStorage, MemoryStorage Added: docs/architecture/data-storage-architecture.md Updated: README.md with architecture docs link
This commit is contained in:
parent
13303c20c2
commit
1966c39f24
9 changed files with 1330 additions and 630 deletions
|
|
@ -555,33 +555,86 @@ export class FileSystemStorage extends BaseStorage {
|
|||
}
|
||||
|
||||
/**
|
||||
* Save metadata to storage
|
||||
* Primitive operation: Write object to path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
public async saveMetadata(id: string, metadata: any): Promise<void> {
|
||||
protected async writeObjectToPath(pathStr: string, data: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const filePath = path.join(this.metadataDir, `${id}.json`)
|
||||
await fs.promises.writeFile(filePath, JSON.stringify(metadata, null, 2))
|
||||
const fullPath = path.join(this.rootDir, pathStr)
|
||||
await this.ensureDirectoryExists(path.dirname(fullPath))
|
||||
await fs.promises.writeFile(fullPath, JSON.stringify(data, null, 2))
|
||||
}
|
||||
|
||||
/**
|
||||
* Get metadata from storage
|
||||
* Primitive operation: Read object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
public async getMetadata(id: string): Promise<any | null> {
|
||||
protected async readObjectFromPath(pathStr: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const filePath = path.join(this.metadataDir, `${id}.json`)
|
||||
const fullPath = path.join(this.rootDir, pathStr)
|
||||
try {
|
||||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||||
const data = await fs.promises.readFile(fullPath, 'utf-8')
|
||||
return JSON.parse(data)
|
||||
} catch (error: any) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.error(`Error reading metadata ${id}:`, error)
|
||||
if (error.code === 'ENOENT') {
|
||||
return null
|
||||
}
|
||||
console.error(`Error reading object from ${pathStr}:`, error)
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Primitive operation: Delete object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async deleteObjectFromPath(pathStr: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const fullPath = path.join(this.rootDir, pathStr)
|
||||
try {
|
||||
await fs.promises.unlink(fullPath)
|
||||
} catch (error: any) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.error(`Error deleting object from ${pathStr}:`, error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Primitive operation: List objects under path prefix
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const fullPath = path.join(this.rootDir, prefix)
|
||||
const paths: string[] = []
|
||||
|
||||
try {
|
||||
const entries = await fs.promises.readdir(fullPath, { withFileTypes: true })
|
||||
|
||||
for (const entry of entries) {
|
||||
if (entry.isFile() && entry.name.endsWith('.json')) {
|
||||
paths.push(path.join(prefix, entry.name))
|
||||
} else if (entry.isDirectory()) {
|
||||
const subdirPaths = await this.listObjectsUnderPath(path.join(prefix, entry.name))
|
||||
paths.push(...subdirPaths)
|
||||
}
|
||||
}
|
||||
|
||||
return paths.sort()
|
||||
} catch (error: any) {
|
||||
if (error.code === 'ENOENT') {
|
||||
return []
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
|
||||
* FileSystem implementation uses controlled concurrency to prevent too many file reads
|
||||
|
|
@ -598,7 +651,7 @@ export class FileSystemStorage extends BaseStorage {
|
|||
|
||||
const batchPromises = batch.map(async (id) => {
|
||||
try {
|
||||
const metadata = await this.getNounMetadata(id)
|
||||
const metadata = await this.getMetadata(id)
|
||||
return { id, metadata }
|
||||
} catch (error) {
|
||||
console.debug(`Failed to read metadata for ${id}:`, error)
|
||||
|
|
@ -621,85 +674,6 @@ export class FileSystemStorage extends BaseStorage {
|
|||
return results
|
||||
}
|
||||
|
||||
/**
|
||||
* Save noun metadata to storage
|
||||
*/
|
||||
protected async saveNounMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with noun vectors)
|
||||
const filePath = this.getShardedPath(this.nounMetadataDir, id)
|
||||
|
||||
// Ensure shard directory exists
|
||||
const shardDir = path.dirname(filePath)
|
||||
await fs.promises.mkdir(shardDir, { recursive: true })
|
||||
|
||||
await fs.promises.writeFile(filePath, JSON.stringify(metadata, null, 2))
|
||||
}
|
||||
|
||||
/**
|
||||
* Get noun metadata from storage
|
||||
*/
|
||||
public async getNounMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with noun vectors)
|
||||
const filePath = this.getShardedPath(this.nounMetadataDir, id)
|
||||
try {
|
||||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||||
return JSON.parse(data)
|
||||
} catch (error: any) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.error(`Error reading noun metadata ${id}:`, error)
|
||||
}
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save verb metadata to storage
|
||||
*/
|
||||
protected async saveVerbMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
console.log(`[DEBUG] Saving verb metadata for ${id} to: ${this.verbMetadataDir}`)
|
||||
const filePath = path.join(this.verbMetadataDir, `${id}.json`)
|
||||
console.log(`[DEBUG] Full file path: ${filePath}`)
|
||||
|
||||
try {
|
||||
await this.ensureDirectoryExists(path.dirname(filePath))
|
||||
console.log(`[DEBUG] Directory ensured: ${path.dirname(filePath)}`)
|
||||
|
||||
await fs.promises.writeFile(filePath, JSON.stringify(metadata, null, 2))
|
||||
console.log(`[DEBUG] File written successfully: ${filePath}`)
|
||||
|
||||
// Verify the file was actually written
|
||||
const exists = await fs.promises.access(filePath).then(() => true).catch(() => false)
|
||||
console.log(`[DEBUG] File exists after write: ${exists}`)
|
||||
} catch (error) {
|
||||
console.error(`[DEBUG] Error saving verb metadata:`, error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get verb metadata from storage
|
||||
*/
|
||||
public async getVerbMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const filePath = path.join(this.verbMetadataDir, `${id}.json`)
|
||||
try {
|
||||
const data = await fs.promises.readFile(filePath, 'utf-8')
|
||||
return JSON.parse(data)
|
||||
} catch (error: any) {
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.error(`Error reading verb metadata ${id}:`, error)
|
||||
}
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get nouns with pagination support
|
||||
* @param options Pagination options
|
||||
|
|
|
|||
|
|
@ -583,134 +583,106 @@ export class GcsStorage extends BaseStorage {
|
|||
}
|
||||
|
||||
/**
|
||||
* Save noun metadata to storage (internal implementation)
|
||||
* Write an object to a specific path in GCS
|
||||
* Primitive operation required by base class
|
||||
* @protected
|
||||
*/
|
||||
protected async saveNounMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
protected async writeObjectToPath(path: string, data: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Use UUID-based sharding for metadata (consistent with noun vectors)
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
const key = `${this.metadataPrefix}${shardId}/${id}.json`
|
||||
this.logger.trace(`Writing object to path: ${path}`)
|
||||
|
||||
this.logger.trace(`Saving noun metadata for ${id} to key: ${key}`)
|
||||
|
||||
// Save to GCS
|
||||
const file = this.bucket!.file(key)
|
||||
await file.save(JSON.stringify(metadata, null, 2), {
|
||||
const file = this.bucket!.file(path)
|
||||
await file.save(JSON.stringify(data, null, 2), {
|
||||
contentType: 'application/json',
|
||||
resumable: false
|
||||
})
|
||||
|
||||
this.logger.debug(`Noun metadata for ${id} saved successfully`)
|
||||
this.logger.trace(`Object written successfully to ${path}`)
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to save noun metadata for ${id}:`, error)
|
||||
throw new Error(`Failed to save noun metadata for ${id}: ${error}`)
|
||||
this.logger.error(`Failed to write object to ${path}:`, error)
|
||||
throw new Error(`Failed to write object to ${path}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save metadata to storage (public API - delegates to saveNounMetadata_internal)
|
||||
* Read an object from a specific path in GCS
|
||||
* Primitive operation required by base class
|
||||
* @protected
|
||||
*/
|
||||
public async saveMetadata(id: string, metadata: any): Promise<void> {
|
||||
return this.saveNounMetadata_internal(id, metadata)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get metadata from storage (public API - delegates to getNounMetadata)
|
||||
*/
|
||||
public async getMetadata(id: string): Promise<any | null> {
|
||||
return this.getNounMetadata(id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get noun metadata from storage
|
||||
*/
|
||||
public async getNounMetadata(id: string): Promise<any | null> {
|
||||
protected async readObjectFromPath(path: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Use UUID-based sharding for metadata
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
const key = `${this.metadataPrefix}${shardId}/${id}.json`
|
||||
this.logger.trace(`Reading object from path: ${path}`)
|
||||
|
||||
this.logger.trace(`Getting noun metadata for ${id} from key: ${key}`)
|
||||
|
||||
// Download from GCS
|
||||
const file = this.bucket!.file(key)
|
||||
const file = this.bucket!.file(path)
|
||||
const [contents] = await file.download()
|
||||
|
||||
// Parse JSON
|
||||
const metadata = JSON.parse(contents.toString())
|
||||
const data = JSON.parse(contents.toString())
|
||||
|
||||
this.logger.trace(`Successfully retrieved noun metadata for ${id}`)
|
||||
return metadata
|
||||
this.logger.trace(`Object read successfully from ${path}`)
|
||||
return data
|
||||
} catch (error: any) {
|
||||
// Check if this is a "not found" error
|
||||
if (error.code === 404) {
|
||||
this.logger.trace(`Noun metadata not found for ${id}`)
|
||||
this.logger.trace(`Object not found at ${path}`)
|
||||
return null
|
||||
}
|
||||
|
||||
// For other types of errors, convert to BrainyError
|
||||
throw BrainyError.fromError(error, `getNounMetadata(${id})`)
|
||||
this.logger.error(`Failed to read object from ${path}:`, error)
|
||||
throw BrainyError.fromError(error, `readObjectFromPath(${path})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save verb metadata to storage (internal implementation)
|
||||
* Delete an object from a specific path in GCS
|
||||
* Primitive operation required by base class
|
||||
* @protected
|
||||
*/
|
||||
protected async saveVerbMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
protected async deleteObjectFromPath(path: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const key = `${this.verbMetadataPrefix}${id}.json`
|
||||
this.logger.trace(`Deleting object at path: ${path}`)
|
||||
|
||||
this.logger.trace(`Saving verb metadata for ${id} to key: ${key}`)
|
||||
const file = this.bucket!.file(path)
|
||||
await file.delete()
|
||||
|
||||
// Save to GCS
|
||||
const file = this.bucket!.file(key)
|
||||
await file.save(JSON.stringify(metadata, null, 2), {
|
||||
contentType: 'application/json',
|
||||
resumable: false
|
||||
})
|
||||
|
||||
this.logger.debug(`Verb metadata for ${id} saved successfully`)
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to save verb metadata for ${id}:`, error)
|
||||
throw new Error(`Failed to save verb metadata for ${id}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get verb metadata from storage
|
||||
*/
|
||||
public async getVerbMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const key = `${this.verbMetadataPrefix}${id}.json`
|
||||
|
||||
this.logger.trace(`Getting verb metadata for ${id} from key: ${key}`)
|
||||
|
||||
// Download from GCS
|
||||
const file = this.bucket!.file(key)
|
||||
const [contents] = await file.download()
|
||||
|
||||
// Parse JSON
|
||||
const metadata = JSON.parse(contents.toString())
|
||||
|
||||
this.logger.trace(`Successfully retrieved verb metadata for ${id}`)
|
||||
return metadata
|
||||
this.logger.trace(`Object deleted successfully from ${path}`)
|
||||
} catch (error: any) {
|
||||
// Check if this is a "not found" error
|
||||
// If already deleted (404), treat as success
|
||||
if (error.code === 404) {
|
||||
this.logger.trace(`Verb metadata not found for ${id}`)
|
||||
return null
|
||||
this.logger.trace(`Object at ${path} not found (already deleted)`)
|
||||
return
|
||||
}
|
||||
|
||||
// For other types of errors, convert to BrainyError
|
||||
throw BrainyError.fromError(error, `getVerbMetadata(${id})`)
|
||||
this.logger.error(`Failed to delete object from ${path}:`, error)
|
||||
throw new Error(`Failed to delete object from ${path}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* List all objects under a specific prefix in GCS
|
||||
* Primitive operation required by base class
|
||||
* @protected
|
||||
*/
|
||||
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
this.logger.trace(`Listing objects under prefix: ${prefix}`)
|
||||
|
||||
const [files] = await this.bucket!.getFiles({ prefix })
|
||||
|
||||
const paths = files.map((file: any) => file.name).filter((name: string) => name && name.length > 0)
|
||||
|
||||
this.logger.trace(`Found ${paths.length} objects under ${prefix}`)
|
||||
return paths
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to list objects under ${prefix}:`, error)
|
||||
throw new Error(`Failed to list objects under ${prefix}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -17,11 +17,22 @@ export class MemoryStorage extends BaseStorage {
|
|||
// Single map of noun ID to noun
|
||||
private nouns: Map<string, HNSWNoun> = new Map()
|
||||
private verbs: Map<string, HNSWVerb> = new Map()
|
||||
private metadata: Map<string, any> = new Map()
|
||||
private nounMetadata: Map<string, any> = new Map()
|
||||
private verbMetadata: Map<string, any> = new Map()
|
||||
private statistics: StatisticsData | null = null
|
||||
|
||||
// Unified object store for primitive operations (replaces metadata, nounMetadata, verbMetadata)
|
||||
private objectStore: Map<string, any> = new Map()
|
||||
|
||||
// Backward compatibility aliases
|
||||
private get metadata(): Map<string, any> {
|
||||
return this.objectStore
|
||||
}
|
||||
private get nounMetadata(): Map<string, any> {
|
||||
return this.objectStore
|
||||
}
|
||||
private get verbMetadata(): Map<string, any> {
|
||||
return this.objectStore
|
||||
}
|
||||
|
||||
constructor() {
|
||||
super()
|
||||
}
|
||||
|
|
@ -520,22 +531,46 @@ export class MemoryStorage extends BaseStorage {
|
|||
}
|
||||
|
||||
/**
|
||||
* Save metadata to storage
|
||||
* Primitive operation: Write object to path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
public async saveMetadata(id: string, metadata: any): Promise<void> {
|
||||
this.metadata.set(id, JSON.parse(JSON.stringify(metadata)))
|
||||
protected async writeObjectToPath(path: string, data: any): Promise<void> {
|
||||
// Store in unified object store using path as key
|
||||
this.objectStore.set(path, JSON.parse(JSON.stringify(data)))
|
||||
}
|
||||
|
||||
/**
|
||||
* Get metadata from storage
|
||||
* Primitive operation: Read object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
public async getMetadata(id: string): Promise<any | null> {
|
||||
const metadata = this.metadata.get(id)
|
||||
if (!metadata) {
|
||||
protected async readObjectFromPath(path: string): Promise<any | null> {
|
||||
const data = this.objectStore.get(path)
|
||||
if (!data) {
|
||||
return null
|
||||
}
|
||||
return JSON.parse(JSON.stringify(data))
|
||||
}
|
||||
|
||||
return JSON.parse(JSON.stringify(metadata))
|
||||
/**
|
||||
* Primitive operation: Delete object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async deleteObjectFromPath(path: string): Promise<void> {
|
||||
this.objectStore.delete(path)
|
||||
}
|
||||
|
||||
/**
|
||||
* Primitive operation: List objects under path prefix
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
|
||||
const paths: string[] = []
|
||||
for (const key of this.objectStore.keys()) {
|
||||
if (key.startsWith(prefix)) {
|
||||
paths.push(key)
|
||||
}
|
||||
}
|
||||
return paths.sort()
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -544,75 +579,27 @@ export class MemoryStorage extends BaseStorage {
|
|||
*/
|
||||
public async getMetadataBatch(ids: string[]): Promise<Map<string, any>> {
|
||||
const results = new Map<string, any>()
|
||||
|
||||
|
||||
// Memory storage can handle all IDs at once since it's in-memory
|
||||
for (const id of ids) {
|
||||
const metadata = this.metadata.get(id)
|
||||
const metadata = await this.getMetadata(id)
|
||||
if (metadata) {
|
||||
// Deep clone to prevent mutation
|
||||
results.set(id, JSON.parse(JSON.stringify(metadata)))
|
||||
results.set(id, metadata)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
/**
|
||||
* Save noun metadata to storage (internal implementation)
|
||||
*/
|
||||
protected async saveNounMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
this.nounMetadata.set(id, JSON.parse(JSON.stringify(metadata)))
|
||||
}
|
||||
|
||||
/**
|
||||
* Get noun metadata from storage
|
||||
*/
|
||||
public async getNounMetadata(id: string): Promise<any | null> {
|
||||
const metadata = this.nounMetadata.get(id)
|
||||
if (!metadata) {
|
||||
return null
|
||||
}
|
||||
|
||||
return JSON.parse(JSON.stringify(metadata))
|
||||
}
|
||||
|
||||
/**
|
||||
* Save verb metadata to storage (internal implementation)
|
||||
*/
|
||||
protected async saveVerbMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
const isNew = !this.verbMetadata.has(id)
|
||||
this.verbMetadata.set(id, JSON.parse(JSON.stringify(metadata)))
|
||||
|
||||
// Update counts for new verbs
|
||||
if (isNew) {
|
||||
const type = metadata?.verb || metadata?.type || 'default'
|
||||
this.incrementVerbCount(type)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get verb metadata from storage
|
||||
*/
|
||||
public async getVerbMetadata(id: string): Promise<any | null> {
|
||||
const metadata = this.verbMetadata.get(id)
|
||||
if (!metadata) {
|
||||
return null
|
||||
}
|
||||
|
||||
return JSON.parse(JSON.stringify(metadata))
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all data from storage
|
||||
*/
|
||||
public async clear(): Promise<void> {
|
||||
this.nouns.clear()
|
||||
this.verbs.clear()
|
||||
this.metadata.clear()
|
||||
this.nounMetadata.clear()
|
||||
this.verbMetadata.clear()
|
||||
this.objectStore.clear()
|
||||
this.statistics = null
|
||||
|
||||
|
||||
// Clear the statistics cache
|
||||
this.statisticsCache = null
|
||||
this.statisticsModified = false
|
||||
|
|
@ -634,7 +621,7 @@ export class MemoryStorage extends BaseStorage {
|
|||
details: {
|
||||
nodeCount: this.nouns.size,
|
||||
edgeCount: this.verbs.size,
|
||||
metadataCount: this.metadata.size
|
||||
metadataCount: this.objectStore.size
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -640,47 +640,149 @@ export class OPFSStorage extends BaseStorage {
|
|||
}
|
||||
|
||||
/**
|
||||
* Save metadata to storage
|
||||
* Primitive operation: Write object to path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
public async saveMetadata(id: string, metadata: any): Promise<void> {
|
||||
protected async writeObjectToPath(path: string, data: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Create or get the file for this metadata
|
||||
const fileHandle = await this.metadataDir!.getFileHandle(`${id}.json`, {
|
||||
create: true
|
||||
})
|
||||
// Parse path to get directory structure and filename
|
||||
// Path format: "dir1/dir2/file.json"
|
||||
const parts = path.split('/')
|
||||
const filename = parts.pop()!
|
||||
|
||||
// Write the metadata to the file
|
||||
// Navigate to the correct directory, creating as needed
|
||||
let currentDir = this.rootDir!
|
||||
for (const dirName of parts) {
|
||||
currentDir = await currentDir.getDirectoryHandle(dirName, { create: true })
|
||||
}
|
||||
|
||||
// Create or get the file
|
||||
const fileHandle = await currentDir.getFileHandle(filename, { create: true })
|
||||
|
||||
// Write the data to the file
|
||||
const writable = await fileHandle.createWritable()
|
||||
await writable.write(JSON.stringify(metadata))
|
||||
await writable.write(JSON.stringify(data, null, 2))
|
||||
await writable.close()
|
||||
} catch (error) {
|
||||
console.error(`Failed to save metadata ${id}:`, error)
|
||||
throw new Error(`Failed to save metadata ${id}: ${error}`)
|
||||
console.error(`Failed to write object to ${path}:`, error)
|
||||
throw new Error(`Failed to write object to ${path}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get metadata from storage
|
||||
* Primitive operation: Read object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
public async getMetadata(id: string): Promise<any | null> {
|
||||
protected async readObjectFromPath(path: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Get the file handle for this metadata
|
||||
const fileHandle = await this.metadataDir!.getFileHandle(`${id}.json`)
|
||||
// Parse path to get directory structure and filename
|
||||
const parts = path.split('/')
|
||||
const filename = parts.pop()!
|
||||
|
||||
// Read the metadata from the file
|
||||
// Navigate to the correct directory
|
||||
let currentDir = this.rootDir!
|
||||
for (const dirName of parts) {
|
||||
currentDir = await currentDir.getDirectoryHandle(dirName)
|
||||
}
|
||||
|
||||
// Get the file handle
|
||||
const fileHandle = await currentDir.getFileHandle(filename)
|
||||
|
||||
// Read the data from the file
|
||||
const file = await fileHandle.getFile()
|
||||
const text = await file.text()
|
||||
return JSON.parse(text)
|
||||
} catch (error) {
|
||||
// Metadata not found or other error
|
||||
} catch (error: any) {
|
||||
// NotFoundError means object doesn't exist
|
||||
if (error.name === 'NotFoundError') {
|
||||
return null
|
||||
}
|
||||
console.error(`Failed to read object from ${path}:`, error)
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Primitive operation: Delete object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async deleteObjectFromPath(path: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Parse path to get directory structure and filename
|
||||
const parts = path.split('/')
|
||||
const filename = parts.pop()!
|
||||
|
||||
// Navigate to the correct directory
|
||||
let currentDir = this.rootDir!
|
||||
for (const dirName of parts) {
|
||||
currentDir = await currentDir.getDirectoryHandle(dirName)
|
||||
}
|
||||
|
||||
// Delete the file
|
||||
await currentDir.removeEntry(filename)
|
||||
} catch (error: any) {
|
||||
// NotFoundError is ok (already deleted)
|
||||
if (error.name === 'NotFoundError') {
|
||||
return
|
||||
}
|
||||
console.error(`Failed to delete object from ${path}:`, error)
|
||||
throw new Error(`Failed to delete object from ${path}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Primitive operation: List objects under path prefix
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const paths: string[] = []
|
||||
|
||||
// Parse prefix to get directory structure
|
||||
const parts = prefix.split('/')
|
||||
|
||||
// Navigate to the directory
|
||||
let currentDir = this.rootDir!
|
||||
for (const dirName of parts) {
|
||||
if (dirName) {
|
||||
currentDir = await currentDir.getDirectoryHandle(dirName)
|
||||
}
|
||||
}
|
||||
|
||||
// Recursively list all files
|
||||
const listFiles = async (dir: FileSystemDirectoryHandle, pathPrefix: string): Promise<void> => {
|
||||
for await (const [name, handle] of dir.entries()) {
|
||||
const fullPath = pathPrefix ? `${pathPrefix}/${name}` : name
|
||||
|
||||
if (handle.kind === 'file') {
|
||||
paths.push(`${prefix}${fullPath}`)
|
||||
} else if (handle.kind === 'directory') {
|
||||
await listFiles(handle as FileSystemDirectoryHandle, fullPath)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await listFiles(currentDir, '')
|
||||
|
||||
return paths
|
||||
} catch (error: any) {
|
||||
// NotFoundError means directory doesn't exist
|
||||
if (error.name === 'NotFoundError') {
|
||||
return []
|
||||
}
|
||||
console.error(`Failed to list objects under ${prefix}:`, error)
|
||||
throw new Error(`Failed to list objects under ${prefix}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
|
||||
* OPFS implementation uses controlled concurrency for file operations
|
||||
|
|
@ -720,107 +822,6 @@ export class OPFSStorage extends BaseStorage {
|
|||
return results
|
||||
}
|
||||
|
||||
/**
|
||||
* Save verb metadata to storage
|
||||
*/
|
||||
protected async saveVerbMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with verb vectors)
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
|
||||
// Get or create the shard directory
|
||||
const shardDir = await (
|
||||
this.verbMetadataDir as FileSystemDirectoryHandle
|
||||
).getDirectoryHandle(shardId, { create: true })
|
||||
|
||||
// Create or get the file in the shard directory
|
||||
const fileName = `${id}.json`
|
||||
const fileHandle = await shardDir.getFileHandle(fileName, { create: true })
|
||||
const writable = await (fileHandle as FileSystemFileHandle).createWritable()
|
||||
await writable.write(JSON.stringify(metadata, null, 2))
|
||||
await writable.close()
|
||||
}
|
||||
|
||||
/**
|
||||
* Get verb metadata from storage
|
||||
*/
|
||||
public async getVerbMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with verb vectors)
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
|
||||
const fileName = `${id}.json`
|
||||
try {
|
||||
// Get the shard directory
|
||||
const shardDir = await (
|
||||
this.verbMetadataDir as FileSystemDirectoryHandle
|
||||
).getDirectoryHandle(shardId)
|
||||
|
||||
// Get the file from the shard directory
|
||||
const fileHandle = await shardDir.getFileHandle(fileName)
|
||||
const file = await safeGetFile(fileHandle)
|
||||
const text = await file.text()
|
||||
return JSON.parse(text)
|
||||
} catch (error: any) {
|
||||
if (error.name !== 'NotFoundError') {
|
||||
console.error(`Error reading verb metadata ${id}:`, error)
|
||||
}
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save noun metadata to storage
|
||||
*/
|
||||
protected async saveNounMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with noun vectors)
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
|
||||
// Get or create the shard directory
|
||||
const shardDir = await (
|
||||
this.nounMetadataDir as FileSystemDirectoryHandle
|
||||
).getDirectoryHandle(shardId, { create: true })
|
||||
|
||||
// Create or get the file in the shard directory
|
||||
const fileName = `${id}.json`
|
||||
const fileHandle = await shardDir.getFileHandle(fileName, { create: true })
|
||||
const writable = await fileHandle.createWritable()
|
||||
await writable.write(JSON.stringify(metadata, null, 2))
|
||||
await writable.close()
|
||||
}
|
||||
|
||||
/**
|
||||
* Get noun metadata from storage
|
||||
*/
|
||||
public async getNounMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with noun vectors)
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
|
||||
const fileName = `${id}.json`
|
||||
try {
|
||||
// Get the shard directory
|
||||
const shardDir = await (
|
||||
this.nounMetadataDir as FileSystemDirectoryHandle
|
||||
).getDirectoryHandle(shardId)
|
||||
|
||||
// Get the file from the shard directory
|
||||
const fileHandle = await shardDir.getFileHandle(fileName)
|
||||
const file = await safeGetFile(fileHandle)
|
||||
const text = await file.text()
|
||||
return JSON.parse(text)
|
||||
} catch (error: any) {
|
||||
if (error.name !== 'NotFoundError') {
|
||||
console.error(`Error reading noun metadata ${id}:`, error)
|
||||
}
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all data from storage
|
||||
|
|
|
|||
|
|
@ -1926,152 +1926,120 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
}
|
||||
|
||||
/**
|
||||
* Save metadata to storage
|
||||
* Primitive operation: Write object to path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
public async saveMetadata(id: string, metadata: any): Promise<void> {
|
||||
protected async writeObjectToPath(path: string, data: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
// Apply backpressure before starting operation
|
||||
const requestId = await this.applyBackpressure()
|
||||
|
||||
try {
|
||||
// Import the PutObjectCommand only when needed
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
const body = JSON.stringify(data, null, 2)
|
||||
|
||||
const key = `${this.metadataPrefix}${id}.json`
|
||||
const body = JSON.stringify(metadata, null, 2)
|
||||
this.logger.trace(`Writing object to path: ${path}`)
|
||||
|
||||
this.logger.trace(`Saving metadata for ${id} to key: ${key}`)
|
||||
|
||||
// Save the metadata to S3-compatible storage
|
||||
const result = await this.s3Client!.send(
|
||||
await this.s3Client!.send(
|
||||
new PutObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: key,
|
||||
Key: path,
|
||||
Body: body,
|
||||
ContentType: 'application/json'
|
||||
})
|
||||
)
|
||||
|
||||
this.logger.debug(`Metadata for ${id} saved successfully`)
|
||||
this.logger.debug(`Object written successfully to ${path}`)
|
||||
|
||||
// Log the change for efficient synchronization
|
||||
await this.appendToChangeLog({
|
||||
timestamp: Date.now(),
|
||||
operation: 'add', // Could be 'update' if we track existing metadata
|
||||
operation: 'add',
|
||||
entityType: 'metadata',
|
||||
entityId: id,
|
||||
data: metadata
|
||||
entityId: path,
|
||||
data: data
|
||||
})
|
||||
|
||||
// Verify the metadata was saved by trying to retrieve it
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
try {
|
||||
const verifyResponse = await this.s3Client!.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: key
|
||||
})
|
||||
)
|
||||
|
||||
if (verifyResponse && verifyResponse.Body) {
|
||||
this.logger.trace(`Verified metadata for ${id} was saved correctly`)
|
||||
} else {
|
||||
this.logger.warn(
|
||||
`Failed to verify metadata for ${id} was saved correctly: no response or body`
|
||||
)
|
||||
}
|
||||
} catch (verifyError) {
|
||||
this.logger.warn(
|
||||
`Failed to verify metadata for ${id} was saved correctly:`,
|
||||
verifyError
|
||||
)
|
||||
}
|
||||
|
||||
// Release backpressure on success
|
||||
this.releaseBackpressure(true, requestId)
|
||||
} catch (error) {
|
||||
// Release backpressure on error
|
||||
this.releaseBackpressure(false, requestId)
|
||||
this.logger.error(`Failed to save metadata for ${id}:`, error)
|
||||
throw new Error(`Failed to save metadata for ${id}: ${error}`)
|
||||
this.logger.error(`Failed to write object to ${path}:`, error)
|
||||
throw new Error(`Failed to write object to ${path}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save verb metadata to storage
|
||||
* Primitive operation: Read object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async saveVerbMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
protected async readObjectFromPath(path: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Import the PutObjectCommand only when needed
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
const key = `${this.verbMetadataPrefix}${id}.json`
|
||||
const body = JSON.stringify(metadata, null, 2)
|
||||
|
||||
this.logger.trace(`Saving verb metadata for ${id} to key: ${key}`)
|
||||
|
||||
// Save the verb metadata to S3-compatible storage
|
||||
const result = await this.s3Client!.send(
|
||||
new PutObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: key,
|
||||
Body: body,
|
||||
ContentType: 'application/json'
|
||||
})
|
||||
)
|
||||
|
||||
this.logger.debug(`Verb metadata for ${id} saved successfully`)
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to save verb metadata for ${id}:`, error)
|
||||
throw new Error(`Failed to save verb metadata for ${id}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get verb metadata from storage
|
||||
*/
|
||||
public async getVerbMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Import the GetObjectCommand only when needed
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
const key = `${this.verbMetadataPrefix}${id}.json`
|
||||
this.logger.trace(`Getting verb metadata for ${id} from key: ${key}`)
|
||||
|
||||
// Try to get the verb metadata
|
||||
const response = await this.s3Client!.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: key
|
||||
})
|
||||
)
|
||||
|
||||
// Check if response is null or undefined
|
||||
if (!response || !response.Body) {
|
||||
this.logger.trace(`No verb metadata found for ${id}`)
|
||||
return null
|
||||
}
|
||||
|
||||
// Convert the response body to a string
|
||||
const bodyContents = await response.Body.transformToString()
|
||||
this.logger.trace(`Retrieved verb metadata body for ${id}`)
|
||||
|
||||
// Parse the JSON string
|
||||
return this.operationExecutors.executeGet(async () => {
|
||||
try {
|
||||
const parsedMetadata = JSON.parse(bodyContents)
|
||||
this.logger.trace(`Successfully retrieved verb metadata for ${id}`)
|
||||
return parsedMetadata
|
||||
} catch (parseError) {
|
||||
this.logger.error(`Failed to parse verb metadata for ${id}:`, parseError)
|
||||
return null
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
this.logger.trace(`Reading object from path: ${path}`)
|
||||
|
||||
const response = await this.s3Client!.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: path
|
||||
})
|
||||
)
|
||||
|
||||
if (!response || !response.Body) {
|
||||
this.logger.trace(`Object not found at ${path}`)
|
||||
return null
|
||||
}
|
||||
|
||||
const bodyContents = await response.Body.transformToString()
|
||||
const data = JSON.parse(bodyContents)
|
||||
this.logger.trace(`Object read successfully from ${path}`)
|
||||
return data
|
||||
} catch (error: any) {
|
||||
// 404 errors return null (object doesn't exist)
|
||||
if (
|
||||
error.name === 'NoSuchKey' ||
|
||||
(error.message &&
|
||||
(error.message.includes('NoSuchKey') ||
|
||||
error.message.includes('not found') ||
|
||||
error.message.includes('does not exist')))
|
||||
) {
|
||||
this.logger.trace(`Object not found at ${path}`)
|
||||
return null
|
||||
}
|
||||
|
||||
throw BrainyError.fromError(error, `readObjectFromPath(${path})`)
|
||||
}
|
||||
}, `readObjectFromPath(${path})`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Primitive operation: Delete object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async deleteObjectFromPath(path: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const { DeleteObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
this.logger.trace(`Deleting object at path: ${path}`)
|
||||
|
||||
await this.s3Client!.send(
|
||||
new DeleteObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: path
|
||||
})
|
||||
)
|
||||
|
||||
this.logger.trace(`Object deleted successfully from ${path}`)
|
||||
} catch (error: any) {
|
||||
// Check if this is a "NoSuchKey" error (object doesn't exist)
|
||||
// 404 errors are ok (already deleted)
|
||||
if (
|
||||
error.name === 'NoSuchKey' ||
|
||||
(error.message &&
|
||||
|
|
@ -2079,46 +2047,48 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
error.message.includes('not found') ||
|
||||
error.message.includes('does not exist')))
|
||||
) {
|
||||
this.logger.trace(`Verb metadata not found for ${id}`)
|
||||
return null
|
||||
this.logger.trace(`Object at ${path} not found (already deleted)`)
|
||||
return
|
||||
}
|
||||
|
||||
// For other types of errors, convert to BrainyError for better classification
|
||||
throw BrainyError.fromError(error, `getVerbMetadata(${id})`)
|
||||
this.logger.error(`Failed to delete object from ${path}:`, error)
|
||||
throw new Error(`Failed to delete object from ${path}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Save noun metadata to storage
|
||||
* Primitive operation: List objects under path prefix
|
||||
* All metadata operations use this internally via base class routing
|
||||
*/
|
||||
protected async saveNounMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
protected async listObjectsUnderPath(prefix: string): Promise<string[]> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Import the PutObjectCommand only when needed
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
const { ListObjectsV2Command } = await import('@aws-sdk/client-s3')
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with noun vectors)
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
const key = `${this.metadataPrefix}${shardId}/${id}.json`
|
||||
const body = JSON.stringify(metadata, null, 2)
|
||||
this.logger.trace(`Listing objects under prefix: ${prefix}`)
|
||||
|
||||
this.logger.trace(`Saving noun metadata for ${id} to key: ${key}`)
|
||||
|
||||
// Save the noun metadata to S3-compatible storage
|
||||
const result = await this.s3Client!.send(
|
||||
new PutObjectCommand({
|
||||
const response = await this.s3Client!.send(
|
||||
new ListObjectsV2Command({
|
||||
Bucket: this.bucketName,
|
||||
Key: key,
|
||||
Body: body,
|
||||
ContentType: 'application/json'
|
||||
Prefix: prefix
|
||||
})
|
||||
)
|
||||
|
||||
this.logger.debug(`Noun metadata for ${id} saved successfully`)
|
||||
if (!response || !response.Contents || response.Contents.length === 0) {
|
||||
this.logger.trace(`No objects found under ${prefix}`)
|
||||
return []
|
||||
}
|
||||
|
||||
const paths = response.Contents
|
||||
.map((object: any) => object.Key)
|
||||
.filter((key: string | undefined) => key && key.length > 0) as string[]
|
||||
|
||||
this.logger.trace(`Found ${paths.length} objects under ${prefix}`)
|
||||
return paths
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to save noun metadata for ${id}:`, error)
|
||||
throw new Error(`Failed to save noun metadata for ${id}: ${error}`)
|
||||
this.logger.error(`Failed to list objects under ${prefix}:`, error)
|
||||
throw new Error(`Failed to list objects under ${prefix}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2237,131 +2207,6 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
return results
|
||||
}
|
||||
|
||||
/**
|
||||
* Get noun metadata from storage
|
||||
*/
|
||||
public async getNounMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
// Import the GetObjectCommand only when needed
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
// Use UUID-based sharding for metadata (consistent with noun vectors)
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
const key = `${this.metadataPrefix}${shardId}/${id}.json`
|
||||
this.logger.trace(`Getting noun metadata for ${id} from key: ${key}`)
|
||||
|
||||
// Try to get the noun metadata
|
||||
const response = await this.s3Client!.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: key
|
||||
})
|
||||
)
|
||||
|
||||
// Check if response is null or undefined
|
||||
if (!response || !response.Body) {
|
||||
this.logger.trace(`No noun metadata found for ${id}`)
|
||||
return null
|
||||
}
|
||||
|
||||
// Convert the response body to a string
|
||||
const bodyContents = await response.Body.transformToString()
|
||||
this.logger.trace(`Retrieved noun metadata body for ${id}`)
|
||||
|
||||
// Parse the JSON string
|
||||
try {
|
||||
const parsedMetadata = JSON.parse(bodyContents)
|
||||
this.logger.trace(`Successfully retrieved noun metadata for ${id}`)
|
||||
return parsedMetadata
|
||||
} catch (parseError) {
|
||||
this.logger.error(`Failed to parse noun metadata for ${id}:`, parseError)
|
||||
return null
|
||||
}
|
||||
} catch (error: any) {
|
||||
// Check if this is a "NoSuchKey" error (object doesn't exist)
|
||||
if (
|
||||
error.name === 'NoSuchKey' ||
|
||||
(error.message &&
|
||||
(error.message.includes('NoSuchKey') ||
|
||||
error.message.includes('not found') ||
|
||||
error.message.includes('does not exist')))
|
||||
) {
|
||||
this.logger.trace(`Noun metadata not found for ${id}`)
|
||||
return null
|
||||
}
|
||||
|
||||
// For other types of errors, convert to BrainyError for better classification
|
||||
throw BrainyError.fromError(error, `getNounMetadata(${id})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get metadata from storage
|
||||
*/
|
||||
public async getMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
return this.operationExecutors.executeGet(async () => {
|
||||
try {
|
||||
// Import the GetObjectCommand only when needed
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
|
||||
prodLog.debug(`Getting metadata for ${id} from bucket ${this.bucketName}`)
|
||||
const key = `${this.metadataPrefix}${id}.json`
|
||||
prodLog.debug(`Looking for metadata at key: ${key}`)
|
||||
|
||||
// Try to get the metadata from the metadata directory
|
||||
const response = await this.s3Client!.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: key
|
||||
})
|
||||
)
|
||||
|
||||
// Check if response is null or undefined (can happen in mock implementations)
|
||||
if (!response || !response.Body) {
|
||||
prodLog.debug(`No metadata found for ${id}`)
|
||||
return null
|
||||
}
|
||||
|
||||
// Convert the response body to a string
|
||||
const bodyContents = await response.Body.transformToString()
|
||||
prodLog.debug(`Retrieved metadata body: ${bodyContents}`)
|
||||
|
||||
// Parse the JSON string
|
||||
try {
|
||||
const parsedMetadata = JSON.parse(bodyContents)
|
||||
prodLog.debug(
|
||||
`Successfully retrieved metadata for ${id}:`,
|
||||
parsedMetadata
|
||||
)
|
||||
return parsedMetadata
|
||||
} catch (parseError) {
|
||||
prodLog.error(`Failed to parse metadata for ${id}:`, parseError)
|
||||
return null
|
||||
}
|
||||
} catch (error: any) {
|
||||
// Check if this is a "NoSuchKey" error (object doesn't exist)
|
||||
// In AWS SDK, this would be error.name === 'NoSuchKey'
|
||||
// In our mock, we might get different error types
|
||||
if (
|
||||
error.name === 'NoSuchKey' ||
|
||||
(error.message &&
|
||||
(error.message.includes('NoSuchKey') ||
|
||||
error.message.includes('not found') ||
|
||||
error.message.includes('does not exist')))
|
||||
) {
|
||||
prodLog.debug(`Metadata not found for ${id}`)
|
||||
return null
|
||||
}
|
||||
|
||||
// For other types of errors, convert to BrainyError for better classification
|
||||
throw BrainyError.fromError(error, `getMetadata(${id})`)
|
||||
}
|
||||
}, `getMetadata(${id})`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all data from storage
|
||||
|
|
|
|||
|
|
@ -9,6 +9,19 @@ import { GraphVerb, HNSWNoun, HNSWVerb, StatisticsData } from '../coreTypes.js'
|
|||
import { BaseStorageAdapter } from './adapters/baseStorageAdapter.js'
|
||||
import { validateNounType, validateVerbType } from '../utils/typeValidation.js'
|
||||
import { NounType, VerbType } from '../types/graphTypes.js'
|
||||
import { getShardIdFromUuid } from './sharding.js'
|
||||
|
||||
/**
|
||||
* Storage key analysis result
|
||||
* Used to determine whether a key is a system key or entity key, and its storage path
|
||||
*/
|
||||
interface StorageKeyInfo {
|
||||
original: string
|
||||
isEntity: boolean
|
||||
shardId: string | null
|
||||
directory: string
|
||||
fullPath: string
|
||||
}
|
||||
|
||||
// Common directory/prefix names
|
||||
// Option A: Entity-Based Directory Structure
|
||||
|
|
@ -66,6 +79,76 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
protected graphIndex?: GraphAdjacencyIndex
|
||||
protected readOnly = false
|
||||
|
||||
/**
|
||||
* Analyze a storage key to determine its routing and path
|
||||
* @param id - The key to analyze (UUID or system key)
|
||||
* @param context - The context for the key (noun-metadata, verb-metadata, or system)
|
||||
* @returns Storage key information including path and shard ID
|
||||
* @private
|
||||
*/
|
||||
private analyzeKey(id: string, context: 'noun-metadata' | 'verb-metadata' | 'system'): StorageKeyInfo {
|
||||
// System resource detection
|
||||
const isSystemKey =
|
||||
id.startsWith('__metadata_') ||
|
||||
id.startsWith('__index_') ||
|
||||
id.startsWith('__system_') ||
|
||||
id.startsWith('statistics_') ||
|
||||
id === 'statistics'
|
||||
|
||||
if (isSystemKey) {
|
||||
return {
|
||||
original: id,
|
||||
isEntity: false,
|
||||
shardId: null,
|
||||
directory: SYSTEM_DIR,
|
||||
fullPath: `${SYSTEM_DIR}/${id}.json`
|
||||
}
|
||||
}
|
||||
|
||||
// UUID validation for entity keys
|
||||
const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
|
||||
if (!uuidRegex.test(id)) {
|
||||
console.warn(`[Storage] Unknown key format: ${id} - treating as system resource`)
|
||||
return {
|
||||
original: id,
|
||||
isEntity: false,
|
||||
shardId: null,
|
||||
directory: SYSTEM_DIR,
|
||||
fullPath: `${SYSTEM_DIR}/${id}.json`
|
||||
}
|
||||
}
|
||||
|
||||
// Valid entity UUID - apply sharding
|
||||
const shardId = getShardIdFromUuid(id)
|
||||
|
||||
if (context === 'noun-metadata') {
|
||||
return {
|
||||
original: id,
|
||||
isEntity: true,
|
||||
shardId,
|
||||
directory: `${NOUNS_METADATA_DIR}/${shardId}`,
|
||||
fullPath: `${NOUNS_METADATA_DIR}/${shardId}/${id}.json`
|
||||
}
|
||||
} else if (context === 'verb-metadata') {
|
||||
return {
|
||||
original: id,
|
||||
isEntity: true,
|
||||
shardId,
|
||||
directory: `${VERBS_METADATA_DIR}/${shardId}`,
|
||||
fullPath: `${VERBS_METADATA_DIR}/${shardId}/${id}.json`
|
||||
}
|
||||
} else {
|
||||
// system context - but UUID format
|
||||
return {
|
||||
original: id,
|
||||
isEntity: false,
|
||||
shardId: null,
|
||||
directory: SYSTEM_DIR,
|
||||
fullPath: `${SYSTEM_DIR}/${id}.json`
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the storage adapter
|
||||
* This method should be implemented by each specific adapter
|
||||
|
|
@ -693,20 +776,63 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
}>
|
||||
|
||||
/**
|
||||
* Save metadata to storage
|
||||
* This method should be implemented by each specific adapter
|
||||
* Write a JSON object to a specific path in storage
|
||||
* This is a primitive operation that all adapters must implement
|
||||
* @param path - Full path including filename (e.g., "_system/statistics.json" or "entities/nouns/metadata/3f/3fa85f64-....json")
|
||||
* @param data - Data to write (will be JSON.stringify'd)
|
||||
* @protected
|
||||
*/
|
||||
public abstract saveMetadata(id: string, metadata: any): Promise<void>
|
||||
protected abstract writeObjectToPath(path: string, data: any): Promise<void>
|
||||
|
||||
/**
|
||||
* Read a JSON object from a specific path in storage
|
||||
* This is a primitive operation that all adapters must implement
|
||||
* @param path - Full path including filename
|
||||
* @returns The parsed JSON object, or null if not found
|
||||
* @protected
|
||||
*/
|
||||
protected abstract readObjectFromPath(path: string): Promise<any | null>
|
||||
|
||||
/**
|
||||
* Delete an object from a specific path in storage
|
||||
* This is a primitive operation that all adapters must implement
|
||||
* @param path - Full path including filename
|
||||
* @protected
|
||||
*/
|
||||
protected abstract deleteObjectFromPath(path: string): Promise<void>
|
||||
|
||||
/**
|
||||
* List all object paths under a given prefix
|
||||
* This is a primitive operation that all adapters must implement
|
||||
* @param prefix - Directory prefix to list (e.g., "entities/nouns/metadata/3f/")
|
||||
* @returns Array of full paths
|
||||
* @protected
|
||||
*/
|
||||
protected abstract listObjectsUnderPath(prefix: string): Promise<string[]>
|
||||
|
||||
/**
|
||||
* Save metadata to storage
|
||||
* Routes to correct location (system or entity) based on key format
|
||||
*/
|
||||
public async saveMetadata(id: string, metadata: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
const keyInfo = this.analyzeKey(id, 'system')
|
||||
return this.writeObjectToPath(keyInfo.fullPath, metadata)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get metadata from storage
|
||||
* This method should be implemented by each specific adapter
|
||||
* Routes to correct location (system or entity) based on key format
|
||||
*/
|
||||
public abstract getMetadata(id: string): Promise<any | null>
|
||||
public async getMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
const keyInfo = this.analyzeKey(id, 'system')
|
||||
return this.readObjectFromPath(keyInfo.fullPath)
|
||||
}
|
||||
|
||||
/**
|
||||
* Save noun metadata to storage
|
||||
* This method should be implemented by each specific adapter
|
||||
* Routes to correct sharded location based on UUID
|
||||
*/
|
||||
public async saveNounMetadata(id: string, metadata: any): Promise<void> {
|
||||
// Validate noun type in metadata - storage boundary protection
|
||||
|
|
@ -718,19 +844,28 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
|
||||
/**
|
||||
* Internal method for saving noun metadata
|
||||
* This method should be implemented by each specific adapter
|
||||
* Uses routing logic to handle both UUIDs (sharded) and system keys (unsharded)
|
||||
* @protected
|
||||
*/
|
||||
protected abstract saveNounMetadata_internal(id: string, metadata: any): Promise<void>
|
||||
protected async saveNounMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
const keyInfo = this.analyzeKey(id, 'noun-metadata')
|
||||
return this.writeObjectToPath(keyInfo.fullPath, metadata)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get noun metadata from storage
|
||||
* This method should be implemented by each specific adapter
|
||||
* Uses routing logic to handle both UUIDs (sharded) and system keys (unsharded)
|
||||
*/
|
||||
public abstract getNounMetadata(id: string): Promise<any | null>
|
||||
public async getNounMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
const keyInfo = this.analyzeKey(id, 'noun-metadata')
|
||||
return this.readObjectFromPath(keyInfo.fullPath)
|
||||
}
|
||||
|
||||
/**
|
||||
* Save verb metadata to storage
|
||||
* This method should be implemented by each specific adapter
|
||||
* Routes to correct sharded location based on UUID
|
||||
*/
|
||||
public async saveVerbMetadata(id: string, metadata: any): Promise<void> {
|
||||
// Validate verb type in metadata - storage boundary protection
|
||||
|
|
@ -742,15 +877,24 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
|
||||
/**
|
||||
* Internal method for saving verb metadata
|
||||
* This method should be implemented by each specific adapter
|
||||
* Uses routing logic to handle both UUIDs (sharded) and system keys (unsharded)
|
||||
* @protected
|
||||
*/
|
||||
protected abstract saveVerbMetadata_internal(id: string, metadata: any): Promise<void>
|
||||
protected async saveVerbMetadata_internal(id: string, metadata: any): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
const keyInfo = this.analyzeKey(id, 'verb-metadata')
|
||||
return this.writeObjectToPath(keyInfo.fullPath, metadata)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get verb metadata from storage
|
||||
* This method should be implemented by each specific adapter
|
||||
* Uses routing logic to handle both UUIDs (sharded) and system keys (unsharded)
|
||||
*/
|
||||
public abstract getVerbMetadata(id: string): Promise<any | null>
|
||||
public async getVerbMetadata(id: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
const keyInfo = this.analyzeKey(id, 'verb-metadata')
|
||||
return this.readObjectFromPath(keyInfo.fullPath)
|
||||
}
|
||||
|
||||
/**
|
||||
* Save a noun to storage
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue