soulcraft/brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

Merge remote-tracking branch 'origin/main'

Browse source 
# Conflicts:
#	CHANGES.md
#	package-lock.json
#	package.json
#	src/storage/adapters/fileSystemStorage.ts
This commit is contained in:
David Snelling 2025-07-28 10:04:58 -07:00
commit 3337c9f78e
49 changed files with 5099 additions and 4391 deletions
Download patch file Download diff file Expand all files Collapse all files

320
src/storage/adapters/baseStorageAdapter.ts Normal file
View file

@ -0,0 +1,320 @@
/**
* Base Storage Adapter
* Provides common functionality for all storage adapters, including statistics tracking
*/
import { StatisticsData, StorageAdapter } from '../../coreTypes.js'
/**
* Base class for storage adapters that implements statistics tracking
*/
export abstract class BaseStorageAdapter implements StorageAdapter {
// Abstract methods that must be implemented by subclasses
abstract init(): Promise<void>
abstract saveNoun(noun: any): Promise<void>
abstract getNoun(id: string): Promise<any | null>
abstract getAllNouns(): Promise<any[]>
abstract getNounsByNounType(nounType: string): Promise<any[]>
abstract deleteNoun(id: string): Promise<void>
abstract saveVerb(verb: any): Promise<void>
abstract getVerb(id: string): Promise<any | null>
abstract getAllVerbs(): Promise<any[]>
abstract getVerbsBySource(sourceId: string): Promise<any[]>
abstract getVerbsByTarget(targetId: string): Promise<any[]>
abstract getVerbsByType(type: string): Promise<any[]>
abstract deleteVerb(id: string): Promise<void>
abstract saveMetadata(id: string, metadata: any): Promise<void>
abstract getMetadata(id: string): Promise<any | null>
abstract clear(): Promise<void>
abstract getStorageStatus(): Promise<{
type: string
used: number
quota: number | null
details?: Record<string, any>
}>
// Statistics cache
protected statisticsCache: StatisticsData | null = null
// Batch update timer ID
protected statisticsBatchUpdateTimerId: NodeJS.Timeout | null = null
// Flag to indicate if statistics have been modified since last save
protected statisticsModified = false
// Time of last statistics flush to storage
protected lastStatisticsFlushTime = 0
// Minimum time between statistics flushes (5 seconds)
protected readonly MIN_FLUSH_INTERVAL_MS = 5000
// Maximum time to wait before flushing statistics (30 seconds)
protected readonly MAX_FLUSH_DELAY_MS = 30000
// Statistics-specific methods that must be implemented by subclasses
protected abstract saveStatisticsData(statistics: StatisticsData): Promise<void>
protected abstract getStatisticsData(): Promise<StatisticsData | null>
/**
* Save statistics data
* @param statistics The statistics data to save
*/
async saveStatistics(statistics: StatisticsData): Promise<void> {
// Update the cache with a deep copy to avoid reference issues
this.statisticsCache = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated
}
// Schedule a batch update instead of saving immediately
this.scheduleBatchUpdate()
}
/**
* Get statistics data
* @returns Promise that resolves to the statistics data
*/
async getStatistics(): Promise<StatisticsData | null> {
// If we have cached statistics, return a deep copy
if (this.statisticsCache) {
return {
nounCount: {...this.statisticsCache.nounCount},
verbCount: {...this.statisticsCache.verbCount},
metadataCount: {...this.statisticsCache.metadataCount},
hnswIndexSize: this.statisticsCache.hnswIndexSize,
lastUpdated: this.statisticsCache.lastUpdated
}
}
// Otherwise, get from storage
const statistics = await this.getStatisticsData()
// If we found statistics, update the cache
if (statistics) {
// Update the cache with a deep copy
this.statisticsCache = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated
}
}
return statistics
}
/**
* Schedule a batch update of statistics
*/
protected scheduleBatchUpdate(): void {
// Mark statistics as modified
this.statisticsModified = true
// If a timer is already set, don't set another one
if (this.statisticsBatchUpdateTimerId !== null) {
return
}
// Calculate time since last flush
const now = Date.now()
const timeSinceLastFlush = now - this.lastStatisticsFlushTime
// If we've recently flushed, wait longer before the next flush
const delayMs = timeSinceLastFlush < this.MIN_FLUSH_INTERVAL_MS
? this.MAX_FLUSH_DELAY_MS
: this.MIN_FLUSH_INTERVAL_MS
// Schedule the batch update
this.statisticsBatchUpdateTimerId = setTimeout(() => {
this.flushStatistics()
}, delayMs)
}
/**
* Flush statistics to storage
*/
protected async flushStatistics(): Promise<void> {
// Clear the timer
if (this.statisticsBatchUpdateTimerId !== null) {
clearTimeout(this.statisticsBatchUpdateTimerId)
this.statisticsBatchUpdateTimerId = null
}
// If statistics haven't been modified, no need to flush
if (!this.statisticsModified || !this.statisticsCache) {
return
}
try {
// Save the statistics to storage
await this.saveStatisticsData(this.statisticsCache)
// Update the last flush time
this.lastStatisticsFlushTime = Date.now()
// Reset the modified flag
this.statisticsModified = false
} catch (error) {
console.error('Failed to flush statistics data:', error)
// Mark as still modified so we'll try again later
this.statisticsModified = true
// Don't throw the error to avoid disrupting the application
}
}
/**
* Increment a statistic counter
* @param type The type of statistic to increment ('noun', 'verb', 'metadata')
* @param service The service that inserted the data
* @param amount The amount to increment by (default: 1)
*/
async incrementStatistic(
type: 'noun' | 'verb' | 'metadata',
service: string,
amount: number = 1
): Promise<void> {
// Get current statistics from cache or storage
let statistics = this.statisticsCache
if (!statistics) {
statistics = await this.getStatisticsData()
if (!statistics) {
statistics = this.createDefaultStatistics()
}
// Update the cache
this.statisticsCache = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated
}
}
// Increment the appropriate counter
const counterMap = {
noun: this.statisticsCache!.nounCount,
verb: this.statisticsCache!.verbCount,
metadata: this.statisticsCache!.metadataCount
}
const counter = counterMap[type]
counter[service] = (counter[service] || 0) + amount
// Update timestamp
this.statisticsCache!.lastUpdated = new Date().toISOString()
// Schedule a batch update instead of saving immediately
this.scheduleBatchUpdate()
}
/**
* Decrement a statistic counter
* @param type The type of statistic to decrement ('noun', 'verb', 'metadata')
* @param service The service that inserted the data
* @param amount The amount to decrement by (default: 1)
*/
async decrementStatistic(
type: 'noun' | 'verb' | 'metadata',
service: string,
amount: number = 1
): Promise<void> {
// Get current statistics from cache or storage
let statistics = this.statisticsCache
if (!statistics) {
statistics = await this.getStatisticsData()
if (!statistics) {
statistics = this.createDefaultStatistics()
}
// Update the cache
this.statisticsCache = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated
}
}
// Decrement the appropriate counter
const counterMap = {
noun: this.statisticsCache!.nounCount,
verb: this.statisticsCache!.verbCount,
metadata: this.statisticsCache!.metadataCount
}
const counter = counterMap[type]
counter[service] = Math.max(0, (counter[service] || 0) - amount)
// Update timestamp
this.statisticsCache!.lastUpdated = new Date().toISOString()
// Schedule a batch update instead of saving immediately
this.scheduleBatchUpdate()
}
/**
* Update the HNSW index size statistic
* @param size The new size of the HNSW index
*/
async updateHnswIndexSize(size: number): Promise<void> {
// Get current statistics from cache or storage
let statistics = this.statisticsCache
if (!statistics) {
statistics = await this.getStatisticsData()
if (!statistics) {
statistics = this.createDefaultStatistics()
}
// Update the cache
this.statisticsCache = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated
}
}
// Update HNSW index size
this.statisticsCache!.hnswIndexSize = size
// Update timestamp
this.statisticsCache!.lastUpdated = new Date().toISOString()
// Schedule a batch update instead of saving immediately
this.scheduleBatchUpdate()
}
/**
* Force an immediate flush of statistics to storage
* This ensures that any pending statistics updates are written to persistent storage
*/
async flushStatisticsToStorage(): Promise<void> {
// If there are no statistics in cache or they haven't been modified, nothing to flush
if (!this.statisticsCache || !this.statisticsModified) {
return
}
// Call the protected flushStatistics method to immediately write to storage
await this.flushStatistics()
}
/**
* Create default statistics data
* @returns Default statistics data
*/
protected createDefaultStatistics(): StatisticsData {
return {
nounCount: {},
verbCount: {},
metadataCount: {},
hnswIndexSize: 0,
lastUpdated: new Date().toISOString()
}
}
}

317
src/storage/adapters/memoryStorage.ts
View file

@ -3,18 +3,10 @@
* In-memory storage adapter for environments where persistent storage is not available or needed
*/
import { GraphVerb, HNSWNoun } from '../../coreTypes.js'
import { BaseStorage } from '../baseStorage.js'
import { GraphVerb, HNSWNoun, StatisticsData } from '../../coreTypes.js'
import { BaseStorage, STATISTICS_KEY } from '../baseStorage.js'
/**
* Type alias for HNSWNoun to make the code more readable
*/
type HNSWNode = HNSWNoun
/**
* Type alias for GraphVerb to make the code more readable
*/
type Edge = GraphVerb
// No type aliases needed - using the original types directly
/**
* In-memory storage adapter
@ -22,9 +14,10 @@ type Edge = GraphVerb
*/
export class MemoryStorage extends BaseStorage {
// Single map of noun ID to noun
private nouns: Map<string, HNSWNode> = new Map()
private verbs: Map<string, Edge> = new Map()
private nouns: Map<string, HNSWNoun> = new Map()
private verbs: Map<string, GraphVerb> = new Map()
private metadata: Map<string, any> = new Map()
private statistics: StatisticsData | null = null
constructor() {
super()
@ -39,237 +32,270 @@ export class MemoryStorage extends BaseStorage {
}
/**
* Save a node to storage
* Save a noun to storage
*/
protected async saveNode(node: HNSWNode): Promise<void> {
protected async saveNoun_internal(noun: HNSWNoun): Promise<void> {
// Create a deep copy to avoid reference issues
const nodeCopy: HNSWNode = {
id: node.id,
vector: [...node.vector],
const nounCopy: HNSWNoun = {
id: noun.id,
vector: [...noun.vector],
connections: new Map()
}
// Copy connections
for (const [level, connections] of node.connections.entries()) {
nodeCopy.connections.set(level, new Set(connections))
for (const [level, connections] of noun.connections.entries()) {
nounCopy.connections.set(level, new Set(connections))
}
// Save the node directly in the nouns map
this.nouns.set(node.id, nodeCopy)
// Save the noun directly in the nouns map
this.nouns.set(noun.id, nounCopy)
}
/**
* Get a node from storage
* Get a noun from storage
*/
protected async getNode(id: string): Promise<HNSWNode | null> {
// Get the node directly from the nouns map
const node = this.nouns.get(id)
protected async getNoun_internal(id: string): Promise<HNSWNoun | null> {
// Get the noun directly from the nouns map
const noun = this.nouns.get(id)
// If not found, return null
if (!node) {
if (!noun) {
return null
}
// Return a deep copy to avoid reference issues
const nodeCopy: HNSWNode = {
id: node.id,
vector: [...node.vector],
const nounCopy: HNSWNoun = {
id: noun.id,
vector: [...noun.vector],
connections: new Map()
}
// Copy connections
for (const [level, connections] of node.connections.entries()) {
nodeCopy.connections.set(level, new Set(connections))
for (const [level, connections] of noun.connections.entries()) {
nounCopy.connections.set(level, new Set(connections))
}
return nodeCopy
return nounCopy
}
/**
* Get all nodes from storage
* Get all nouns from storage
*/
protected async getAllNodes(): Promise<HNSWNode[]> {
const allNodes: HNSWNode[] = []
protected async getAllNouns_internal(): Promise<HNSWNoun[]> {
const allNouns: HNSWNoun[] = []
// Iterate through all nodes in the nouns map
for (const [nodeId, node] of this.nouns.entries()) {
// Iterate through all nouns in the nouns map
for (const [nounId, noun] of this.nouns.entries()) {
// Return a deep copy to avoid reference issues
const nodeCopy: HNSWNode = {
id: node.id,
vector: [...node.vector],
const nounCopy: HNSWNoun = {
id: noun.id,
vector: [...noun.vector],
connections: new Map()
}
// Copy connections
for (const [level, connections] of node.connections.entries()) {
nodeCopy.connections.set(level, new Set(connections))
for (const [level, connections] of noun.connections.entries()) {
nounCopy.connections.set(level, new Set(connections))
}
allNodes.push(nodeCopy)
allNouns.push(nounCopy)
}
return allNodes
return allNouns
}
/**
* Get nodes by noun type
* Get nouns by noun type
* @param nounType The noun type to filter by
* @returns Promise that resolves to an array of nodes of the specified noun type
* @returns Promise that resolves to an array of nouns of the specified noun type
*/
protected async getNodesByNounType(nounType: string): Promise<HNSWNode[]> {
const nodes: HNSWNode[] = []
protected async getNounsByNounType_internal(nounType: string): Promise<HNSWNoun[]> {
const nouns: HNSWNoun[] = []
// Iterate through all nodes and filter by noun type using metadata
for (const [nodeId, node] of this.nouns.entries()) {
// Iterate through all nouns and filter by noun type using metadata
for (const [nounId, noun] of this.nouns.entries()) {
// Get the metadata to check the noun type
const metadata = await this.getMetadata(nodeId)
const metadata = await this.getMetadata(nounId)
// Include the node if its noun type matches the requested type
// Include the noun if its noun type matches the requested type
if (metadata && metadata.noun === nounType) {
// Return a deep copy to avoid reference issues
const nodeCopy: HNSWNode = {
id: node.id,
vector: [...node.vector],
const nounCopy: HNSWNoun = {
id: noun.id,
vector: [...noun.vector],
connections: new Map()
}
// Copy connections
for (const [level, connections] of node.connections.entries()) {
nodeCopy.connections.set(level, new Set(connections))
for (const [level, connections] of noun.connections.entries()) {
nounCopy.connections.set(level, new Set(connections))
}
nodes.push(nodeCopy)
nouns.push(nounCopy)
}
}
return nodes
return nouns
}
/**
* Delete a node from storage
* Delete a noun from storage
*/
protected async deleteNode(id: string): Promise<void> {
// Delete the node directly from the nouns map
protected async deleteNoun_internal(id: string): Promise<void> {
this.nouns.delete(id)
}
/**
* Save an edge to storage
* Save a verb to storage
*/
protected async saveEdge(edge: Edge): Promise<void> {
protected async saveVerb_internal(verb: GraphVerb): Promise<void> {
// Create a deep copy to avoid reference issues
const edgeCopy: Edge = {
id: edge.id,
vector: [...edge.vector],
const verbCopy: GraphVerb = {
id: verb.id,
vector: [...verb.vector],
connections: new Map(),
sourceId: edge.sourceId,
targetId: edge.targetId,
type: edge.type,
weight: edge.weight,
metadata: edge.metadata
sourceId: verb.sourceId,
targetId: verb.targetId,
type: verb.type,
weight: verb.weight,
metadata: verb.metadata
}
// Copy connections
for (const [level, connections] of edge.connections.entries()) {
edgeCopy.connections.set(level, new Set(connections))
for (const [level, connections] of verb.connections.entries()) {
verbCopy.connections.set(level, new Set(connections))
}
// Save the edge directly in the verbs map
this.verbs.set(edge.id, edgeCopy)
// Save the verb directly in the verbs map
this.verbs.set(verb.id, verbCopy)
}
/**
* Get an edge from storage
* Get a verb from storage
*/
protected async getEdge(id: string): Promise<Edge | null> {
// Get the edge directly from the verbs map
const edge = this.verbs.get(id)
protected async getVerb_internal(id: string): Promise<GraphVerb | null> {
// Get the verb directly from the verbs map
const verb = this.verbs.get(id)
// If not found, return null
if (!edge) {
if (!verb) {
return null
}
// Create default timestamp if not present
const defaultTimestamp = {
seconds: Math.floor(Date.now() / 1000),
nanoseconds: (Date.now() % 1000) * 1000000
}
// Create default createdBy if not present
const defaultCreatedBy = {
augmentation: 'unknown',
version: '1.0'
}
// Return a deep copy to avoid reference issues
const edgeCopy: Edge = {
id: edge.id,
vector: [...edge.vector],
const verbCopy: GraphVerb = {
id: verb.id,
vector: [...verb.vector],
connections: new Map(),
sourceId: edge.sourceId,
targetId: edge.targetId,
type: edge.type,
weight: edge.weight,
metadata: edge.metadata
sourceId: (verb.sourceId || verb.source || ""),
targetId: (verb.targetId || verb.target || ""),
source: (verb.sourceId || verb.source || ""),
target: (verb.targetId || verb.target || ""),
verb: verb.type || verb.verb,
weight: verb.weight,
metadata: verb.metadata,
createdAt: verb.createdAt || defaultTimestamp,
updatedAt: verb.updatedAt || defaultTimestamp,
createdBy: verb.createdBy || defaultCreatedBy
}
// Copy connections
for (const [level, connections] of edge.connections.entries()) {
edgeCopy.connections.set(level, new Set(connections))
for (const [level, connections] of verb.connections.entries()) {
verbCopy.connections.set(level, new Set(connections))
}
return edgeCopy
return verbCopy
}
/**
* Get all edges from storage
* Get all verbs from storage
*/
protected async getAllEdges(): Promise<Edge[]> {
const allEdges: Edge[] = []
protected async getAllVerbs_internal(): Promise<GraphVerb[]> {
const allVerbs: GraphVerb[] = []
// Iterate through all verbs in the verbs map
for (const [verbId, verb] of this.verbs.entries()) {
// Create default timestamp if not present
const defaultTimestamp = {
seconds: Math.floor(Date.now() / 1000),
nanoseconds: (Date.now() % 1000) * 1000000
}
// Create default createdBy if not present
const defaultCreatedBy = {
augmentation: 'unknown',
version: '1.0'
}
// Iterate through all edges in the verbs map
for (const [edgeId, edge] of this.verbs.entries()) {
// Return a deep copy to avoid reference issues
const edgeCopy: Edge = {
id: edge.id,
vector: [...edge.vector],
const verbCopy: GraphVerb = {
id: verb.id,
vector: [...verb.vector],
connections: new Map(),
sourceId: edge.sourceId,
targetId: edge.targetId,
type: edge.type,
weight: edge.weight,
metadata: edge.metadata
sourceId: (verb.sourceId || verb.source || ""),
targetId: (verb.targetId || verb.target || ""),
source: (verb.sourceId || verb.source || ""),
target: (verb.targetId || verb.target || ""),
verb: verb.type || verb.verb,
weight: verb.weight,
metadata: verb.metadata,
createdAt: verb.createdAt || defaultTimestamp,
updatedAt: verb.updatedAt || defaultTimestamp,
createdBy: verb.createdBy || defaultCreatedBy
}
// Copy connections
for (const [level, connections] of edge.connections.entries()) {
edgeCopy.connections.set(level, new Set(connections))
for (const [level, connections] of verb.connections.entries()) {
verbCopy.connections.set(level, new Set(connections))
}
allEdges.push(edgeCopy)
allVerbs.push(verbCopy)
}
return allEdges
return allVerbs
}
/**
* Get edges by source
* Get verbs by source
*/
protected async getEdgesBySource(sourceId: string): Promise<Edge[]> {
const edges = await this.getAllEdges()
return edges.filter((edge) => edge.sourceId === sourceId)
protected async getVerbsBySource_internal(sourceId: string): Promise<GraphVerb[]> {
const allVerbs = await this.getAllVerbs_internal()
return allVerbs.filter((verb: GraphVerb) => (verb.sourceId || verb.source) === sourceId)
}
/**
* Get edges by target
* Get verbs by target
*/
protected async getEdgesByTarget(targetId: string): Promise<Edge[]> {
const edges = await this.getAllEdges()
return edges.filter((edge) => edge.targetId === targetId)
protected async getVerbsByTarget_internal(targetId: string): Promise<GraphVerb[]> {
const allVerbs = await this.getAllVerbs_internal()
return allVerbs.filter((verb: GraphVerb) => (verb.targetId || verb.target) === targetId)
}
/**
* Get edges by type
* Get verbs by type
*/
protected async getEdgesByType(type: string): Promise<Edge[]> {
const edges = await this.getAllEdges()
return edges.filter((edge) => edge.type === type)
protected async getVerbsByType_internal(type: string): Promise<GraphVerb[]> {
const allVerbs = await this.getAllVerbs_internal()
return allVerbs.filter((verb: GraphVerb) => (verb.type || verb.verb) === type)
}
/**
* Delete an edge from storage
* Delete a verb from storage
*/
protected async deleteEdge(id: string): Promise<void> {
// Delete the edge directly from the verbs map
protected async deleteVerb_internal(id: string): Promise<void> {
// Delete the verb directly from the verbs map
this.verbs.delete(id)
}
@ -321,4 +347,45 @@ export class MemoryStorage extends BaseStorage {
}
}
}
/**
* Save statistics data to storage
* @param statistics The statistics data to save
*/
protected async saveStatisticsData(statistics: StatisticsData): Promise<void> {
// For memory storage, we just need to store the statistics in memory
// Create a deep copy to avoid reference issues
this.statistics = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated
}
// Since this is in-memory, there's no need for time-based partitioning
// or legacy file handling
}
/**
* Get statistics data from storage
* @returns Promise that resolves to the statistics data or null if not found
*/
protected async getStatisticsData(): Promise<StatisticsData | null> {
if (!this.statistics) {
return null
}
// Return a deep copy to avoid reference issues
return {
nounCount: {...this.statistics.nounCount},
verbCount: {...this.statistics.verbCount},
metadataCount: {...this.statistics.metadataCount},
hnswIndexSize: this.statistics.hnswIndexSize,
lastUpdated: this.statistics.lastUpdated
}
// Since this is in-memory, there's no need for fallback mechanisms
// to check multiple storage locations
}
}

374
src/storage/adapters/opfsStorage.ts
View file

@ -3,10 +3,18 @@
* Provides persistent storage for the vector database using the Origin Private File System API
*/
import {GraphVerb, HNSWNoun} from '../../coreTypes.js'
import {BaseStorage, NOUNS_DIR, VERBS_DIR, METADATA_DIR, INDEX_DIR} from '../baseStorage.js'
import {GraphVerb, HNSWNoun, StatisticsData} from '../../coreTypes.js'
import {BaseStorage, NOUNS_DIR, VERBS_DIR, METADATA_DIR, INDEX_DIR, STATISTICS_KEY} from '../baseStorage.js'
import '../../types/fileSystemTypes.js'
// Type alias for HNSWNode
type HNSWNode = HNSWNoun
/**
* Type alias for GraphVerb to make the code more readable
*/
type Edge = GraphVerb
/**
* Helper function to safely get a file from a FileSystemHandle
* This is needed because TypeScript doesn't recognize that a FileSystemHandle
@ -18,8 +26,8 @@ async function safeGetFile(handle: FileSystemHandle): Promise<File> {
}
// Type aliases for better readability
type HNSWNode = HNSWNoun
type Edge = GraphVerb
type HNSWNoun_internal = HNSWNoun
type Verb = GraphVerb
// Root directory name for OPFS storage
const ROOT_DIR = 'opfs-vector-db'
@ -37,6 +45,7 @@ export class OPFSStorage extends BaseStorage {
private isAvailable = false
private isPersistentRequested = false
private isPersistentGranted = false
private statistics: StatisticsData | null = null
constructor() {
super()
@ -148,54 +157,54 @@ export class OPFSStorage extends BaseStorage {
}
/**
* Save a node to storage
* Save a noun to storage
*/
protected async saveNode(node: HNSWNode): Promise<void> {
protected async saveNoun_internal(noun: HNSWNoun_internal): Promise<void> {
await this.ensureInitialized()
try {
// Convert connections Map to a serializable format
const serializableNode = {
...node,
connections: this.mapToObject(node.connections, (set) =>
const serializableNoun = {
...noun,
connections: this.mapToObject(noun.connections, (set) =>
Array.from(set as Set<string>)
)
}
// Create or get the file for this noun
const fileHandle = await this.nounsDir!.getFileHandle(node.id, {
const fileHandle = await this.nounsDir!.getFileHandle(noun.id, {
create: true
})
// Write the noun data to the file
const writable = await fileHandle.createWritable()
await writable.write(JSON.stringify(serializableNode))
await writable.write(JSON.stringify(serializableNoun))
await writable.close()
} catch (error) {
console.error(`Failed to save node ${node.id}:`, error)
throw new Error(`Failed to save node ${node.id}: ${error}`)
console.error(`Failed to save noun ${noun.id}:`, error)
throw new Error(`Failed to save noun ${noun.id}: ${error}`)
}
}
/**
* Get a node from storage
* Get a noun from storage
*/
protected async getNode(id: string): Promise<HNSWNode | null> {
protected async getNoun_internal(id: string): Promise<HNSWNoun_internal | null> {
await this.ensureInitialized()
try {
// Get the file handle for this node
// Get the file handle for this noun
const fileHandle = await this.nounsDir!.getFileHandle(id)
// Read the node data from the file
// Read the noun data from the file
const file = await fileHandle.getFile()
const text = await file.text()
const data = JSON.parse(text)
// Convert serialized connections back to Map<number, Set<string>>
const connections = new Map<number, Set<string>>()
for (const [level, nodeIds] of Object.entries(data.connections)) {
connections.set(Number(level), new Set(nodeIds as string[]))
for (const [level, nounIds] of Object.entries(data.connections)) {
connections.set(Number(level), new Set(nounIds as string[]))
}
return {
@ -204,41 +213,41 @@ export class OPFSStorage extends BaseStorage {
connections
}
} catch (error) {
// Node not found or other error
// Noun not found or other error
return null
}
}
/**
* Get all nodes from storage
* Get all nouns from storage
*/
protected async getAllNodes(): Promise<HNSWNode[]> {
protected async getAllNouns_internal(): Promise<HNSWNoun_internal[]> {
await this.ensureInitialized()
const allNodes: HNSWNode[] = []
const allNouns: HNSWNoun_internal[] = []
try {
// Iterate through all files in the nouns directory
for await (const [name, handle] of this.nounsDir!.entries()) {
if (handle.kind === 'file') {
try {
// Read the node data from the file
// Read the noun data from the file
const file = await safeGetFile(handle)
const text = await file.text()
const data = JSON.parse(text)
// Convert serialized connections back to Map<number, Set<string>>
const connections = new Map<number, Set<string>>()
for (const [level, nodeIds] of Object.entries(data.connections)) {
connections.set(Number(level), new Set(nodeIds as string[]))
for (const [level, nounIds] of Object.entries(data.connections)) {
connections.set(Number(level), new Set(nounIds as string[]))
}
allNodes.push({
allNouns.push({
id: data.id,
vector: data.vector,
connections
})
} catch (error) {
console.error(`Error reading node file ${name}:`, error)
console.error(`Error reading noun file ${name}:`, error)
}
}
}
@ -246,7 +255,16 @@ export class OPFSStorage extends BaseStorage {
console.error('Error reading nouns directory:', error)
}
return allNodes
return allNouns
}
/**
* Get nouns by noun type (internal implementation)
* @param nounType The noun type to filter by
* @returns Promise that resolves to an array of nouns of the specified noun type
*/
protected async getNounsByNounType_internal(nounType: string): Promise<HNSWNoun[]> {
return this.getNodesByNounType(nounType)
}
/**
@ -298,6 +316,13 @@ export class OPFSStorage extends BaseStorage {
return nodes
}
/**
* Delete a noun from storage (internal implementation)
*/
protected async deleteNoun_internal(id: string): Promise<void> {
return this.deleteNode(id)
}
/**
* Delete a node from storage
*/
@ -315,6 +340,13 @@ export class OPFSStorage extends BaseStorage {
}
}
/**
* Save a verb to storage (internal implementation)
*/
protected async saveVerb_internal(verb: GraphVerb): Promise<void> {
return this.saveEdge(verb)
}
/**
* Save an edge to storage
*/
@ -345,6 +377,13 @@ export class OPFSStorage extends BaseStorage {
}
}
/**
* Get a verb from storage (internal implementation)
*/
protected async getVerb_internal(id: string): Promise<GraphVerb | null> {
return this.getEdge(id)
}
/**
* Get an edge from storage
*/
@ -366,15 +405,32 @@ export class OPFSStorage extends BaseStorage {
connections.set(Number(level), new Set(nodeIds as string[]))
}
// Create default timestamp if not present
const defaultTimestamp = {
seconds: Math.floor(Date.now() / 1000),
nanoseconds: (Date.now() % 1000) * 1000000
}
// Create default createdBy if not present
const defaultCreatedBy = {
augmentation: 'unknown',
version: '1.0'
}
return {
id: data.id,
vector: data.vector,
connections,
sourceId: data.sourceId,
targetId: data.targetId,
type: data.type,
sourceId: data.sourceId || data.source,
targetId: data.targetId || data.target,
source: data.sourceId || data.source,
target: data.targetId || data.target,
verb: data.type || data.verb,
weight: data.weight,
metadata: data.metadata
metadata: data.metadata,
createdAt: data.createdAt || defaultTimestamp,
updatedAt: data.updatedAt || defaultTimestamp,
createdBy: data.createdBy || defaultCreatedBy
}
} catch (error) {
// Edge not found or other error
@ -382,6 +438,13 @@ export class OPFSStorage extends BaseStorage {
}
}
/**
* Get all verbs from storage (internal implementation)
*/
protected async getAllVerbs_internal(): Promise<GraphVerb[]> {
return this.getAllEdges()
}
/**
* Get all edges from storage
*/
@ -405,15 +468,32 @@ export class OPFSStorage extends BaseStorage {
connections.set(Number(level), new Set(nodeIds as string[]))
}
// Create default timestamp if not present
const defaultTimestamp = {
seconds: Math.floor(Date.now() / 1000),
nanoseconds: (Date.now() % 1000) * 1000000
}
// Create default createdBy if not present
const defaultCreatedBy = {
augmentation: 'unknown',
version: '1.0'
}
allEdges.push({
id: data.id,
vector: data.vector,
connections,
sourceId: data.sourceId,
targetId: data.targetId,
type: data.type,
sourceId: data.sourceId || data.source,
targetId: data.targetId || data.target,
source: data.sourceId || data.source,
target: data.targetId || data.target,
verb: data.type || data.verb,
weight: data.weight,
metadata: data.metadata
metadata: data.metadata,
createdAt: data.createdAt || defaultTimestamp,
updatedAt: data.updatedAt || defaultTimestamp,
createdBy: data.createdBy || defaultCreatedBy
})
} catch (error) {
console.error(`Error reading edge file ${name}:`, error)
@ -427,12 +507,26 @@ export class OPFSStorage extends BaseStorage {
return allEdges
}
/**
* Get verbs by source (internal implementation)
*/
protected async getVerbsBySource_internal(sourceId: string): Promise<GraphVerb[]> {
return this.getEdgesBySource(sourceId)
}
/**
* Get edges by source
*/
protected async getEdgesBySource(sourceId: string): Promise<Edge[]> {
const edges = await this.getAllEdges()
return edges.filter((edge) => edge.sourceId === sourceId)
return edges.filter((edge) => (edge.sourceId || edge.source) === sourceId)
}
/**
* Get verbs by target (internal implementation)
*/
protected async getVerbsByTarget_internal(targetId: string): Promise<GraphVerb[]> {
return this.getEdgesByTarget(targetId)
}
/**
@ -440,7 +534,14 @@ export class OPFSStorage extends BaseStorage {
*/
protected async getEdgesByTarget(targetId: string): Promise<Edge[]> {
const edges = await this.getAllEdges()
return edges.filter((edge) => edge.targetId === targetId)
return edges.filter((edge) => (edge.targetId || edge.target) === targetId)
}
/**
* Get verbs by type (internal implementation)
*/
protected async getVerbsByType_internal(type: string): Promise<GraphVerb[]> {
return this.getEdgesByType(type)
}
/**
@ -448,7 +549,14 @@ export class OPFSStorage extends BaseStorage {
*/
protected async getEdgesByType(type: string): Promise<Edge[]> {
const edges = await this.getAllEdges()
return edges.filter((edge) => edge.type === type)
return edges.filter((edge) => (edge.type || edge.verb) === type)
}
/**
* Delete a verb from storage (internal implementation)
*/
protected async deleteVerb_internal(id: string): Promise<void> {
return this.deleteEdge(id)
}
/**
@ -688,4 +796,190 @@ export class OPFSStorage extends BaseStorage {
}
}
}
/**
* Get the statistics key for a specific date
* @param date The date to get the key for
* @returns The statistics key for the specified date
*/
private getStatisticsKeyForDate(date: Date): string {
const year = date.getUTCFullYear()
const month = String(date.getUTCMonth() + 1).padStart(2, '0')
const day = String(date.getUTCDate()).padStart(2, '0')
return `statistics_${year}${month}${day}.json`
}
/**
* Get the current statistics key
* @returns The current statistics key
*/
private getCurrentStatisticsKey(): string {
return this.getStatisticsKeyForDate(new Date())
}
/**
* Get the legacy statistics key (for backward compatibility)
* @returns The legacy statistics key
*/
private getLegacyStatisticsKey(): string {
return 'statistics.json'
}
/**
* Save statistics data to storage
* @param statistics The statistics data to save
*/
protected async saveStatisticsData(statistics: StatisticsData): Promise<void> {
// Create a deep copy to avoid reference issues
this.statistics = {
nounCount: {...statistics.nounCount},
verbCount: {...statistics.verbCount},
metadataCount: {...statistics.metadataCount},
hnswIndexSize: statistics.hnswIndexSize,
lastUpdated: statistics.lastUpdated
}
try {
// Ensure the root directory is initialized
await this.ensureInitialized()
// Get or create the index directory
if (!this.indexDir) {
throw new Error('Index directory not initialized')
}
// Get the current statistics key
const currentKey = this.getCurrentStatisticsKey()
// Create a file for the statistics data
const fileHandle = await this.indexDir.getFileHandle(currentKey, {
create: true
})
// Create a writable stream
const writable = await fileHandle.createWritable()
// Write the statistics data to the file
await writable.write(JSON.stringify(this.statistics, null, 2))
// Close the stream
await writable.close()
// Also update the legacy key for backward compatibility, but less frequently
if (Math.random() < 0.1) {
const legacyKey = this.getLegacyStatisticsKey()
const legacyFileHandle = await this.indexDir.getFileHandle(legacyKey, {
create: true
})
const legacyWritable = await legacyFileHandle.createWritable()
await legacyWritable.write(JSON.stringify(this.statistics, null, 2))
await legacyWritable.close()
}
} catch (error) {
console.error('Failed to save statistics data:', error)
throw new Error(`Failed to save statistics data: ${error}`)
}
}
/**
* Get statistics data from storage
* @returns Promise that resolves to the statistics data or null if not found
*/
protected async getStatisticsData(): Promise<StatisticsData | null> {
// If we have cached statistics, return a deep copy
if (this.statistics) {
return {
nounCount: {...this.statistics.nounCount},
verbCount: {...this.statistics.verbCount},
metadataCount: {...this.statistics.metadataCount},
hnswIndexSize: this.statistics.hnswIndexSize,
lastUpdated: this.statistics.lastUpdated
}
}
try {
// Ensure the root directory is initialized
await this.ensureInitialized()
if (!this.indexDir) {
throw new Error('Index directory not initialized')
}
// First try to get statistics from today's file
const currentKey = this.getCurrentStatisticsKey()
try {
const fileHandle = await this.indexDir.getFileHandle(currentKey, {
create: false
})
const file = await fileHandle.getFile()
const text = await file.text()
this.statistics = JSON.parse(text)
if (this.statistics) {
return {
nounCount: {...this.statistics.nounCount},
verbCount: {...this.statistics.verbCount},
metadataCount: {...this.statistics.metadataCount},
hnswIndexSize: this.statistics.hnswIndexSize,
lastUpdated: this.statistics.lastUpdated
}
}
} catch (error) {
// If today's file doesn't exist, try yesterday's file
const yesterday = new Date()
yesterday.setDate(yesterday.getDate() - 1)
const yesterdayKey = this.getStatisticsKeyForDate(yesterday)
try {
const fileHandle = await this.indexDir.getFileHandle(yesterdayKey, {
create: false
})
const file = await fileHandle.getFile()
const text = await file.text()
this.statistics = JSON.parse(text)
if (this.statistics) {
return {
nounCount: {...this.statistics.nounCount},
verbCount: {...this.statistics.verbCount},
metadataCount: {...this.statistics.metadataCount},
hnswIndexSize: this.statistics.hnswIndexSize,
lastUpdated: this.statistics.lastUpdated
}
}
} catch (error) {
// If yesterday's file doesn't exist, try the legacy file
const legacyKey = this.getLegacyStatisticsKey()
try {
const fileHandle = await this.indexDir.getFileHandle(legacyKey, {
create: false
})
const file = await fileHandle.getFile()
const text = await file.text()
this.statistics = JSON.parse(text)
if (this.statistics) {
return {
nounCount: {...this.statistics.nounCount},
verbCount: {...this.statistics.verbCount},
metadataCount: {...this.statistics.metadataCount},
hnswIndexSize: this.statistics.hnswIndexSize,
lastUpdated: this.statistics.lastUpdated
}
}
} catch (error) {
// If the legacy file doesn't exist either, return null
return null
}
}
}
// If we get here and statistics is null, return default statistics
return this.statistics ? this.statistics : null
} catch (error) {
console.error('Failed to get statistics data:', error)
throw new Error(`Failed to get statistics data: ${error}`)
}
}
}

2206
src/storage/adapters/s3CompatibleStorage.ts
View file

File diff suppressed because it is too large Load diff