David Snelling
c0f6ccd958
The 7.x branch system stored Virtual Filesystem content as blobs in its
copy-on-write area (`_cow/`). 8.0 removed that system and stores content blobs
in the content-addressed store (`_cas/`), but the one-time layout migration only
moves entities — it never adopted the `_cow/` content blobs. A store that used
the VFS was left with every VFS-backed read throwing "Blob metadata not found"
and its pages 500ing, with no first-party recovery and the pre-upgrade backup
already removed on entity-migration "success".
Add an on-open recovery pass (`autoAdoptLegacyVfsBlobsIfNeeded`) that runs right
after the layout migration and adopts every orphaned `_cow/` blob into `_cas/` —
copying both the bytes and the metadata via the raw object primitives,
idempotently and non-destructively (the `_cow/` originals are never deleted). It
is a separate phase, not folded into the migration, so it also heals a store
already upgraded by an earlier 8.0.x that stranded the blobs (whose layout-
migration marker is stamped): it is gated on the presence of `_cow/` and its own
`_system/vfs-blob-adoption.json` marker. Filesystem-only; native-8.0 and non-VFS
stores no-op on a cheap existence check.
- `BaseStorage.adoptLegacyCowBlobs()` — the scan/copy primitive, returning
`{ cowBlobs, adopted, alreadyPresent, incomplete }`; skips (does not half-adopt)
a blob missing its bytes or metadata.
- `brain.vfs.adoptOrphanedBlobs()` — the explicit force path; self-initializes.
- Backup retention: the automatic pre-upgrade backup is now kept if any blob
can't be fully adopted (`incomplete > 0`), instead of being removed on
entity-migration success alone — a blob-parity gate the migration signal
cannot provide.
Adds an end-to-end integration test (storage-level adopt with idempotency and
incomplete handling; self-heal on reopen; the explicit API) and an upgrade guide.
3168 lines
No EOL
102 KiB
TypeScript
3168 lines
No EOL
102 KiB
TypeScript
/**
|
||
* Virtual Filesystem Implementation
|
||
*
|
||
* Virtual filesystem built on Brainy
|
||
* Real code, no mocks, actual working implementation
|
||
*/
|
||
|
||
import { Readable, Writable } from 'stream'
|
||
import crypto from 'crypto'
|
||
import { v4 as uuidv4 } from '../universal/uuid.js'
|
||
import { Brainy } from '../brainy.js'
|
||
import { Entity, AddParams, RelateParams, FindParams, Relation } from '../types/brainy.types.js'
|
||
import { NounType, VerbType } from '../types/graphTypes.js'
|
||
import { PathResolver } from './PathResolver.js'
|
||
import { mimeDetector } from './MimeTypeDetector.js'
|
||
import {
|
||
SemanticPathResolver,
|
||
ProjectionRegistry,
|
||
ConceptProjection,
|
||
AuthorProjection,
|
||
TemporalProjection,
|
||
RelationshipProjection,
|
||
SimilarityProjection,
|
||
TagProjection
|
||
} from './semantic/index.js'
|
||
// Knowledge Layer can remain as optional augmentation for now
|
||
import {
|
||
IVirtualFileSystem,
|
||
VFSConfig,
|
||
VFSEntity,
|
||
VFSMetadata,
|
||
VFSStats,
|
||
VFSDirent,
|
||
VFSTodo,
|
||
VFSError,
|
||
VFSErrorCode,
|
||
WriteOptions,
|
||
ReadOptions,
|
||
MkdirOptions,
|
||
ReaddirOptions,
|
||
CopyOptions,
|
||
SearchOptions,
|
||
SearchResult,
|
||
SimilarOptions,
|
||
RelatedOptions,
|
||
ReadStreamOptions,
|
||
WriteStreamOptions,
|
||
WatchListener
|
||
} from './types.js'
|
||
|
||
/**
|
||
* Main Virtual Filesystem Implementation
|
||
*
|
||
* This is REAL, production-ready code that:
|
||
* - Maps filesystem operations to Brainy entities
|
||
* - Uses graph relationships for directory structure
|
||
* - Provides semantic search and AI features
|
||
* - Scales to millions of files
|
||
*/
|
||
export class VirtualFileSystem implements IVirtualFileSystem {
|
||
private brain: Brainy
|
||
private pathResolver!: SemanticPathResolver
|
||
private projectionRegistry!: ProjectionRegistry
|
||
private config: Required<Omit<VFSConfig, 'rootEntityId'>> & { rootEntityId?: string }
|
||
private rootEntityId?: string
|
||
private initialized = false
|
||
private currentUser: string = 'system' // Track current user for collaboration
|
||
|
||
// Knowledge Layer features available via augmentation (brain.use('knowledge'))
|
||
|
||
// Caches for performance
|
||
private contentCache: Map<string, { data: Buffer, timestamp: number }>
|
||
private statCache: Map<string, { stats: VFSStats, timestamp: number }>
|
||
|
||
// Watch system
|
||
private watchers: Map<string, Set<WatchListener>>
|
||
|
||
// Background task timer
|
||
private backgroundTimer: NodeJS.Timeout | null = null
|
||
|
||
// Mutex for preventing race conditions in directory creation
|
||
private mkdirLocks: Map<string, Promise<void>> = new Map()
|
||
|
||
// Singleton promise for root initialization (prevents duplicate roots)
|
||
private rootInitPromise: Promise<string> | null = null
|
||
|
||
// Fixed VFS root ID (prevents duplicates across instances)
|
||
// Uses deterministic UUID format for storage compatibility
|
||
private static readonly VFS_ROOT_ID = '00000000-0000-0000-0000-000000000000'
|
||
|
||
/**
|
||
* Construct a VFS bound to a Brainy instance.
|
||
*
|
||
* The VFS stores every file and directory as a Brainy entity and models the
|
||
* directory tree with graph relationships. No I/O happens here — call
|
||
* {@link init} (or rely on `brain.init()`, which auto-initializes the VFS)
|
||
* before using any filesystem operation.
|
||
*
|
||
* @param brain - The Brainy instance backing this filesystem. When omitted, a
|
||
* fresh `new Brainy()` is created and owned by this VFS.
|
||
*/
|
||
constructor(brain?: Brainy) {
|
||
this.brain = brain || new Brainy()
|
||
this.contentCache = new Map()
|
||
this.statCache = new Map()
|
||
this.watchers = new Map()
|
||
|
||
// Default configuration (will be overridden in init)
|
||
this.config = this.getDefaultConfig()
|
||
}
|
||
|
||
/**
|
||
* Access to BlobStorage for unified file storage
|
||
*/
|
||
private get blobStorage() {
|
||
// Bracket access reaches Brainy's private `storage` field; `blobStorage`
|
||
// is a public (optional) member of BaseStorage, set during brain.init().
|
||
const storage = this.brain['storage']
|
||
if (!storage?.blobStorage) {
|
||
throw new Error(
|
||
'BlobStorage not available. The storage adapter must run ' +
|
||
'initializeBlobStorage() before the VFS is used (brain.init() does this).'
|
||
)
|
||
}
|
||
return storage.blobStorage
|
||
}
|
||
|
||
/**
|
||
* Initialize the VFS
|
||
*/
|
||
async init(config?: VFSConfig): Promise<void> {
|
||
if (this.initialized) return
|
||
|
||
// Merge config with defaults
|
||
this.config = { ...this.getDefaultConfig(), ...config }
|
||
|
||
// VFS is now auto-initialized during brain.init()
|
||
// Brain is guaranteed to be initialized when this is called
|
||
// Removed brain.init() check to prevent infinite recursion
|
||
|
||
// Create or find root entity
|
||
this.rootEntityId = await this.initializeRoot()
|
||
|
||
// Clean up old UUID-based roots (one-time migration)
|
||
await this.cleanupOldRoots()
|
||
|
||
// Initialize projection registry with auto-discovery of built-in projections
|
||
this.projectionRegistry = new ProjectionRegistry()
|
||
this.registerBuiltInProjections()
|
||
|
||
// Initialize semantic path resolver (zero-config, uses brain.config)
|
||
this.pathResolver = new SemanticPathResolver(
|
||
this.brain,
|
||
this, // Pass VFS instance for resolvePath
|
||
this.rootEntityId,
|
||
this.projectionRegistry
|
||
)
|
||
|
||
// Knowledge Layer is now a separate augmentation
|
||
// Enable with: brain.use('knowledge')
|
||
|
||
// Start background tasks
|
||
this.startBackgroundTasks()
|
||
|
||
this.initialized = true
|
||
}
|
||
|
||
/**
|
||
* Create or find the root directory entity
|
||
*/
|
||
/**
|
||
* Auto-register built-in projection strategies
|
||
* Zero-config: All semantic dimensions work out of the box
|
||
*/
|
||
private registerBuiltInProjections(): void {
|
||
const projections = [
|
||
ConceptProjection,
|
||
AuthorProjection,
|
||
TemporalProjection,
|
||
RelationshipProjection,
|
||
SimilarityProjection,
|
||
TagProjection
|
||
]
|
||
|
||
for (const ProjectionClass of projections) {
|
||
try {
|
||
this.projectionRegistry.register(new ProjectionClass())
|
||
} catch (err) {
|
||
// Silently skip if already registered (e.g., in tests)
|
||
if (!(err instanceof Error && err.message.includes('already registered'))) {
|
||
throw err
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* CRITICAL FIX - Prevent duplicate root creation
|
||
* Uses singleton promise pattern to ensure only ONE root initialization
|
||
* happens even with concurrent init() calls
|
||
*/
|
||
private async initializeRoot(): Promise<string> {
|
||
// If initialization already in progress, wait for it (automatic mutex)
|
||
if (this.rootInitPromise) {
|
||
return await this.rootInitPromise
|
||
}
|
||
|
||
// Start initialization and cache the promise
|
||
this.rootInitPromise = this.doInitializeRoot()
|
||
|
||
try {
|
||
const rootId = await this.rootInitPromise
|
||
return rootId
|
||
} catch (error) {
|
||
// On error, clear promise so retry is possible
|
||
this.rootInitPromise = null
|
||
throw error
|
||
}
|
||
// NOTE: On success, we intentionally keep the promise cached
|
||
// This prevents re-initialization and serves as a cache
|
||
}
|
||
|
||
/**
|
||
* Atomic root initialization with fixed ID
|
||
* Uses deterministic ID to prevent duplicates across all VFS instances
|
||
*
|
||
* ARCHITECTURAL FIX: Instead of query-then-create (race condition),
|
||
* we use a fixed ID so storage-level uniqueness prevents duplicates.
|
||
*/
|
||
private async doInitializeRoot(): Promise<string> {
|
||
const rootId = VirtualFileSystem.VFS_ROOT_ID
|
||
|
||
// Try to get existing root by fixed ID (O(1) lookup, not query)
|
||
try {
|
||
const existingRoot = await this.brain.get(rootId)
|
||
|
||
if (existingRoot) {
|
||
// Root exists - verify metadata is correct
|
||
const metadata = existingRoot.metadata || existingRoot
|
||
|
||
if (!metadata.vfsType || metadata.vfsType !== 'directory') {
|
||
console.warn('⚠️ VFS: Root metadata incomplete, repairing...')
|
||
await this.brain.update({
|
||
id: rootId,
|
||
// Re-assert system visibility on repair so a pre-8.0 root (created before the
|
||
// tier existed) is moved out of the default-visible counts/find() too.
|
||
visibility: 'system',
|
||
metadata: this.getRootMetadata()
|
||
})
|
||
}
|
||
|
||
return rootId
|
||
}
|
||
} catch (error) {
|
||
// Root doesn't exist yet - proceed to creation
|
||
}
|
||
|
||
// Create root with fixed ID (idempotent - fails gracefully if exists)
|
||
try {
|
||
console.log('VFS: Creating root directory (fixed ID: 00000000-0000-0000-0000-000000000000)')
|
||
|
||
await this.brain.add({
|
||
id: rootId, // Fixed ID - storage ensures uniqueness
|
||
data: '/',
|
||
type: NounType.Collection,
|
||
subtype: 'vfs-root', // Standard subtype for the VFS root collection (7.30+)
|
||
// visibility 'system' (8.0): the VFS root is Brainy's own plumbing, not user data,
|
||
// so it is hidden everywhere by default (getNounCount()/find()/stats()) and surfaces
|
||
// only via find({ includeSystem: true }). 'system' is intentionally not part of the
|
||
// public AddParams.visibility union ('public' | 'internal') — this is the single
|
||
// sanctioned internal setter, hence the cast.
|
||
visibility: 'system' as 'public' | 'internal',
|
||
metadata: this.getRootMetadata()
|
||
})
|
||
|
||
return rootId
|
||
} catch (error: any) {
|
||
// If creation failed due to duplicate ID, another instance created it
|
||
// This is normal in concurrent scenarios - just return the fixed ID
|
||
const errorMsg = error?.message?.toLowerCase() || ''
|
||
if (errorMsg.includes('already exists') ||
|
||
errorMsg.includes('duplicate') ||
|
||
errorMsg.includes('eexist')) {
|
||
console.log('VFS: Root already created by another instance, using existing')
|
||
return rootId
|
||
}
|
||
|
||
// Unexpected error
|
||
throw error
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Get standard root metadata
|
||
* Centralized to ensure consistency
|
||
*/
|
||
private getRootMetadata(): VFSMetadata {
|
||
return {
|
||
path: '/',
|
||
name: '',
|
||
vfsType: 'directory',
|
||
isVFS: true,
|
||
isVFSEntity: true,
|
||
size: 0,
|
||
permissions: 0o755,
|
||
owner: 'root',
|
||
group: 'root',
|
||
accessed: Date.now(),
|
||
modified: Date.now()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Cleanup old UUID-based VFS roots
|
||
* Called during init to remove duplicate roots created before fixed-ID fix
|
||
*
|
||
* This is a one-time migration helper that can be removed in future versions.
|
||
*/
|
||
private async cleanupOldRoots(): Promise<void> {
|
||
try {
|
||
// Find any old VFS roots with UUID-based IDs (not our fixed ID)
|
||
const oldRoots = await this.brain.find({
|
||
type: NounType.Collection,
|
||
where: {
|
||
path: '/',
|
||
vfsType: 'directory'
|
||
},
|
||
limit: 100,
|
||
excludeVFS: false
|
||
})
|
||
|
||
// Filter out our fixed-ID root
|
||
const duplicates = oldRoots.filter(r => r.id !== VirtualFileSystem.VFS_ROOT_ID)
|
||
|
||
if (duplicates.length > 0) {
|
||
console.log(`VFS: Found ${duplicates.length} old UUID-based root(s), cleaning up...`)
|
||
|
||
for (const duplicate of duplicates) {
|
||
try {
|
||
await this.brain.remove(duplicate.id)
|
||
console.log(`VFS: Deleted old root ${duplicate.id.substring(0, 8)}`)
|
||
} catch (error) {
|
||
console.warn(`VFS: Failed to delete old root ${duplicate.id}:`, error)
|
||
}
|
||
}
|
||
|
||
console.log('VFS: Cleanup complete - all old roots removed')
|
||
}
|
||
} catch (error) {
|
||
// Non-critical error - log and continue
|
||
console.warn('VFS: Cleanup of old roots failed (non-critical):', error)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @description Recover VFS content blobs orphaned by a 7→8 upgrade. 7.x stored
|
||
* VFS blobs under the copy-on-write area (`_cow/`) of the branch/versioning
|
||
* system that 8.0 removed; a brain upgraded before the migration adopted them
|
||
* reads a stranded blob and throws "Blob metadata not found" (every page
|
||
* backed by it 500s). This adopts every orphaned `_cow/` blob into the 8.0
|
||
* content-addressed store (`_cas/`) IN PLACE — no snapshot restore — then
|
||
* clears the VFS caches so the recovered content reads live immediately.
|
||
*
|
||
* Idempotent and non-destructive: blobs already adopted are skipped and the
|
||
* `_cow/` originals are never deleted (a re-run or rollback stays possible).
|
||
* Safe to run on a healthy or native-8.0 brain (returns zeros). Delegates to
|
||
* {@link BaseStorage.adoptLegacyCowBlobs}.
|
||
*
|
||
* @returns `{ cowBlobs, adopted, alreadyPresent, incomplete }` counts.
|
||
* @example
|
||
* const brain = new Brainy({ storage: { type: 'filesystem', path: '/data/brain' } })
|
||
* await brain.init()
|
||
* const r = await brain.vfs.adoptOrphanedBlobs()
|
||
* console.log(`Adopted ${r.adopted} stranded VFS blobs; ${r.alreadyPresent} already present.`)
|
||
*/
|
||
async adoptOrphanedBlobs(): Promise<{
|
||
cowBlobs: number
|
||
adopted: number
|
||
alreadyPresent: number
|
||
incomplete: number
|
||
}> {
|
||
// The recovery scan works on raw storage objects, so it only needs the
|
||
// brain's storage wired up — not the VFS's own init. brain.init() is
|
||
// idempotent (a no-op on an already-open brain) and, on a cold brain,
|
||
// sets up storage AND runs the on-open auto-adoption itself; the explicit
|
||
// scan below is then the idempotent "force re-scan" equivalent.
|
||
await this.brain.init()
|
||
const storage = this.brain['storage'] as unknown as
|
||
| { adoptLegacyCowBlobs?: () => Promise<{ cowBlobs: number; adopted: number; alreadyPresent: number; incomplete: number }> }
|
||
| undefined
|
||
if (!storage || typeof storage.adoptLegacyCowBlobs !== 'function') {
|
||
return { cowBlobs: 0, adopted: 0, alreadyPresent: 0, incomplete: 0 }
|
||
}
|
||
const result = await storage.adoptLegacyCowBlobs()
|
||
// Drop any cached content/stat so a re-read hits the freshly adopted blobs.
|
||
this.contentCache.clear()
|
||
this.statCache.clear()
|
||
return result
|
||
}
|
||
|
||
// ============= File Operations =============
|
||
|
||
/**
|
||
* Read a file's content
|
||
*/
|
||
async readFile(path: string, options?: ReadOptions): Promise<Buffer> {
|
||
await this.ensureInitialized()
|
||
|
||
// Check cache first
|
||
if (options?.cache !== false && this.contentCache.has(path)) {
|
||
const cached = this.contentCache.get(path)!
|
||
if (Date.now() - cached.timestamp < (this.config.cache?.ttl || 300000)) {
|
||
return cached.data
|
||
}
|
||
}
|
||
|
||
// Resolve path to entity
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Verify it's a file
|
||
if (entity.metadata.vfsType !== 'file') {
|
||
throw new VFSError(VFSErrorCode.EISDIR, `Is a directory: ${path}`, path, 'readFile')
|
||
}
|
||
|
||
// Unified blob storage - ONE path only
|
||
if (!entity.metadata.storage?.type || entity.metadata.storage.type !== 'blob') {
|
||
throw new VFSError(
|
||
VFSErrorCode.EIO,
|
||
`File has no blob storage: ${path}. Requires blob storage format.`,
|
||
path,
|
||
'readFile'
|
||
)
|
||
}
|
||
|
||
// CRITICAL FIX - Isolate blob errors from VFS tree corruption
|
||
// Blob read errors MUST NOT cascade to VFS tree structure
|
||
try {
|
||
// Read from BlobStorage (handles decompression automatically)
|
||
const content = await this.blobStorage.read(entity.metadata.storage.hash)
|
||
|
||
// REMOVED updateAccessTime() for performance
|
||
// Access time updates caused 50-100ms GCS write on EVERY file read
|
||
// Modern file systems use 'noatime' for same reason (performance)
|
||
// Field 'accessed' still exists in metadata for backward compat but won't update
|
||
// await this.updateAccessTime(entityId) // ← REMOVED
|
||
|
||
// Cache the content
|
||
if (options?.cache !== false) {
|
||
this.contentCache.set(path, { data: content, timestamp: Date.now() })
|
||
}
|
||
|
||
// Apply encoding if requested
|
||
if (options?.encoding) {
|
||
return Buffer.from(content.toString(options.encoding))
|
||
}
|
||
|
||
return content
|
||
} catch (blobError) {
|
||
// Blob error isolated - VFS tree structure remains intact
|
||
const errorMsg = blobError instanceof Error ? blobError.message : String(blobError)
|
||
|
||
console.error(`VFS: Cannot read blob for ${path}:`, errorMsg)
|
||
|
||
// Throw VFSError (not blob error) - prevents cascading corruption
|
||
throw new VFSError(
|
||
VFSErrorCode.EIO,
|
||
`File read failed: ${errorMsg}`,
|
||
path,
|
||
'readFile'
|
||
)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Write a file
|
||
*/
|
||
async writeFile(path: string, data: Buffer | string, options?: WriteOptions): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Convert string to buffer
|
||
const buffer = Buffer.isBuffer(data) ? data : Buffer.from(data, options?.encoding)
|
||
|
||
// Check size limits
|
||
if (this.config.limits?.maxFileSize && buffer.length > this.config.limits.maxFileSize) {
|
||
throw new VFSError(VFSErrorCode.ENOSPC, `File too large: ${buffer.length} bytes`, path, 'writeFile')
|
||
}
|
||
|
||
// Parse path to get parent and name
|
||
const parentPath = this.getParentPath(path)
|
||
const name = this.getBasename(path)
|
||
|
||
// Ensure parent directory exists
|
||
const parentId = await this.ensureDirectory(parentPath)
|
||
|
||
// Check if file already exists
|
||
let existingId: string | null = null
|
||
try {
|
||
existingId = await this.pathResolver.resolve(path, { cache: false })
|
||
// Verify the entity still exists in the brain
|
||
const existing = await this.brain.get(existingId)
|
||
if (!existing) {
|
||
existingId = null // Entity was deleted but cache wasn't cleared
|
||
}
|
||
} catch (err) {
|
||
// File doesn't exist, which is fine
|
||
existingId = null
|
||
}
|
||
|
||
// Detect MIME type BEFORE the blob write so the store's auto-compression
|
||
// policy can skip zstd for already-compressed media (JPEG, MP4, ZIP, …).
|
||
const mimeType = mimeDetector.detectMimeType(name, buffer)
|
||
|
||
// Unified blob storage for ALL files (no size-based branching)
|
||
// Store in BlobStorage (content-addressable, auto-deduplication)
|
||
const blobHash = await this.blobStorage.write(buffer, { mimeType })
|
||
|
||
// Get blob metadata (size, compression info)
|
||
const blobMetadata = await this.blobStorage.getMetadata(blobHash)
|
||
|
||
const storageStrategy: VFSMetadata['storage'] = {
|
||
type: 'blob',
|
||
hash: blobHash,
|
||
size: buffer.length,
|
||
compressed: blobMetadata ? blobMetadata.compression !== 'none' : undefined
|
||
}
|
||
|
||
// Create metadata
|
||
const metadata: VFSMetadata = {
|
||
path,
|
||
name,
|
||
parent: parentId,
|
||
vfsType: 'file',
|
||
isVFS: true, // Mark as VFS entity (internal)
|
||
isVFSEntity: true, // Explicit flag for developer filtering
|
||
size: buffer.length,
|
||
mimeType,
|
||
extension: this.getExtension(name),
|
||
permissions: options?.mode || this.config.permissions?.defaultFile || 0o644,
|
||
owner: 'user', // In production, get from auth context
|
||
group: 'users',
|
||
accessed: Date.now(),
|
||
modified: Date.now(),
|
||
storage: storageStrategy
|
||
// No rawData - content is in BlobStorage
|
||
// Backward compatibility: readFile() checks for rawData for legacy files
|
||
}
|
||
|
||
// Extract additional metadata if enabled
|
||
if (this.config.intelligence?.autoExtract && options?.extractMetadata !== false) {
|
||
Object.assign(metadata, await this.extractMetadata(buffer, mimeType))
|
||
}
|
||
|
||
if (existingId) {
|
||
// Update existing file
|
||
// No entity.data - content is in BlobStorage
|
||
await this.brain.update({
|
||
id: existingId,
|
||
metadata
|
||
})
|
||
|
||
// Ensure Contains relationship exists (fix for missing relationships)
|
||
const existingRelations = await this.brain.related({
|
||
from: parentId,
|
||
to: existingId,
|
||
type: VerbType.Contains
|
||
})
|
||
|
||
// Create relationship if it doesn't exist
|
||
if (existingRelations.length === 0) {
|
||
await this.brain.relate({
|
||
from: parentId,
|
||
to: existingId,
|
||
type: VerbType.Contains,
|
||
subtype: 'vfs-contains', // Standard subtype for VFS containment edges (7.30+)
|
||
metadata: { isVFS: true } // Mark as VFS relationship
|
||
})
|
||
}
|
||
} else {
|
||
// Create new file entity
|
||
// For embedding: use text content, for storage: use raw data
|
||
const embeddingData = mimeDetector.isTextFile(mimeType) ? buffer.toString('utf-8') : `File: ${name} (${mimeType}, ${buffer.length} bytes)`
|
||
|
||
const entity = await this.brain.add({
|
||
data: embeddingData, // Always provide string for embeddings
|
||
type: this.getFileNounType(mimeType),
|
||
subtype: 'vfs-file', // Standard subtype for VFS file entities (7.30+)
|
||
metadata
|
||
})
|
||
|
||
// Create parent-child relationship (no need to check for duplicates on new entities)
|
||
await this.brain.relate({
|
||
from: parentId,
|
||
to: entity,
|
||
type: VerbType.Contains,
|
||
subtype: 'vfs-contains', // Standard subtype for VFS containment edges (7.30+)
|
||
metadata: { isVFS: true } // Mark as VFS relationship
|
||
})
|
||
|
||
// Update path resolver cache
|
||
await this.pathResolver.createPath(path, entity)
|
||
}
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
|
||
// Trigger watchers
|
||
this.triggerWatchers(path, existingId ? 'change' : 'rename')
|
||
|
||
// Knowledge Layer hooks will be added by augmentation if enabled
|
||
|
||
// Knowledge Layer hooks will be added by augmentation if enabled
|
||
}
|
||
|
||
/**
|
||
* Append to a file
|
||
*/
|
||
async appendFile(path: string, data: Buffer | string, options?: WriteOptions): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Read existing content
|
||
let existing: Buffer
|
||
try {
|
||
existing = await this.readFile(path)
|
||
} catch (err) {
|
||
// File doesn't exist, create it
|
||
return this.writeFile(path, data, options)
|
||
}
|
||
|
||
// Append new data
|
||
const newData = Buffer.isBuffer(data) ? data : Buffer.from(data, options?.encoding)
|
||
const combined = Buffer.concat([existing, newData])
|
||
|
||
// Write combined content
|
||
await this.writeFile(path, combined, options)
|
||
}
|
||
|
||
/**
|
||
* Delete a file
|
||
*/
|
||
async unlink(path: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Verify it's a file
|
||
if (entity.metadata.vfsType !== 'file') {
|
||
throw new VFSError(VFSErrorCode.EISDIR, `Is a directory: ${path}`, path, 'unlink')
|
||
}
|
||
|
||
// Delete blob from BlobStorage (decrements ref count)
|
||
if (entity.metadata.storage?.type === 'blob') {
|
||
await this.blobStorage.delete(entity.metadata.storage.hash)
|
||
}
|
||
|
||
// Delete the entity
|
||
await this.brain.remove(entityId)
|
||
|
||
// Invalidate caches
|
||
this.pathResolver.invalidatePath(path)
|
||
this.invalidateCaches(path)
|
||
|
||
// Trigger watchers
|
||
this.triggerWatchers(path, 'rename')
|
||
|
||
// Knowledge Layer hooks will be added by augmentation if enabled
|
||
}
|
||
|
||
// ============= Tree Operations (NEW) =============
|
||
|
||
/**
|
||
* Get only direct children of a directory - guaranteed no self-inclusion
|
||
* This is the SAFE way to get children for building tree UIs
|
||
*/
|
||
async getDirectChildren(path: string): Promise<VFSEntity[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Verify it's a directory
|
||
if (entity.metadata.vfsType !== 'directory') {
|
||
throw new VFSError(VFSErrorCode.ENOTDIR, `Not a directory: ${path}`, path, 'getDirectChildren')
|
||
}
|
||
|
||
// Use the safe getChildren from PathResolver
|
||
const children = await this.pathResolver.getChildren(entityId)
|
||
|
||
// Double-check no self-inclusion (paranoid safety)
|
||
return children.filter(child => child.metadata.path !== path)
|
||
}
|
||
|
||
/**
|
||
* Gather descendants using graph traversal + bulk fetch
|
||
*
|
||
* ARCHITECTURE:
|
||
* 1. Traverse graph to collect entity IDs (in-memory, fast)
|
||
* 2. Batch-fetch all entities in ONE storage call
|
||
* 3. Return flat list of VFSEntity objects
|
||
*
|
||
* This is the ONLY correct approach:
|
||
* - Uses GraphAdjacencyIndex (in-memory graph) to traverse relationships
|
||
* - Makes ONE storage call to fetch all entities (not N calls)
|
||
* - Respects maxDepth to limit scope (billion-scale safe)
|
||
*
|
||
* Performance (GCS):
|
||
* - OLD: 111 directories × 50ms each = 5,550ms
|
||
* - NEW: Graph traversal (1ms) + 1 batch fetch (100ms) = 101ms
|
||
* - 55x faster on cloud storage
|
||
*
|
||
* @param rootId - Root directory entity ID
|
||
* @param maxDepth - Maximum depth to traverse
|
||
* @returns All descendant entities (flat list)
|
||
*/
|
||
private async gatherDescendants(rootId: string, maxDepth: number): Promise<VFSEntity[]> {
|
||
const entityIds = new Set<string>()
|
||
const visited = new Set<string>([rootId])
|
||
let currentLevel = [rootId]
|
||
let depth = 0
|
||
|
||
// Phase 1: Traverse graph in-memory to collect all entity IDs
|
||
// GraphAdjacencyIndex is in-memory LSM-tree, so this is fast (<10ms for 10k relationships)
|
||
while (currentLevel.length > 0 && depth < maxDepth) {
|
||
const nextLevel: string[] = []
|
||
|
||
// Get all Contains relationships for this level (in-memory query)
|
||
for (const parentId of currentLevel) {
|
||
const relations = await this.brain.related({
|
||
from: parentId,
|
||
type: VerbType.Contains
|
||
})
|
||
|
||
// Collect child IDs
|
||
for (const rel of relations) {
|
||
if (!visited.has(rel.to)) {
|
||
visited.add(rel.to)
|
||
entityIds.add(rel.to)
|
||
nextLevel.push(rel.to) // Queue for next level
|
||
}
|
||
}
|
||
}
|
||
|
||
currentLevel = nextLevel
|
||
depth++
|
||
}
|
||
|
||
// Phase 2: Batch-fetch all entities in ONE storage call
|
||
// This is the optimization: ONE GCS call instead of 111+ GCS calls
|
||
const entityIdArray = Array.from(entityIds)
|
||
if (entityIdArray.length === 0) {
|
||
return []
|
||
}
|
||
|
||
const entitiesMap = await this.brain.batchGet(entityIdArray)
|
||
|
||
// Convert to VFSEntity array
|
||
const entities: VFSEntity[] = []
|
||
for (const id of entityIdArray) {
|
||
const entity = entitiesMap.get(id)
|
||
if (entity && entity.metadata?.vfsType) {
|
||
entities.push(entity as VFSEntity)
|
||
}
|
||
}
|
||
|
||
return entities
|
||
}
|
||
|
||
/**
|
||
* Get a properly structured tree for the given path
|
||
*
|
||
* Graph traversal + ONE batch fetch (55x faster on cloud storage)
|
||
*
|
||
* Architecture:
|
||
* 1. Resolve path to entity ID
|
||
* 2. Traverse graph in-memory to collect all descendant IDs
|
||
* 3. Batch-fetch all entities in ONE storage call
|
||
* 4. Build tree structure
|
||
*
|
||
* Performance:
|
||
* - GCS: 5,300ms → ~100ms (53x faster)
|
||
* - FileSystem: 200ms → ~50ms (4x faster)
|
||
*/
|
||
async getTreeStructure(path: string, options?: {
|
||
maxDepth?: number
|
||
includeHidden?: boolean
|
||
sort?: 'name' | 'modified' | 'size'
|
||
}): Promise<any> {
|
||
await this.ensureInitialized()
|
||
const { VFSTreeUtils } = await import('./TreeUtils.js')
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
if (entity.metadata.vfsType !== 'directory') {
|
||
throw new VFSError(VFSErrorCode.ENOTDIR, `Not a directory: ${path}`, path, 'getTreeStructure')
|
||
}
|
||
|
||
const maxDepth = options?.maxDepth ?? 10
|
||
|
||
// Gather all descendants (graph traversal + ONE batch fetch)
|
||
const allEntities = await this.gatherDescendants(entityId, maxDepth)
|
||
|
||
// Build tree structure
|
||
return VFSTreeUtils.buildTree(allEntities, path, options || {})
|
||
}
|
||
|
||
/**
|
||
* Get all descendants of a directory (flat list)
|
||
*
|
||
* Same optimization as getTreeStructure
|
||
*/
|
||
async getDescendants(path: string, options?: {
|
||
includeAncestor?: boolean
|
||
type?: 'file' | 'directory'
|
||
}): Promise<VFSEntity[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
if (entity.metadata.vfsType !== 'directory') {
|
||
throw new VFSError(VFSErrorCode.ENOTDIR, `Not a directory: ${path}`, path, 'getDescendants')
|
||
}
|
||
|
||
// Gather all descendants (no depth limit for this API)
|
||
const descendants = await this.gatherDescendants(entityId, Infinity)
|
||
|
||
// Filter by type if specified
|
||
const filtered = options?.type
|
||
? descendants.filter(d => d.metadata.vfsType === options.type)
|
||
: descendants
|
||
|
||
// Include ancestor if requested
|
||
if (options?.includeAncestor) {
|
||
return [entity, ...filtered]
|
||
}
|
||
|
||
return filtered
|
||
}
|
||
|
||
/**
|
||
* Inspect a path and return structured information
|
||
* This is the recommended method for file explorers to use
|
||
*/
|
||
async inspect(path: string): Promise<{
|
||
node: VFSEntity
|
||
children: VFSEntity[]
|
||
parent: VFSEntity | null
|
||
stats: VFSStats
|
||
}> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
const stats = await this.stat(path)
|
||
|
||
let children: VFSEntity[] = []
|
||
if (entity.metadata.vfsType === 'directory') {
|
||
children = await this.getDirectChildren(path)
|
||
}
|
||
|
||
let parent: VFSEntity | null = null
|
||
if (path !== '/') {
|
||
const parentPath = path.substring(0, path.lastIndexOf('/')) || '/'
|
||
const parentId = await this.pathResolver.resolve(parentPath)
|
||
parent = await this.getEntityById(parentId)
|
||
}
|
||
|
||
return {
|
||
node: entity,
|
||
children,
|
||
parent,
|
||
stats
|
||
}
|
||
}
|
||
|
||
// ============= Directory Operations =============
|
||
|
||
/**
|
||
* Create a directory
|
||
*/
|
||
async mkdir(path: string, options?: MkdirOptions): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Use mutex to prevent race conditions when creating the same directory concurrently
|
||
// If another call is already creating this directory, wait for it to complete
|
||
const existingLock = this.mkdirLocks.get(path)
|
||
if (existingLock) {
|
||
await existingLock
|
||
// After waiting, check if directory now exists
|
||
try {
|
||
const existing = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(existing)
|
||
if (entity.metadata.vfsType === 'directory') {
|
||
return // Directory was created by the other call
|
||
}
|
||
} catch (err) {
|
||
// Still doesn't exist, proceed to create
|
||
}
|
||
}
|
||
|
||
// Create a lock promise for this path
|
||
let resolveLock: () => void
|
||
const lockPromise = new Promise<void>(resolve => { resolveLock = resolve })
|
||
this.mkdirLocks.set(path, lockPromise)
|
||
|
||
try {
|
||
// Check if already exists
|
||
try {
|
||
const existing = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(existing)
|
||
if (entity.metadata.vfsType === 'directory') {
|
||
if (!options?.recursive) {
|
||
throw new VFSError(VFSErrorCode.EEXIST, `Directory exists: ${path}`, path, 'mkdir')
|
||
}
|
||
return // Already exists and recursive is true
|
||
} else {
|
||
// Path exists but it's not a directory
|
||
throw new VFSError(VFSErrorCode.EEXIST, `File exists: ${path}`, path, 'mkdir')
|
||
}
|
||
} catch (err) {
|
||
// Only proceed if it's a ENOENT error (path doesn't exist)
|
||
if (err instanceof VFSError && err.code !== VFSErrorCode.ENOENT) {
|
||
throw err // Re-throw non-ENOENT errors
|
||
}
|
||
// Doesn't exist, proceed to create
|
||
}
|
||
|
||
// Parse path
|
||
const parentPath = this.getParentPath(path)
|
||
const name = this.getBasename(path)
|
||
|
||
// Ensure parent exists (recursive mkdir if needed)
|
||
let parentId: string
|
||
if (parentPath === '/' || parentPath === null) {
|
||
parentId = this.rootEntityId!
|
||
} else if (options?.recursive) {
|
||
parentId = await this.ensureDirectory(parentPath)
|
||
} else {
|
||
try {
|
||
parentId = await this.pathResolver.resolve(parentPath)
|
||
} catch (err) {
|
||
throw new VFSError(VFSErrorCode.ENOENT, `Parent directory not found: ${parentPath}`, path, 'mkdir')
|
||
}
|
||
}
|
||
|
||
// Create directory entity
|
||
const metadata: VFSMetadata = {
|
||
path,
|
||
name,
|
||
parent: parentId,
|
||
vfsType: 'directory',
|
||
isVFS: true, // Mark as VFS entity (internal)
|
||
isVFSEntity: true, // Explicit flag for developer filtering
|
||
size: 0,
|
||
permissions: options?.mode || this.config.permissions?.defaultDirectory || 0o755,
|
||
owner: 'user',
|
||
group: 'users',
|
||
accessed: Date.now(),
|
||
modified: Date.now(),
|
||
...options?.metadata
|
||
}
|
||
|
||
const entity = await this.brain.add({
|
||
data: path, // Directory path as string content
|
||
type: NounType.Collection,
|
||
subtype: 'vfs-directory', // Standard subtype for VFS directory entities (7.30+)
|
||
metadata
|
||
})
|
||
|
||
// Create parent-child relationship (no need to check for duplicates on new entities)
|
||
if (parentId !== entity) { // Don't relate to self (root)
|
||
await this.brain.relate({
|
||
from: parentId,
|
||
to: entity,
|
||
type: VerbType.Contains,
|
||
subtype: 'vfs-contains', // Standard subtype for VFS containment edges (7.30+)
|
||
metadata: {
|
||
isVFS: true, // Mark as VFS relationship
|
||
relationshipType: 'vfs' // Standardized relationship type metadata
|
||
}
|
||
})
|
||
}
|
||
|
||
// Update path resolver cache
|
||
await this.pathResolver.createPath(path, entity)
|
||
|
||
// Trigger watchers
|
||
this.triggerWatchers(path, 'rename')
|
||
} finally {
|
||
// Release the lock
|
||
resolveLock!()
|
||
this.mkdirLocks.delete(path)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Remove a directory
|
||
*
|
||
* Optimized for cloud storage using batch operations
|
||
* - Uses gatherDescendants() for efficient graph traversal + batch fetch
|
||
* - Uses removeMany() for chunked transactional deletion
|
||
* - Parallel blob cleanup with chunking
|
||
*
|
||
* Performance improvement: 4-8x faster on cloud storage (GCS, S3, R2, Azure)
|
||
* - 15 files on GCS: 120s → 15-30s
|
||
*/
|
||
async rmdir(path: string, options?: { recursive?: boolean }): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
if (path === '/') {
|
||
throw new VFSError(VFSErrorCode.EACCES, 'Cannot remove root directory', path, 'rmdir')
|
||
}
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Verify it's a directory
|
||
if (entity.metadata.vfsType !== 'directory') {
|
||
throw new VFSError(VFSErrorCode.ENOTDIR, `Not a directory: ${path}`, path, 'rmdir')
|
||
}
|
||
|
||
// Check if empty (unless recursive)
|
||
const children = await this.pathResolver.getChildren(entityId)
|
||
if (children.length > 0 && !options?.recursive) {
|
||
throw new VFSError(VFSErrorCode.ENOTEMPTY, `Directory not empty: ${path}`, path, 'rmdir')
|
||
}
|
||
|
||
// OPTIMIZED batch deletion for recursive case
|
||
if (options?.recursive && children.length > 0) {
|
||
// Phase 1: Gather all descendants in ONE batch fetch
|
||
const descendants = await this.gatherDescendants(entityId, Infinity)
|
||
|
||
// Phase 2: Parallel blob cleanup (chunked to avoid overwhelming storage)
|
||
// Blob deletion is reference-counted, so safe to call for all files
|
||
const blobFiles = descendants.filter(d =>
|
||
d.metadata.vfsType === 'file' && d.metadata.storage?.type === 'blob'
|
||
)
|
||
|
||
const BLOB_CHUNK_SIZE = 20 // Parallel delete 20 blobs at a time
|
||
for (let i = 0; i < blobFiles.length; i += BLOB_CHUNK_SIZE) {
|
||
const chunk = blobFiles.slice(i, i + BLOB_CHUNK_SIZE)
|
||
await Promise.all(chunk.map(f =>
|
||
this.blobStorage.delete(f.metadata.storage!.hash)
|
||
))
|
||
}
|
||
|
||
// Phase 3: Batch delete all entities (including root directory)
|
||
const allIds = [...descendants.map(d => d.id), entityId]
|
||
await this.brain.removeMany({ ids: allIds, continueOnError: false })
|
||
} else {
|
||
// No children or not recursive - just delete the directory entity
|
||
await this.brain.remove(entityId)
|
||
}
|
||
|
||
// Invalidate caches (recursive invalidation handles all descendants)
|
||
this.pathResolver.invalidatePath(path, true)
|
||
this.invalidateCaches(path, true)
|
||
|
||
// Trigger watchers
|
||
this.triggerWatchers(path, 'rename')
|
||
}
|
||
|
||
/**
|
||
* Read directory contents
|
||
*/
|
||
async readdir(path: string, options?: ReaddirOptions): Promise<string[] | VFSDirent[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Verify it's a directory
|
||
if (entity.metadata.vfsType !== 'directory') {
|
||
throw new VFSError(VFSErrorCode.ENOTDIR, `Not a directory: ${path}`, path, 'readdir')
|
||
}
|
||
|
||
// Get children
|
||
let children = await this.pathResolver.getChildren(entityId)
|
||
|
||
// Apply filters
|
||
if (options?.filter) {
|
||
children = this.filterDirectoryEntries(children, options.filter)
|
||
}
|
||
|
||
// Sort if requested
|
||
if (options?.sort) {
|
||
children = this.sortDirectoryEntries(children, options.sort, options.order)
|
||
}
|
||
|
||
// Apply pagination
|
||
if (options?.offset) {
|
||
children = children.slice(options.offset)
|
||
}
|
||
if (options?.limit) {
|
||
children = children.slice(0, options.limit)
|
||
}
|
||
|
||
// REMOVED updateAccessTime() for performance
|
||
// Directory access time updates caused 50-100ms GCS write on EVERY readdir
|
||
// await this.updateAccessTime(entityId) // ← REMOVED
|
||
|
||
// Return appropriate format
|
||
if (options?.withFileTypes) {
|
||
return children.map(child => ({
|
||
name: child.metadata.name,
|
||
path: child.metadata.path,
|
||
type: child.metadata.vfsType,
|
||
entityId: child.id
|
||
} as VFSDirent))
|
||
}
|
||
|
||
return children.map(child => child.metadata.name)
|
||
}
|
||
|
||
// ============= Metadata Operations =============
|
||
|
||
/**
|
||
* Get file/directory statistics
|
||
*/
|
||
async stat(path: string): Promise<VFSStats> {
|
||
await this.ensureInitialized()
|
||
|
||
// Check cache
|
||
if (this.statCache.has(path)) {
|
||
const cached = this.statCache.get(path)!
|
||
if (Date.now() - cached.timestamp < 5000) { // 5 second cache
|
||
return cached.stats
|
||
}
|
||
}
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
const stats: VFSStats = {
|
||
size: entity.metadata.size,
|
||
mode: entity.metadata.permissions,
|
||
uid: 1000, // In production, map owner to UID
|
||
gid: 1000, // In production, map group to GID
|
||
atime: new Date(entity.metadata.accessed),
|
||
mtime: new Date(entity.metadata.modified),
|
||
ctime: new Date(entity.updatedAt || entity.createdAt),
|
||
birthtime: new Date(entity.createdAt),
|
||
isFile: () => entity.metadata.vfsType === 'file',
|
||
isDirectory: () => entity.metadata.vfsType === 'directory',
|
||
isSymbolicLink: () => entity.metadata.vfsType === 'symlink',
|
||
path,
|
||
entityId: entity.id,
|
||
vector: entity.vector,
|
||
connections: await this.countRelationships(entityId)
|
||
}
|
||
|
||
// Cache stats
|
||
this.statCache.set(path, { stats, timestamp: Date.now() })
|
||
|
||
return stats
|
||
}
|
||
|
||
/**
|
||
* lstat - same as stat for now (symlinks not fully implemented)
|
||
*/
|
||
async lstat(path: string): Promise<VFSStats> {
|
||
return this.stat(path)
|
||
}
|
||
|
||
/**
|
||
* Check if path exists
|
||
*/
|
||
async exists(path: string): Promise<boolean> {
|
||
await this.ensureInitialized()
|
||
|
||
try {
|
||
await this.pathResolver.resolve(path)
|
||
return true
|
||
} catch (err) {
|
||
return false
|
||
}
|
||
}
|
||
|
||
// ============= Semantic Operations =============
|
||
|
||
/**
|
||
* Search files with natural language
|
||
*/
|
||
async search(query: string, options?: SearchOptions): Promise<SearchResult[]> {
|
||
await this.ensureInitialized()
|
||
|
||
// Build find params
|
||
const params: FindParams = {
|
||
query,
|
||
type: [NounType.File, NounType.Document, NounType.Media],
|
||
limit: options?.limit || 10,
|
||
offset: options?.offset,
|
||
where: {
|
||
vfsType: 'file' // Search VFS files
|
||
}
|
||
}
|
||
|
||
// Add path filter if specified
|
||
if (options?.path) {
|
||
params.where = {
|
||
...params.where,
|
||
path: { $startsWith: options.path }
|
||
}
|
||
}
|
||
|
||
// Add metadata filters
|
||
if (options?.where) {
|
||
Object.assign(params.where || {}, options.where)
|
||
}
|
||
|
||
// Execute search using Brainy's Triple Intelligence
|
||
const results = await this.brain.find(params)
|
||
|
||
// Convert to search results
|
||
return results.map(r => {
|
||
const entity = r.entity as VFSEntity
|
||
return {
|
||
path: entity.metadata.path,
|
||
entityId: entity.id,
|
||
score: r.score,
|
||
type: entity.metadata.vfsType,
|
||
size: entity.metadata.size,
|
||
modified: new Date(entity.metadata.modified),
|
||
explanation: r.explanation
|
||
}
|
||
})
|
||
}
|
||
|
||
/**
|
||
* Find files similar to a given file
|
||
*/
|
||
async findSimilar(path: string, options?: SimilarOptions): Promise<SearchResult[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
|
||
// Use Brainy's similarity search
|
||
const results = await this.brain.similar({
|
||
to: entityId,
|
||
limit: options?.limit || 10,
|
||
threshold: options?.threshold || 0.7,
|
||
type: [NounType.File, NounType.Document, NounType.Media],
|
||
where: {
|
||
vfsType: 'file' // Find similar VFS files
|
||
}
|
||
})
|
||
|
||
return results.map(r => {
|
||
const entity = r.entity as VFSEntity
|
||
return {
|
||
path: entity.metadata.path,
|
||
entityId: entity.id,
|
||
score: r.score,
|
||
type: entity.metadata.vfsType,
|
||
size: entity.metadata.size,
|
||
modified: new Date(entity.metadata.modified)
|
||
}
|
||
})
|
||
}
|
||
|
||
// ============= Helper Methods =============
|
||
|
||
private async ensureInitialized(): Promise<void> {
|
||
if (!this.initialized) {
|
||
throw new VFSError(
|
||
VFSErrorCode.EINVAL,
|
||
'VFS not initialized. Call await vfs.init() before using VFS operations.\n\n' +
|
||
'✅ After brain.import():\n' +
|
||
' await brain.import(file, { vfsPath: "/imports/data" })\n' +
|
||
' const vfs = brain.vfs\n' +
|
||
' await vfs.init() // ← Required! Safe to call multiple times\n' +
|
||
' const files = await vfs.readdir("/imports/data")\n\n' +
|
||
'✅ Direct VFS usage:\n' +
|
||
' const vfs = brain.vfs\n' +
|
||
' await vfs.init() // ← Always required before first use\n' +
|
||
' await vfs.writeFile("/docs/readme.md", "Hello")\n\n' +
|
||
'📖 Docs: https://github.com/soulcraftlabs/brainy/blob/main/docs/vfs/QUICK_START.md',
|
||
'<unknown>',
|
||
'VFS'
|
||
)
|
||
}
|
||
}
|
||
|
||
private async ensureDirectory(path: string): Promise<string> {
|
||
if (!path || path === '/') {
|
||
return this.rootEntityId!
|
||
}
|
||
|
||
try {
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
if (entity.metadata.vfsType !== 'directory') {
|
||
throw new VFSError(VFSErrorCode.ENOTDIR, `Not a directory: ${path}`, path)
|
||
}
|
||
return entityId
|
||
} catch (err) {
|
||
// Only create directory if it doesn't exist (ENOENT error)
|
||
if (err instanceof VFSError && err.code === VFSErrorCode.ENOENT) {
|
||
await this.mkdir(path, { recursive: true })
|
||
return await this.pathResolver.resolve(path)
|
||
}
|
||
// Re-throw other errors (like ENOTDIR)
|
||
throw err
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Fetch a Brainy entity by id and normalize it into a {@link VFSEntity}.
|
||
*
|
||
* Backfills VFS metadata for legacy entities that stored fields at the top
|
||
* level (pre-nested-metadata layout) and guarantees the root directory always
|
||
* carries valid directory metadata.
|
||
*
|
||
* @param id - The Brainy entity id to fetch.
|
||
* @returns The entity with a populated `metadata.vfsType` and related VFS fields.
|
||
* @throws {VFSError} ENOENT when no entity exists for the given id.
|
||
*/
|
||
async getEntityById(id: string): Promise<VFSEntity> {
|
||
const entity = await this.brain.get(id)
|
||
|
||
if (!entity) {
|
||
throw new VFSError(VFSErrorCode.ENOENT, `Entity not found: ${id}`)
|
||
}
|
||
|
||
// Ensure entity has proper VFS metadata structure
|
||
// Handle both nested and flat metadata structures for compatibility
|
||
if (!entity.metadata || !entity.metadata.vfsType) {
|
||
// Check if metadata is at top level (legacy structure). Legacy entities
|
||
// carried VFS fields as extra top-level properties not modeled on Entity.
|
||
const anyEntity = entity as Entity & Record<string, unknown>
|
||
if (anyEntity.vfsType || anyEntity.path) {
|
||
entity.metadata = {
|
||
path: anyEntity.path || '/',
|
||
name: anyEntity.name || '',
|
||
vfsType: anyEntity.vfsType || (anyEntity.path === '/' ? 'directory' : 'file'),
|
||
size: anyEntity.size || 0,
|
||
permissions: anyEntity.permissions || (anyEntity.vfsType === 'directory' ? 0o755 : 0o644),
|
||
owner: anyEntity.owner || 'user',
|
||
group: anyEntity.group || 'users',
|
||
accessed: anyEntity.accessed || Date.now(),
|
||
modified: anyEntity.modified || Date.now(),
|
||
...entity.metadata // Preserve any existing nested metadata
|
||
}
|
||
} else if (entity.id === this.rootEntityId) {
|
||
// Special case: ensure root directory always has proper metadata
|
||
entity.metadata = {
|
||
path: '/',
|
||
name: '',
|
||
vfsType: 'directory',
|
||
size: 0,
|
||
permissions: 0o755,
|
||
owner: 'root',
|
||
group: 'root',
|
||
accessed: Date.now(),
|
||
modified: Date.now(),
|
||
...entity.metadata
|
||
}
|
||
}
|
||
}
|
||
|
||
return entity as VFSEntity
|
||
}
|
||
|
||
private getParentPath(path: string): string {
|
||
const normalized = path.replace(/\/+/g, '/').replace(/\/$/, '')
|
||
const lastSlash = normalized.lastIndexOf('/')
|
||
if (lastSlash <= 0) return '/'
|
||
return normalized.substring(0, lastSlash)
|
||
}
|
||
|
||
private getBasename(path: string): string {
|
||
const normalized = path.replace(/\/+/g, '/').replace(/\/$/, '')
|
||
const lastSlash = normalized.lastIndexOf('/')
|
||
return normalized.substring(lastSlash + 1)
|
||
}
|
||
|
||
private getExtension(filename: string): string | undefined {
|
||
const lastDot = filename.lastIndexOf('.')
|
||
if (lastDot === -1 || lastDot === 0) return undefined
|
||
return filename.substring(lastDot + 1).toLowerCase()
|
||
}
|
||
|
||
// MIME detection moved to MimeTypeDetector service
|
||
// Removed detectMimeType() and isTextFile() - now using mimeDetector singleton
|
||
|
||
private getFileNounType(mimeType: string): NounType {
|
||
if (mimeType.startsWith('text/') || mimeType.includes('json')) {
|
||
return NounType.Document
|
||
}
|
||
if (mimeType.startsWith('image/') || mimeType.startsWith('video/') || mimeType.startsWith('audio/')) {
|
||
return NounType.Media
|
||
}
|
||
return NounType.File
|
||
}
|
||
|
||
// Removed compression methods (shouldCompress, compress, decompress)
|
||
// BlobStorage handles all compression automatically with zstd
|
||
|
||
private async generateEmbedding(buffer: Buffer, mimeType: string): Promise<number[] | undefined> {
|
||
try {
|
||
// Use text content for text files, description for binary
|
||
let content: string
|
||
if (mimeDetector.isTextFile(mimeType)) {
|
||
// Use first 10KB for embedding
|
||
content = buffer.toString('utf8', 0, Math.min(10240, buffer.length))
|
||
} else {
|
||
// For binary files, create a description
|
||
content = `Binary file: ${mimeType}, size: ${buffer.length} bytes`
|
||
}
|
||
|
||
// Ensure content is actually a string
|
||
if (typeof content !== 'string') {
|
||
console.debug('Content is not a string:', typeof content, content)
|
||
return undefined
|
||
}
|
||
|
||
// Ensure content is not empty or invalid
|
||
if (!content || content.length === 0) {
|
||
console.debug('Content is empty')
|
||
return undefined
|
||
}
|
||
|
||
const vector = await this.brain.embed(content)
|
||
return vector
|
||
} catch (error) {
|
||
console.debug('Failed to generate embedding:', error)
|
||
return undefined
|
||
}
|
||
}
|
||
|
||
private async extractMetadata(buffer: Buffer, mimeType: string): Promise<Partial<VFSMetadata>> {
|
||
const metadata: Partial<VFSMetadata> = {}
|
||
|
||
// Extract basic metadata based on content type
|
||
if (mimeDetector.isTextFile(mimeType)) {
|
||
const text = buffer.toString('utf8')
|
||
metadata.lineCount = text.split('\n').length
|
||
metadata.wordCount = text.split(/\s+/).filter(w => w).length
|
||
metadata.charset = 'utf-8'
|
||
|
||
// Extract concepts using brain.extractConcepts() (neural extraction)
|
||
if (this.config.intelligence?.autoConcepts) {
|
||
try {
|
||
const concepts = await this.brain.extractConcepts(text, { limit: 20 })
|
||
metadata.conceptNames = concepts // Flattened for O(log n) queries
|
||
} catch (error) {
|
||
// Concept extraction is optional - don't fail if it errors
|
||
console.debug('Concept extraction failed:', error)
|
||
}
|
||
}
|
||
}
|
||
|
||
// Extract hash for integrity
|
||
const crypto = await import('crypto')
|
||
metadata.hash = crypto.createHash('sha256').update(buffer).digest('hex')
|
||
|
||
return metadata
|
||
}
|
||
|
||
// REMOVED updateAccessTime() method entirely
|
||
// Access time updates caused 50-100ms GCS write on EVERY file/dir read
|
||
// Modern file systems use 'noatime' for same reason
|
||
// Field 'accessed' still exists in metadata for backward compat but won't update
|
||
|
||
private async countRelationships(entityId: string): Promise<number> {
|
||
const relations = await this.brain.related({ from: entityId })
|
||
const relationsTo = await this.brain.related({ to: entityId })
|
||
return relations.length + relationsTo.length
|
||
}
|
||
|
||
private filterDirectoryEntries(entries: VFSEntity[], filter: any): VFSEntity[] {
|
||
return entries.filter(entry => {
|
||
if (filter.type && entry.metadata.vfsType !== filter.type) return false
|
||
if (filter.pattern && !this.matchGlob(entry.metadata.name, filter.pattern)) return false
|
||
if (filter.minSize && entry.metadata.size < filter.minSize) return false
|
||
if (filter.maxSize && entry.metadata.size > filter.maxSize) return false
|
||
if (filter.modifiedAfter && entry.metadata.modified < filter.modifiedAfter.getTime()) return false
|
||
if (filter.modifiedBefore && entry.metadata.modified > filter.modifiedBefore.getTime()) return false
|
||
return true
|
||
})
|
||
}
|
||
|
||
private sortDirectoryEntries(entries: VFSEntity[], sort: string, order?: 'asc' | 'desc'): VFSEntity[] {
|
||
const sorted = [...entries].sort((a, b) => {
|
||
let comparison = 0
|
||
switch (sort) {
|
||
case 'name':
|
||
comparison = a.metadata.name.localeCompare(b.metadata.name)
|
||
break
|
||
case 'size':
|
||
comparison = a.metadata.size - b.metadata.size
|
||
break
|
||
case 'modified':
|
||
comparison = a.metadata.modified - b.metadata.modified
|
||
break
|
||
case 'created':
|
||
comparison = a.createdAt - b.createdAt
|
||
break
|
||
}
|
||
return order === 'desc' ? -comparison : comparison
|
||
})
|
||
return sorted
|
||
}
|
||
|
||
private matchGlob(name: string, pattern: string): boolean {
|
||
// Simple glob matching (in production, use proper glob library)
|
||
const regex = pattern
|
||
.replace(/\*/g, '.*')
|
||
.replace(/\?/g, '.')
|
||
return new RegExp(`^${regex}$`).test(name)
|
||
}
|
||
|
||
private invalidateCaches(path: string, recursive = false): void {
|
||
this.contentCache.delete(path)
|
||
this.statCache.delete(path)
|
||
|
||
if (recursive) {
|
||
const prefix = path.endsWith('/') ? path : path + '/'
|
||
for (const cachedPath of this.contentCache.keys()) {
|
||
if (cachedPath.startsWith(prefix)) {
|
||
this.contentCache.delete(cachedPath)
|
||
}
|
||
}
|
||
for (const cachedPath of this.statCache.keys()) {
|
||
if (cachedPath.startsWith(prefix)) {
|
||
this.statCache.delete(cachedPath)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private triggerWatchers(path: string, event: 'rename' | 'change'): void {
|
||
const watchers = this.watchers.get(path)
|
||
if (watchers) {
|
||
for (const listener of watchers) {
|
||
listener(event, path)
|
||
}
|
||
}
|
||
}
|
||
|
||
private async updateChildrenPaths(parentId: string, oldParentPath: string, newParentPath: string): Promise<void> {
|
||
// Get all children recursively
|
||
const children = await this.pathResolver.getChildren(parentId)
|
||
|
||
for (const child of children) {
|
||
const oldChildPath = child.metadata.path as string
|
||
const relativePath = oldChildPath.substring(oldParentPath.length)
|
||
const newChildPath = newParentPath + relativePath
|
||
|
||
// Update child entity — metadata-only (mirrors rename() above). Spreading
|
||
// the whole child forwards its `vector` field into update(), which fails
|
||
// dimension validation when the child was fetched without vectors (and
|
||
// would needlessly touch the vector index when it wasn't).
|
||
await this.brain.update({
|
||
id: child.id,
|
||
metadata: {
|
||
...child.metadata,
|
||
path: newChildPath,
|
||
modified: Date.now()
|
||
}
|
||
})
|
||
|
||
// Update path cache
|
||
this.pathResolver.invalidatePath(oldChildPath)
|
||
await this.pathResolver.createPath(newChildPath, child.id)
|
||
|
||
// Recursively update if it's a directory
|
||
if (child.metadata.vfsType === 'directory') {
|
||
await this.updateChildrenPaths(child.id, oldChildPath, newChildPath)
|
||
}
|
||
}
|
||
}
|
||
|
||
private startBackgroundTasks(): void {
|
||
// Clean up caches periodically
|
||
this.backgroundTimer = setInterval(() => {
|
||
const now = Date.now()
|
||
|
||
// Clean content cache
|
||
for (const [path, entry] of this.contentCache) {
|
||
if (now - entry.timestamp > (this.config.cache?.ttl || 300000)) {
|
||
this.contentCache.delete(path)
|
||
}
|
||
}
|
||
|
||
// Clean stat cache
|
||
for (const [path, entry] of this.statCache) {
|
||
if (now - entry.timestamp > 5000) {
|
||
this.statCache.delete(path)
|
||
}
|
||
}
|
||
}, 60000) // Every minute
|
||
// Cache maintenance must never keep the host process alive.
|
||
if (this.backgroundTimer && typeof this.backgroundTimer.unref === 'function') {
|
||
this.backgroundTimer.unref()
|
||
}
|
||
}
|
||
|
||
private getDefaultConfig(): Required<Omit<VFSConfig, 'rootEntityId'>> & { rootEntityId?: string } {
|
||
return {
|
||
root: '/',
|
||
rootEntityId: undefined,
|
||
cache: {
|
||
enabled: true,
|
||
maxPaths: 100_000,
|
||
maxContent: 100_000_000, // 100MB
|
||
ttl: 5 * 60 * 1000 // 5 minutes
|
||
},
|
||
storage: {
|
||
inline: {
|
||
maxSize: 100_000 // 100KB
|
||
},
|
||
chunking: {
|
||
enabled: true,
|
||
chunkSize: 5_000_000, // 5MB
|
||
parallel: 4
|
||
},
|
||
compression: {
|
||
enabled: true,
|
||
minSize: 10_000, // 10KB
|
||
algorithm: 'gzip'
|
||
}
|
||
},
|
||
intelligence: {
|
||
enabled: true,
|
||
autoEmbed: true,
|
||
autoExtract: true,
|
||
autoTag: false,
|
||
autoConcepts: false
|
||
},
|
||
permissions: {
|
||
defaultFile: 0o644,
|
||
defaultDirectory: 0o755,
|
||
umask: 0o022
|
||
},
|
||
limits: {
|
||
maxFileSize: 1_000_000_000, // 1GB
|
||
maxPathLength: 4096,
|
||
maxDirectoryEntries: 100_000
|
||
}
|
||
}
|
||
}
|
||
|
||
// ============= Lifecycle, POSIX & Extended Operations =============
|
||
|
||
/**
|
||
* Release all resources held by the VFS.
|
||
*
|
||
* Stops background cache eviction, tears down the path resolver, and clears the
|
||
* content cache and watcher registry. After close the VFS is marked
|
||
* uninitialized; call {@link init} again before reusing it.
|
||
*
|
||
* @returns A promise that resolves once cleanup is complete.
|
||
*/
|
||
async close(): Promise<void> {
|
||
// Cleanup PathResolver resources
|
||
if (this.pathResolver) {
|
||
this.pathResolver.cleanup()
|
||
}
|
||
|
||
// Stop background tasks
|
||
if (this.backgroundTimer) {
|
||
clearInterval(this.backgroundTimer)
|
||
this.backgroundTimer = null
|
||
}
|
||
|
||
// Clear caches
|
||
this.contentCache.clear()
|
||
|
||
// Clear watchers
|
||
this.watchers.clear()
|
||
|
||
this.initialized = false
|
||
}
|
||
|
||
/**
|
||
* Change the permission bits of a file or directory (POSIX `chmod`).
|
||
*
|
||
* @param path - The VFS path whose permissions should change.
|
||
* @param mode - The new permission bits (e.g. `0o644`), stored in entity metadata.
|
||
* @returns A promise that resolves once the permissions are persisted.
|
||
*/
|
||
async chmod(path: string, mode: number): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Update permissions in metadata
|
||
await this.brain.update({
|
||
...entity,
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
permissions: mode,
|
||
modified: Date.now()
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
/**
|
||
* Change the owner and group of a file or directory (POSIX `chown`).
|
||
*
|
||
* @param path - The VFS path whose ownership should change.
|
||
* @param uid - The new owner user id, stored in entity metadata.
|
||
* @param gid - The new owning group id, stored in entity metadata.
|
||
* @returns A promise that resolves once the ownership is persisted.
|
||
*/
|
||
async chown(path: string, uid: number, gid: number): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Update ownership in metadata
|
||
await this.brain.update({
|
||
...entity,
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
uid,
|
||
gid,
|
||
modified: Date.now()
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
/**
|
||
* Set the access and modification timestamps of a file or directory (POSIX `utimes`).
|
||
*
|
||
* @param path - The VFS path to update.
|
||
* @param atime - The new access time.
|
||
* @param mtime - The new modification time.
|
||
* @returns A promise that resolves once the timestamps are persisted.
|
||
*/
|
||
async utimes(path: string, atime: Date, mtime: Date): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Update timestamps in metadata
|
||
await this.brain.update({
|
||
...entity,
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
accessed: atime.getTime(),
|
||
modified: mtime.getTime()
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
/**
|
||
* Rename or move a file or directory in place.
|
||
*
|
||
* Updates the entity's path metadata without rewriting content (preserving the
|
||
* underlying blob and entity id). When the parent directory changes, a new
|
||
* `Contains` edge is added to the destination parent. Renaming a directory
|
||
* cascades the path change to every descendant.
|
||
*
|
||
* @param oldPath - The existing VFS path.
|
||
* @param newPath - The destination VFS path; must not already exist.
|
||
* @returns A promise that resolves once the rename is complete.
|
||
* @throws {VFSError} ENOENT when `oldPath` does not exist.
|
||
* @throws {VFSError} EEXIST when `newPath` already exists.
|
||
*/
|
||
async rename(oldPath: string, newPath: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Check if source exists
|
||
const entityId = await this.pathResolver.resolve(oldPath)
|
||
const entity = await this.brain.get(entityId)
|
||
|
||
if (!entity) {
|
||
throw new VFSError(VFSErrorCode.ENOENT, `No such file or directory: ${oldPath}`, oldPath, 'rename')
|
||
}
|
||
|
||
// Check if target already exists
|
||
try {
|
||
await this.pathResolver.resolve(newPath)
|
||
throw new VFSError(VFSErrorCode.EEXIST, `File exists: ${newPath}`, newPath, 'rename')
|
||
} catch (err: any) {
|
||
if (err.code !== VFSErrorCode.ENOENT) throw err
|
||
}
|
||
|
||
// Update parent relationships if needed
|
||
const oldParentPath = this.getParentPath(oldPath)
|
||
const newParentPath = this.getParentPath(newPath)
|
||
|
||
if (oldParentPath !== newParentPath) {
|
||
// Remove from old parent
|
||
if (oldParentPath) {
|
||
const oldParentId = await this.pathResolver.resolve(oldParentPath)
|
||
// unrelate takes the relation ID, not params - need to find and remove relation
|
||
// For now, skip unrelate as it's not critical for rename
|
||
}
|
||
|
||
// Add to new parent
|
||
if (newParentPath && newParentPath !== '/') {
|
||
const newParentId = await this.pathResolver.resolve(newParentPath)
|
||
await this.brain.relate({
|
||
from: newParentId,
|
||
to: entityId,
|
||
type: VerbType.Contains,
|
||
subtype: 'vfs-contains', // Standard subtype for VFS containment edges (7.30+)
|
||
metadata: { isVFS: true } // Mark as VFS relationship
|
||
})
|
||
}
|
||
}
|
||
|
||
// A rename is a path/metadata change, never a content change — issue a
|
||
// metadata-only update. Spreading the whole entity here used to forward
|
||
// its vector field into update(), which failed dimension validation when
|
||
// the entity was fetched without vectors (and would needlessly touch the
|
||
// vector index when it wasn't).
|
||
await this.brain.update({
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
path: newPath,
|
||
name: this.getBasename(newPath),
|
||
modified: Date.now()
|
||
}
|
||
})
|
||
|
||
// Update path cache
|
||
this.pathResolver.invalidatePath(oldPath, true)
|
||
await this.pathResolver.createPath(newPath, entityId)
|
||
|
||
// If it's a directory, update all children paths
|
||
if (entity.metadata.vfsType === 'directory') {
|
||
await this.updateChildrenPaths(entityId, oldPath, newPath)
|
||
}
|
||
|
||
// Trigger watchers
|
||
this.triggerWatchers(oldPath, 'rename')
|
||
this.triggerWatchers(newPath, 'rename')
|
||
}
|
||
|
||
/**
|
||
* Copy a file or directory to a new path.
|
||
*
|
||
* Files are duplicated as new entities (sharing content via the content-addressed
|
||
* blob store); directories are copied recursively. Unless `overwrite` is set, an
|
||
* existing destination is rejected.
|
||
*
|
||
* @param src - The source VFS path to copy from.
|
||
* @param dest - The destination VFS path to copy to.
|
||
* @param options - Copy options such as `overwrite`, `deepCopy`, `preserveVector`,
|
||
* and `preserveRelationships`.
|
||
* @returns A promise that resolves once the copy is complete.
|
||
* @throws {VFSError} ENOENT when `src` does not exist.
|
||
* @throws {VFSError} EEXIST when `dest` exists and `overwrite` is not set.
|
||
*/
|
||
async copy(src: string, dest: string, options?: CopyOptions): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Get source entity
|
||
const srcEntityId = await this.pathResolver.resolve(src)
|
||
const srcEntity = await this.brain.get(srcEntityId)
|
||
|
||
if (!srcEntity) {
|
||
throw new VFSError(VFSErrorCode.ENOENT, `No such file or directory: ${src}`, src, 'copy')
|
||
}
|
||
|
||
// Check if destination already exists
|
||
if (!options?.overwrite) {
|
||
try {
|
||
await this.pathResolver.resolve(dest)
|
||
throw new VFSError(VFSErrorCode.EEXIST, `File exists: ${dest}`, dest, 'copy')
|
||
} catch (err: any) {
|
||
if (err.code !== VFSErrorCode.ENOENT) throw err
|
||
}
|
||
}
|
||
|
||
// Copy the entity
|
||
if (srcEntity.metadata.vfsType === 'file') {
|
||
await this.copyFile(srcEntity, dest, options)
|
||
} else if (srcEntity.metadata.vfsType === 'directory') {
|
||
await this.copyDirectory(src, dest, options)
|
||
}
|
||
}
|
||
|
||
private async copyFile(srcEntity: Entity, destPath: string, options?: CopyOptions): Promise<void> {
|
||
// Create new entity with same content but different path. Preserve the source
|
||
// entity's subtype when it has one (so a vfs-file stays vfs-file); fall back
|
||
// to 'vfs-file' for the rare case of a VFS entity without subtype (pre-7.30
|
||
// legacy data path that hits the copy operation).
|
||
const newEntity = await this.brain.add({
|
||
type: srcEntity.type,
|
||
subtype: srcEntity.subtype ?? 'vfs-file',
|
||
data: srcEntity.data,
|
||
vector: options?.preserveVector ? srcEntity.vector : undefined,
|
||
metadata: {
|
||
...srcEntity.metadata,
|
||
path: destPath,
|
||
name: this.getBasename(destPath),
|
||
created: Date.now(),
|
||
modified: Date.now(),
|
||
copiedFrom: srcEntity.metadata.path
|
||
}
|
||
})
|
||
|
||
// Add to parent directory
|
||
const parentPath = this.getParentPath(destPath)
|
||
if (parentPath && parentPath !== '/') {
|
||
const parentId = await this.pathResolver.resolve(parentPath)
|
||
await this.brain.relate({
|
||
from: parentId,
|
||
to: newEntity,
|
||
type: VerbType.Contains,
|
||
subtype: 'vfs-contains', // Standard subtype for VFS containment edges (7.30+)
|
||
metadata: { isVFS: true } // Mark as VFS relationship
|
||
})
|
||
}
|
||
|
||
// Update path cache
|
||
await this.pathResolver.createPath(destPath, newEntity)
|
||
|
||
// Copy relationships if requested
|
||
if (options?.preserveRelationships) {
|
||
const relations = await this.brain.related({ from: srcEntity.id })
|
||
for (const relation of relations) {
|
||
if (relation.type !== VerbType.Contains) {
|
||
// Skip relationship without Contains type
|
||
// Future: implement proper relation copying
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Copy a directory recursively
|
||
*
|
||
* Optimized for cloud storage using batch operations
|
||
* - Uses gatherDescendants() for efficient graph traversal + batch fetch
|
||
* - Uses addMany() for batch entity creation
|
||
* - Uses relateMany() for batch relationship creation
|
||
*
|
||
* Performance improvement: 3-6x faster on cloud storage (GCS, S3, R2, Azure)
|
||
*/
|
||
private async copyDirectory(srcPath: string, destPath: string, options?: CopyOptions): Promise<void> {
|
||
// Shallow copy - just create directory
|
||
if (options?.deepCopy === false) {
|
||
await this.mkdir(destPath, { recursive: true })
|
||
return
|
||
}
|
||
|
||
// OPTIMIZED: Batch fetch all source entities in ONE call
|
||
const srcEntityId = await this.pathResolver.resolve(srcPath)
|
||
const descendants = await this.gatherDescendants(srcEntityId, Infinity)
|
||
const srcEntity = await this.getEntityById(srcEntityId)
|
||
const allEntities = [srcEntity, ...descendants]
|
||
|
||
// Build path mapping: srcPath -> destPath
|
||
const pathMap = new Map<string, string>()
|
||
const idMap = new Map<string, string>() // old ID -> new ID
|
||
|
||
for (const entity of allEntities) {
|
||
const relativePath = entity.metadata.path.substring(srcPath.length)
|
||
const newPath = destPath + relativePath
|
||
pathMap.set(entity.metadata.path, newPath)
|
||
}
|
||
|
||
// Phase 1: Create all directories first (maintain hierarchy)
|
||
// Sort by path length to ensure parents are created before children
|
||
const directories = allEntities
|
||
.filter(e => e.metadata.vfsType === 'directory')
|
||
.sort((a, b) => a.metadata.path.length - b.metadata.path.length)
|
||
|
||
for (const dir of directories) {
|
||
const newPath = pathMap.get(dir.metadata.path)!
|
||
await this.mkdir(newPath) // mkdir is relatively fast
|
||
const newId = await this.pathResolver.resolve(newPath)
|
||
idMap.set(dir.id, newId)
|
||
}
|
||
|
||
// Phase 2: Batch-create all files using addMany
|
||
const files = allEntities.filter(e => e.metadata.vfsType === 'file')
|
||
|
||
if (files.length > 0) {
|
||
const items = files.map(srcFile => {
|
||
const newPath = pathMap.get(srcFile.metadata.path)!
|
||
|
||
return {
|
||
type: srcFile.type,
|
||
data: srcFile.data,
|
||
vector: options?.preserveVector ? srcFile.vector : undefined,
|
||
metadata: {
|
||
...srcFile.metadata,
|
||
path: newPath,
|
||
name: this.getBasename(newPath),
|
||
parent: undefined, // Will be set via relationship
|
||
created: Date.now(),
|
||
modified: Date.now(),
|
||
copiedFrom: srcFile.metadata.path
|
||
}
|
||
}
|
||
})
|
||
|
||
const result = await this.brain.addMany({ items, continueOnError: false })
|
||
|
||
// Build ID mapping for new files
|
||
for (let i = 0; i < files.length; i++) {
|
||
idMap.set(files[i].id, result.successful[i])
|
||
}
|
||
|
||
// Phase 3: Batch-create parent relationships using relateMany
|
||
const relations = files.map((srcFile, i) => {
|
||
const newPath = pathMap.get(srcFile.metadata.path)!
|
||
const parentPath = this.getParentPath(newPath)
|
||
|
||
// Find parent ID from directories we created
|
||
let parentId: string
|
||
if (parentPath === '/') {
|
||
parentId = VirtualFileSystem.VFS_ROOT_ID
|
||
} else {
|
||
// Find the source directory that maps to this parent path
|
||
const srcParentDir = directories.find(d => pathMap.get(d.metadata.path) === parentPath)
|
||
parentId = srcParentDir ? idMap.get(srcParentDir.id)! : VirtualFileSystem.VFS_ROOT_ID
|
||
}
|
||
|
||
return {
|
||
from: parentId,
|
||
to: result.successful[i],
|
||
type: VerbType.Contains,
|
||
subtype: 'vfs-contains', // Standard subtype for VFS containment edges (7.30+)
|
||
metadata: { isVFS: true }
|
||
}
|
||
})
|
||
|
||
await this.brain.relateMany({ items: relations })
|
||
|
||
// Phase 4: Update path resolver cache for all new files
|
||
for (let i = 0; i < files.length; i++) {
|
||
const newPath = pathMap.get(files[i].metadata.path)!
|
||
await this.pathResolver.createPath(newPath, result.successful[i])
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Move a file or directory to a new path.
|
||
*
|
||
* Implemented as an in-place {@link rename} rather than copy-then-delete: on the
|
||
* content-addressed blob store a copy would share the source content hash, so
|
||
* deleting the source would orphan the destination. Renaming preserves the blob
|
||
* and entity id and handles directory descendants.
|
||
*
|
||
* @param src - The source VFS path to move from.
|
||
* @param dest - The destination VFS path to move to; must not already exist.
|
||
* @returns A promise that resolves once the move is complete.
|
||
* @throws {VFSError} ENOENT when `src` does not exist.
|
||
* @throws {VFSError} EEXIST when `dest` already exists.
|
||
*/
|
||
async move(src: string, dest: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// A move is a RENAME (in-place path change), NOT copy + delete. The old
|
||
// copy+delete path was broken on the content-addressed blob store: copy()
|
||
// makes the destination reference the SAME content-hash as the source, then
|
||
// unlink(src) deletes that shared blob — orphaning the destination
|
||
// ("Blob metadata not found" on the next read). rename() updates the path
|
||
// in place (preserving the blob, keeping the same entity id) and already
|
||
// handles both files and directories, including child path updates.
|
||
await this.rename(src, dest)
|
||
}
|
||
|
||
/**
|
||
* Create a symbolic link at `path` that points to `target` (POSIX `symlink`).
|
||
*
|
||
* The link is stored as a dedicated `vfs-symlink` entity whose
|
||
* `metadata.symlinkTarget` records the target path; it is linked into its parent
|
||
* directory with a `Contains` edge.
|
||
*
|
||
* @param target - The path the symlink should resolve to.
|
||
* @param path - The VFS path at which to create the symlink; must not already exist.
|
||
* @returns A promise that resolves once the symlink is created.
|
||
* @throws {VFSError} EEXIST when `path` already exists.
|
||
*/
|
||
async symlink(target: string, path: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
// Check if symlink already exists
|
||
try {
|
||
await this.pathResolver.resolve(path)
|
||
throw new VFSError(VFSErrorCode.EEXIST, `File exists: ${path}`, path, 'symlink')
|
||
} catch (err: any) {
|
||
if (err.code !== VFSErrorCode.ENOENT) throw err
|
||
}
|
||
|
||
// Parse path to get parent and name
|
||
const parentPath = this.getParentPath(path)
|
||
const name = this.getBasename(path)
|
||
|
||
// Ensure parent directory exists
|
||
const parentId = await this.ensureDirectory(parentPath)
|
||
|
||
// Create symlink entity
|
||
const metadata: VFSMetadata = {
|
||
path,
|
||
name,
|
||
parent: parentId,
|
||
vfsType: 'symlink',
|
||
isVFS: true, // Infrastructure-bypass marker for strict-mode enforcement
|
||
isVFSEntity: true,
|
||
symlinkTarget: target,
|
||
size: 0,
|
||
permissions: 0o777,
|
||
owner: 'user',
|
||
group: 'users',
|
||
accessed: Date.now(),
|
||
modified: Date.now()
|
||
}
|
||
|
||
const entity = await this.brain.add({
|
||
data: `symlink:${target}`,
|
||
type: NounType.File, // Symlinks are special files
|
||
subtype: 'vfs-symlink', // Distinct from 'vfs-file' so consumers can find symlinks (7.30.1+)
|
||
metadata
|
||
})
|
||
|
||
// Create parent-child relationship
|
||
await this.brain.relate({
|
||
from: parentId,
|
||
to: entity,
|
||
type: VerbType.Contains,
|
||
subtype: 'vfs-contains', // Standard subtype for VFS containment edges (7.30+)
|
||
metadata: { isVFS: true } // Mark as VFS relationship
|
||
})
|
||
|
||
// Update path resolver cache
|
||
await this.pathResolver.createPath(path, entity)
|
||
}
|
||
|
||
/**
|
||
* Read the target of a symbolic link (POSIX `readlink`).
|
||
*
|
||
* @param path - The VFS path of the symlink.
|
||
* @returns The stored target path, or an empty string when no target is recorded.
|
||
* @throws {VFSError} EINVAL when `path` is not a symbolic link.
|
||
*/
|
||
async readlink(path: string): Promise<string> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Verify it's a symlink
|
||
if (entity.metadata.vfsType !== 'symlink') {
|
||
throw new VFSError(VFSErrorCode.EINVAL, `Not a symbolic link: ${path}`, path, 'readlink')
|
||
}
|
||
|
||
return entity.metadata.symlinkTarget || ''
|
||
}
|
||
|
||
/**
|
||
* Resolve a path to its canonical form, following symbolic links (POSIX `realpath`).
|
||
*
|
||
* Symlinks are followed iteratively until a non-symlink target is reached, up to a
|
||
* fixed depth limit that guards against link cycles.
|
||
*
|
||
* @param path - The VFS path to resolve.
|
||
* @returns The fully resolved (non-symlink) path.
|
||
* @throws {VFSError} ENOENT when the path (or a link in the chain) does not exist.
|
||
* @throws {VFSError} ELOOP when too many symbolic links are encountered.
|
||
*/
|
||
async realpath(path: string): Promise<string> {
|
||
await this.ensureInitialized()
|
||
|
||
// Resolve symlinks recursively
|
||
let currentPath = path
|
||
let depth = 0
|
||
const maxDepth = 20 // Prevent infinite loops
|
||
|
||
while (depth < maxDepth) {
|
||
try {
|
||
const entityId = await this.pathResolver.resolve(currentPath)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
if (entity.metadata.vfsType === 'symlink') {
|
||
// Follow the symlink
|
||
currentPath = entity.metadata.symlinkTarget || ''
|
||
depth++
|
||
} else {
|
||
// Not a symlink, we have the real path
|
||
return currentPath
|
||
}
|
||
} catch (err) {
|
||
throw new VFSError(VFSErrorCode.ENOENT, `No such file or directory: ${path}`, path, 'realpath')
|
||
}
|
||
}
|
||
|
||
throw new VFSError(VFSErrorCode.ELOOP, `Too many symbolic links: ${path}`, path, 'realpath')
|
||
}
|
||
|
||
/**
|
||
* Read a single extended attribute of a file or directory (POSIX `getxattr`).
|
||
*
|
||
* @param path - The VFS path to read.
|
||
* @param name - The extended-attribute key to fetch.
|
||
* @returns The attribute value, or `undefined` when the attribute is not set.
|
||
*/
|
||
async getxattr(path: string, name: string): Promise<any> {
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
return entity.metadata.attributes?.[name]
|
||
}
|
||
|
||
/**
|
||
* Set a single extended attribute on a file or directory (POSIX `setxattr`).
|
||
*
|
||
* @param path - The VFS path to update.
|
||
* @param name - The extended-attribute key to set.
|
||
* @param value - The value to store under `name`.
|
||
* @returns A promise that resolves once the attribute is persisted.
|
||
*/
|
||
async setxattr(path: string, name: string, value: any): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Create extended attributes object
|
||
const xattrs = entity.metadata.attributes || {}
|
||
xattrs[name] = value
|
||
|
||
// Update entity metadata
|
||
await this.brain.update({
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
attributes: xattrs
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
/**
|
||
* List the extended-attribute names set on a file or directory (POSIX `listxattr`).
|
||
*
|
||
* @param path - The VFS path to inspect.
|
||
* @returns The list of extended-attribute keys (empty when none are set).
|
||
*/
|
||
async listxattr(path: string): Promise<string[]> {
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
return Object.keys(entity.metadata.attributes || {})
|
||
}
|
||
|
||
/**
|
||
* Remove a single extended attribute from a file or directory (POSIX `removexattr`).
|
||
*
|
||
* @param path - The VFS path to update.
|
||
* @param name - The extended-attribute key to remove.
|
||
* @returns A promise that resolves once the attribute is removed.
|
||
*/
|
||
async removexattr(path: string, name: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Remove from extended attributes
|
||
const xattrs = { ...entity.metadata.attributes }
|
||
delete xattrs[name]
|
||
|
||
// Update entity metadata
|
||
await this.brain.update({
|
||
...entity,
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
attributes: xattrs
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
/**
|
||
* List the paths related to a file or directory via graph edges (both directions).
|
||
*
|
||
* Walks outgoing and incoming relationships and maps each connected entity back to
|
||
* its VFS path, recording the relationship type and direction.
|
||
*
|
||
* @param path - The VFS path whose relationships to list.
|
||
* @param options - Reserved relationship-filtering options.
|
||
* @returns Related entries, each with the related `path`, `relationship` type, and
|
||
* `direction` (`'from'` for outgoing edges, `'to'` for incoming).
|
||
*/
|
||
async getRelated(path: string, options?: RelatedOptions): Promise<Array<{
|
||
path: string
|
||
relationship: string
|
||
direction: 'from' | 'to'
|
||
}>> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const results: Array<{ path: string, relationship: string, direction: 'from' | 'to' }> = []
|
||
|
||
// Use proper Brainy relationship API to get all relationships
|
||
const [fromRelations, toRelations] = await Promise.all([
|
||
this.brain.related({ from: entityId }),
|
||
this.brain.related({ to: entityId })
|
||
])
|
||
|
||
// Add outgoing relationships
|
||
for (const rel of fromRelations) {
|
||
const targetEntity = await this.brain.get(rel.to)
|
||
if (targetEntity && targetEntity.metadata?.path) {
|
||
results.push({
|
||
path: targetEntity.metadata.path,
|
||
relationship: rel.type || 'related',
|
||
direction: 'from'
|
||
})
|
||
}
|
||
}
|
||
|
||
// Add incoming relationships
|
||
for (const rel of toRelations) {
|
||
const sourceEntity = await this.brain.get(rel.from)
|
||
if (sourceEntity && sourceEntity.metadata?.path) {
|
||
results.push({
|
||
path: sourceEntity.metadata.path,
|
||
relationship: rel.type || 'related',
|
||
direction: 'to'
|
||
})
|
||
}
|
||
}
|
||
|
||
return results
|
||
}
|
||
|
||
/**
|
||
* Get the non-hierarchical relationships of a file or directory.
|
||
*
|
||
* Returns both outgoing and incoming edges as {@link Relation} objects, excluding
|
||
* the `Contains` edges that model the directory tree itself.
|
||
*
|
||
* @param path - The VFS path whose relationships to return.
|
||
* @returns The list of semantic relationships connected to the path.
|
||
*/
|
||
async getRelationships(path: string): Promise<Relation[]> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const relationships: Relation[] = []
|
||
|
||
// Use proper Brainy relationship API
|
||
const [fromRelations, toRelations] = await Promise.all([
|
||
this.brain.related({ from: entityId }),
|
||
this.brain.related({ to: entityId })
|
||
])
|
||
|
||
// Process outgoing relationships (excluding Contains for parent-child)
|
||
for (const rel of fromRelations) {
|
||
if (rel.type !== VerbType.Contains) { // Skip filesystem hierarchy
|
||
const targetEntity = await this.brain.get(rel.to)
|
||
if (targetEntity && targetEntity.metadata?.path) {
|
||
relationships.push({
|
||
id: rel.id || crypto.randomUUID(),
|
||
from: entityId,
|
||
to: rel.to,
|
||
type: rel.type,
|
||
createdAt: rel.createdAt || Date.now()
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
// Process incoming relationships (excluding Contains for parent-child)
|
||
for (const rel of toRelations) {
|
||
if (rel.type !== VerbType.Contains) { // Skip filesystem hierarchy
|
||
const sourceEntity = await this.brain.get(rel.from)
|
||
if (sourceEntity && sourceEntity.metadata?.path) {
|
||
relationships.push({
|
||
id: rel.id || crypto.randomUUID(),
|
||
from: rel.from,
|
||
to: entityId,
|
||
type: rel.type,
|
||
createdAt: rel.createdAt || Date.now()
|
||
})
|
||
}
|
||
}
|
||
}
|
||
|
||
return relationships
|
||
}
|
||
|
||
/**
|
||
* Create a graph relationship between two VFS paths.
|
||
*
|
||
* @param from - The source VFS path.
|
||
* @param to - The target VFS path.
|
||
* @param type - The relationship (verb) type to create; coerced to a {@link VerbType}.
|
||
* @returns A promise that resolves once the relationship is created.
|
||
*/
|
||
async addRelationship(from: string, to: string, type: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const fromEntityId = await this.pathResolver.resolve(from)
|
||
const toEntityId = await this.pathResolver.resolve(to)
|
||
|
||
// Create relationship using brain
|
||
await this.brain.relate({
|
||
from: fromEntityId,
|
||
to: toEntityId,
|
||
type: type as VerbType, // Convert string to VerbType
|
||
metadata: { isVFS: true } // Mark as VFS relationship
|
||
})
|
||
|
||
// Invalidate caches for both paths
|
||
this.invalidateCaches(from)
|
||
this.invalidateCaches(to)
|
||
}
|
||
|
||
/**
|
||
* Remove a graph relationship between two VFS paths.
|
||
*
|
||
* Deletes outgoing edges from `from` to `to`; when `type` is given, only edges of
|
||
* that relationship type are removed.
|
||
*
|
||
* @param from - The source VFS path.
|
||
* @param to - The target VFS path.
|
||
* @param type - Optional relationship type to match; when omitted, all matching
|
||
* edges to `to` are removed.
|
||
* @returns A promise that resolves once the relationship(s) are removed.
|
||
*/
|
||
async removeRelationship(from: string, to: string, type?: string): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const fromEntityId = await this.pathResolver.resolve(from)
|
||
const toEntityId = await this.pathResolver.resolve(to)
|
||
|
||
// Find and delete the relationship
|
||
const relations = await this.brain.related({ from: fromEntityId })
|
||
for (const relation of relations) {
|
||
if (relation.to === toEntityId && (!type || relation.type === type)) {
|
||
// Delete the relationship using brain.unrelate
|
||
if (relation.id) {
|
||
await this.brain.unrelate(relation.id)
|
||
}
|
||
}
|
||
}
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(from)
|
||
this.invalidateCaches(to)
|
||
}
|
||
|
||
/**
|
||
* Get the todo items attached to a file or directory.
|
||
*
|
||
* @param path - The VFS path whose todos to read.
|
||
* @returns The stored todo list, or `undefined` when none are attached.
|
||
*/
|
||
async getTodos(path: string): Promise<VFSMetadata['todos']> {
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
return entity.metadata.todos
|
||
}
|
||
|
||
/**
|
||
* Replace the entire todo list attached to a file or directory.
|
||
*
|
||
* @param path - The VFS path to update.
|
||
* @param todos - The full set of todo items to store (overwrites any existing list).
|
||
* @returns A promise that resolves once the todos are persisted.
|
||
*/
|
||
async setTodos(path: string, todos: VFSTodo[]): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Update todos in metadata
|
||
await this.brain.update({
|
||
...entity,
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
todos,
|
||
modified: Date.now()
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
/**
|
||
* Append a single todo item to a file or directory.
|
||
*
|
||
* Generates an id when one is not supplied and applies default `priority`
|
||
* (`'medium'`) and `status` (`'pending'`).
|
||
*
|
||
* @param path - The VFS path to attach the todo to.
|
||
* @param todo - The todo to add; `id`, `priority`, and `status` are optional.
|
||
* @returns A promise that resolves once the todo is persisted.
|
||
*/
|
||
async addTodo(path: string, todo: VFSTodo): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Get existing todos
|
||
const todos = entity.metadata.todos || []
|
||
|
||
// Add new todo with ID if not provided
|
||
const newTodo: VFSTodo = {
|
||
id: todo.id || crypto.randomUUID(),
|
||
task: todo.task,
|
||
priority: todo.priority || 'medium',
|
||
status: todo.status || 'pending',
|
||
assignee: todo.assignee,
|
||
due: todo.due
|
||
}
|
||
|
||
todos.push(newTodo)
|
||
|
||
// Update entity metadata
|
||
await this.brain.update({
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
todos
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
/**
|
||
* Get metadata for a file or directory
|
||
*/
|
||
async getMetadata(path: string): Promise<VFSMetadata | undefined> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
return entity.metadata
|
||
}
|
||
|
||
/**
|
||
* Set custom metadata for a file or directory
|
||
* Merges with existing metadata
|
||
*/
|
||
async setMetadata(path: string, metadata: Partial<VFSMetadata>): Promise<void> {
|
||
await this.ensureInitialized()
|
||
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Merge with existing metadata
|
||
await this.brain.update({
|
||
id: entityId,
|
||
metadata: {
|
||
...entity.metadata,
|
||
...metadata,
|
||
modified: Date.now()
|
||
}
|
||
})
|
||
|
||
// Invalidate caches
|
||
this.invalidateCaches(path)
|
||
}
|
||
|
||
|
||
/**
|
||
* Set the current user for tracking who makes changes
|
||
*/
|
||
setUser(username: string): void {
|
||
this.currentUser = username || 'system'
|
||
}
|
||
|
||
/**
|
||
* Get the current user
|
||
*/
|
||
getCurrentUser(): string {
|
||
return this.currentUser
|
||
}
|
||
|
||
|
||
|
||
/**
|
||
* Search for entities with filters
|
||
*/
|
||
async searchEntities(query: {
|
||
type?: string
|
||
name?: string
|
||
where?: Record<string, any>
|
||
limit?: number
|
||
}): Promise<Array<{
|
||
id: string
|
||
path: string
|
||
type: string
|
||
metadata: any
|
||
}>> {
|
||
await this.ensureInitialized()
|
||
|
||
// Build query for brain.find()
|
||
const searchQuery: any = {
|
||
where: {
|
||
...query.where,
|
||
vfsType: 'entity'
|
||
},
|
||
limit: query.limit || 100
|
||
}
|
||
|
||
if (query.type) {
|
||
searchQuery.where.entityType = query.type
|
||
}
|
||
|
||
if (query.name) {
|
||
searchQuery.query = query.name
|
||
}
|
||
|
||
const results = await this.brain.find(searchQuery)
|
||
|
||
return results.map(result => ({
|
||
id: result.id,
|
||
path: result.entity?.metadata?.path || '',
|
||
type: result.entity?.metadata?.type || result.entity?.metadata?.entityType || 'unknown',
|
||
metadata: result.entity?.metadata || {}
|
||
}))
|
||
}
|
||
|
||
/**
|
||
* Sort bulk operations to prevent race conditions
|
||
*
|
||
* Strategy:
|
||
* 1. mkdir operations first, sorted by path depth (shallowest first)
|
||
* 2. Other operations (write, delete, update) after, in original order
|
||
*
|
||
* This ensures parent directories exist before files are written,
|
||
* preventing duplicate entity creation from concurrent mkdir calls.
|
||
*/
|
||
private sortBulkOperations(operations: Array<{
|
||
type: 'write' | 'delete' | 'mkdir' | 'update'
|
||
path: string
|
||
data?: Buffer | string
|
||
options?: any
|
||
}>): Array<typeof operations[number]> {
|
||
const mkdirOps: typeof operations = []
|
||
const otherOps: typeof operations = []
|
||
|
||
for (const op of operations) {
|
||
if (op.type === 'mkdir') {
|
||
mkdirOps.push(op)
|
||
} else {
|
||
otherOps.push(op)
|
||
}
|
||
}
|
||
|
||
// Sort mkdir by path depth (shallowest first)
|
||
mkdirOps.sort((a, b) => {
|
||
const depthA = (a.path.match(/\//g) || []).length
|
||
const depthB = (b.path.match(/\//g) || []).length
|
||
return depthA !== depthB ? depthA - depthB : a.path.localeCompare(b.path)
|
||
})
|
||
|
||
return [...mkdirOps, ...otherOps]
|
||
}
|
||
|
||
/**
|
||
* Bulk write operations for performance
|
||
*
|
||
* Prevents race condition by processing mkdir operations
|
||
* sequentially before parallel batch processing of other operations.
|
||
*/
|
||
async bulkWrite(operations: Array<{
|
||
type: 'write' | 'delete' | 'mkdir' | 'update'
|
||
path: string
|
||
data?: Buffer | string
|
||
options?: any
|
||
}>): Promise<{
|
||
successful: number
|
||
failed: Array<{ operation: any, error: string }>
|
||
}> {
|
||
await this.ensureInitialized()
|
||
|
||
const result = {
|
||
successful: 0,
|
||
failed: [] as Array<{ operation: any, error: string }>
|
||
}
|
||
|
||
// Sort operations: mkdirs first (by depth), then others
|
||
const sortedOps = this.sortBulkOperations(operations)
|
||
|
||
// Separate mkdir operations for sequential processing
|
||
const mkdirOps = sortedOps.filter(op => op.type === 'mkdir')
|
||
const otherOps = sortedOps.filter(op => op.type !== 'mkdir')
|
||
|
||
// Phase 1: Process mkdir operations SEQUENTIALLY
|
||
// This prevents the race condition where parallel mkdir calls
|
||
// create duplicate directory entities due to mutex timing window
|
||
for (const op of mkdirOps) {
|
||
try {
|
||
await this.mkdir(op.path, op.options)
|
||
result.successful++
|
||
} catch (error: any) {
|
||
result.failed.push({
|
||
operation: op,
|
||
error: error.message || 'Unknown error'
|
||
})
|
||
}
|
||
}
|
||
|
||
// Phase 2: Process other operations in parallel batches
|
||
// These can safely run in parallel since parent directories now exist
|
||
const batchSize = 10
|
||
for (let i = 0; i < otherOps.length; i += batchSize) {
|
||
const batch = otherOps.slice(i, i + batchSize)
|
||
|
||
// Process batch in parallel
|
||
const promises = batch.map(async (op) => {
|
||
try {
|
||
switch (op.type) {
|
||
case 'write':
|
||
await this.writeFile(op.path, op.data || '', op.options)
|
||
break
|
||
case 'delete':
|
||
await this.unlink(op.path)
|
||
break
|
||
case 'update': {
|
||
// Update only metadata without changing content
|
||
const entityId = await this.pathResolver.resolve(op.path)
|
||
await this.brain.update({
|
||
id: entityId,
|
||
metadata: op.options?.metadata
|
||
})
|
||
break
|
||
}
|
||
}
|
||
result.successful++
|
||
} catch (error: any) {
|
||
result.failed.push({
|
||
operation: op,
|
||
error: error.message || 'Unknown error'
|
||
})
|
||
}
|
||
})
|
||
|
||
await Promise.all(promises)
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* Calculate disk usage for a path (POSIX du command)
|
||
* Returns total bytes used by files in directory tree
|
||
*
|
||
* @param path - Path to calculate usage for
|
||
* @param options - Options including maxDepth for safety
|
||
*/
|
||
async du(path: string = '/', options?: {
|
||
maxDepth?: number
|
||
humanReadable?: boolean
|
||
}): Promise<{
|
||
bytes: number
|
||
files: number
|
||
directories: number
|
||
formatted?: string
|
||
}> {
|
||
await this.ensureInitialized()
|
||
|
||
const maxDepth = options?.maxDepth ?? 100 // Safety limit
|
||
let totalBytes = 0
|
||
let fileCount = 0
|
||
let dirCount = 0
|
||
|
||
const traverse = async (currentPath: string, depth: number) => {
|
||
if (depth > maxDepth) {
|
||
throw new Error(`Maximum depth ${maxDepth} exceeded. Use maxDepth option to increase limit.`)
|
||
}
|
||
|
||
try {
|
||
const entityId = await this.pathResolver.resolve(currentPath)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
if (entity.metadata.vfsType === 'directory') {
|
||
dirCount++
|
||
const children = await this.readdir(currentPath)
|
||
for (const child of children) {
|
||
const childPath = currentPath === '/' ? `/${child}` : `${currentPath}/${child}`
|
||
await traverse(childPath, depth + 1)
|
||
}
|
||
} else if (entity.metadata.vfsType === 'file') {
|
||
fileCount++
|
||
totalBytes += entity.metadata.size || 0
|
||
}
|
||
} catch (error) {
|
||
// Skip inaccessible paths
|
||
}
|
||
}
|
||
|
||
await traverse(path, 0)
|
||
|
||
const result: any = {
|
||
bytes: totalBytes,
|
||
files: fileCount,
|
||
directories: dirCount
|
||
}
|
||
|
||
if (options?.humanReadable) {
|
||
const units = ['B', 'KB', 'MB', 'GB', 'TB']
|
||
let size = totalBytes
|
||
let unitIndex = 0
|
||
while (size >= 1024 && unitIndex < units.length - 1) {
|
||
size /= 1024
|
||
unitIndex++
|
||
}
|
||
result.formatted = `${size.toFixed(2)} ${units[unitIndex]}`
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
/**
|
||
* Check file access permissions (POSIX access command)
|
||
* Verifies if path exists and is accessible with specified mode
|
||
*
|
||
* @param path - Path to check
|
||
* @param mode - Access mode: 'r' (read), 'w' (write), 'x' (execute), or 'f' (exists only)
|
||
*/
|
||
async access(path: string, mode: 'r' | 'w' | 'x' | 'f' = 'f'): Promise<boolean> {
|
||
await this.ensureInitialized()
|
||
|
||
try {
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
// Path exists
|
||
if (mode === 'f') {
|
||
return true
|
||
}
|
||
|
||
// Check permissions based on mode
|
||
const permissions = entity.metadata.permissions || 0o644
|
||
|
||
switch (mode) {
|
||
case 'r':
|
||
// Check read permission (owner, group, or other)
|
||
return (permissions & 0o444) !== 0
|
||
case 'w':
|
||
// Check write permission
|
||
return (permissions & 0o222) !== 0
|
||
case 'x':
|
||
// Check execute permission (only meaningful for directories)
|
||
return entity.metadata.vfsType === 'directory' || (permissions & 0o111) !== 0
|
||
default:
|
||
return false
|
||
}
|
||
} catch (error) {
|
||
// Path doesn't exist or not accessible
|
||
return false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Find files matching patterns (Unix find command)
|
||
* Pattern-based file search (complements semantic search())
|
||
*
|
||
* @param path - Starting path for search
|
||
* @param options - Search options including pattern matching
|
||
*/
|
||
async find(path: string = '/', options?: {
|
||
name?: string | RegExp
|
||
type?: 'file' | 'directory' | 'both'
|
||
maxDepth?: number
|
||
minSize?: number
|
||
maxSize?: number
|
||
modified?: { after?: Date, before?: Date }
|
||
limit?: number
|
||
}): Promise<Array<{
|
||
path: string
|
||
type: 'file' | 'directory'
|
||
size?: number
|
||
modified?: Date
|
||
}>> {
|
||
await this.ensureInitialized()
|
||
|
||
const maxDepth = options?.maxDepth ?? 100 // Safety limit
|
||
const limit = options?.limit ?? 1000 // Prevent unbounded results
|
||
const results: Array<{
|
||
path: string
|
||
type: 'file' | 'directory'
|
||
size?: number
|
||
modified?: Date
|
||
}> = []
|
||
|
||
const namePattern = options?.name
|
||
const nameRegex = namePattern instanceof RegExp
|
||
? namePattern
|
||
: namePattern
|
||
? new RegExp(namePattern.replace(/\*/g, '.*').replace(/\?/g, '.'))
|
||
: null
|
||
|
||
const traverse = async (currentPath: string, depth: number) => {
|
||
if (depth > maxDepth || results.length >= limit) {
|
||
return
|
||
}
|
||
|
||
try {
|
||
const entityId = await this.pathResolver.resolve(currentPath)
|
||
const entity = await this.getEntityById(entityId)
|
||
|
||
const vfsType = entity.metadata.vfsType
|
||
const fileName = currentPath.split('/').pop() || ''
|
||
|
||
// Check if this file matches criteria
|
||
let matches = true
|
||
|
||
// Type filter
|
||
if (options?.type && options.type !== 'both') {
|
||
matches = matches && vfsType === options.type
|
||
}
|
||
|
||
// Name pattern filter
|
||
if (nameRegex) {
|
||
matches = matches && nameRegex.test(fileName)
|
||
}
|
||
|
||
// Size filters (files only)
|
||
if (vfsType === 'file') {
|
||
const size = entity.metadata.size || 0
|
||
if (options?.minSize !== undefined) {
|
||
matches = matches && size >= options.minSize
|
||
}
|
||
if (options?.maxSize !== undefined) {
|
||
matches = matches && size <= options.maxSize
|
||
}
|
||
}
|
||
|
||
// Modified time filter
|
||
if (options?.modified && entity.metadata.modified) {
|
||
const modifiedTime = new Date(entity.metadata.modified)
|
||
if (options.modified.after) {
|
||
matches = matches && modifiedTime >= options.modified.after
|
||
}
|
||
if (options.modified.before) {
|
||
matches = matches && modifiedTime <= options.modified.before
|
||
}
|
||
}
|
||
|
||
// Add to results if matches
|
||
if (matches && currentPath !== path) {
|
||
results.push({
|
||
path: currentPath,
|
||
type: vfsType as 'file' | 'directory',
|
||
size: entity.metadata.size,
|
||
modified: entity.metadata.modified ? new Date(entity.metadata.modified) : undefined
|
||
})
|
||
}
|
||
|
||
// Recurse into directories
|
||
if (vfsType === 'directory' && results.length < limit) {
|
||
const children = await this.readdir(currentPath)
|
||
for (const child of children) {
|
||
if (results.length >= limit) break
|
||
const childPath = currentPath === '/' ? `/${child}` : `${currentPath}/${child}`
|
||
await traverse(childPath, depth + 1)
|
||
}
|
||
}
|
||
} catch (error) {
|
||
// Skip inaccessible paths
|
||
}
|
||
}
|
||
|
||
await traverse(path, 0)
|
||
return results
|
||
}
|
||
|
||
/**
|
||
* Open a file as a Node-compatible readable stream.
|
||
*
|
||
* @param path - The VFS path of the file to read.
|
||
* @param options - Stream options (e.g. byte range, chunk size).
|
||
* @returns A readable stream that yields the file's bytes.
|
||
*/
|
||
createReadStream(path: string, options?: ReadStreamOptions): NodeJS.ReadableStream {
|
||
// Lazy import to avoid circular dependencies
|
||
const { VFSReadStream } = require('./streams/VFSReadStream.js')
|
||
return new VFSReadStream(this, path, options)
|
||
}
|
||
|
||
/**
|
||
* Open a file as a Node-compatible writable stream.
|
||
*
|
||
* @param path - The VFS path of the file to write.
|
||
* @param options - Stream options controlling how buffered data is flushed.
|
||
* @returns A writable stream whose contents are persisted to the file on finish.
|
||
*/
|
||
createWriteStream(path: string, options?: WriteStreamOptions): NodeJS.WritableStream {
|
||
// Lazy import to avoid circular dependencies
|
||
const { VFSWriteStream } = require('./streams/VFSWriteStream.js')
|
||
return new VFSWriteStream(this, path, options)
|
||
}
|
||
|
||
/**
|
||
* Watch a path for changes and invoke a listener on filesystem events.
|
||
*
|
||
* @param path - The VFS path to watch.
|
||
* @param listener - Called with the event type (`'rename'` or `'change'`) and the path.
|
||
* @returns A handle whose `close()` method deregisters the listener.
|
||
*/
|
||
watch(path: string, listener: WatchListener): { close(): void } {
|
||
if (!this.watchers.has(path)) {
|
||
this.watchers.set(path, new Set())
|
||
}
|
||
this.watchers.get(path)!.add(listener)
|
||
|
||
return {
|
||
close: () => {
|
||
const watchers = this.watchers.get(path)
|
||
if (watchers) {
|
||
watchers.delete(listener)
|
||
if (watchers.size === 0) {
|
||
this.watchers.delete(path)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// ============= Import/Export Operations =============
|
||
|
||
/**
|
||
* Import a single file from the real filesystem into VFS
|
||
*/
|
||
async importFile(sourcePath: string, targetPath: string): Promise<void> {
|
||
const fs = await import('fs/promises')
|
||
const pathModule = await import('path')
|
||
|
||
// Read file from local filesystem
|
||
const content = await fs.readFile(sourcePath)
|
||
const stats = await fs.stat(sourcePath)
|
||
|
||
// Ensure parent directory exists in VFS
|
||
const parentPath = pathModule.dirname(targetPath)
|
||
if (parentPath !== '/' && parentPath !== '.') {
|
||
try {
|
||
await this.mkdir(parentPath, { recursive: true })
|
||
} catch (error: any) {
|
||
if (error.code !== 'EEXIST') throw error
|
||
}
|
||
}
|
||
|
||
// Write to VFS with metadata from source
|
||
await this.writeFile(targetPath, content, {
|
||
metadata: {
|
||
imported: true,
|
||
importedFrom: sourcePath,
|
||
sourceSize: stats.size,
|
||
sourceMtime: stats.mtime.getTime(),
|
||
sourceMode: stats.mode
|
||
}
|
||
})
|
||
}
|
||
|
||
/**
|
||
* Import a directory from the real filesystem into VFS
|
||
*/
|
||
async importDirectory(sourcePath: string, options?: any): Promise<any> {
|
||
const { DirectoryImporter } = await import('./importers/DirectoryImporter.js')
|
||
const importer = new DirectoryImporter(this, this.brain)
|
||
return await importer.import(sourcePath, options)
|
||
}
|
||
|
||
/**
|
||
* Import a directory with progress tracking
|
||
*/
|
||
async *importStream(sourcePath: string, options?: any): AsyncGenerator<any> {
|
||
const { DirectoryImporter } = await import('./importers/DirectoryImporter.js')
|
||
const importer = new DirectoryImporter(this, this.brain)
|
||
yield* importer.importStream(sourcePath, options)
|
||
}
|
||
|
||
/**
|
||
* Register a change listener for a path (Node `fs.watchFile`-style convenience).
|
||
*
|
||
* Delegates to {@link watch} without returning the watcher handle; remove the
|
||
* listener with {@link unwatchFile}.
|
||
*
|
||
* @param path - The VFS path to watch.
|
||
* @param listener - Called with the event type and path on each change.
|
||
*/
|
||
watchFile(path: string, listener: WatchListener): void {
|
||
this.watch(path, listener)
|
||
}
|
||
|
||
/**
|
||
* Stop watching a path, removing all listeners registered for it.
|
||
*
|
||
* @param path - The VFS path to stop watching.
|
||
*/
|
||
unwatchFile(path: string): void {
|
||
this.watchers.delete(path)
|
||
}
|
||
|
||
/**
|
||
* Resolve a path and return its backing {@link VFSEntity}.
|
||
*
|
||
* @param path - The VFS path to resolve.
|
||
* @returns The entity (with normalized VFS metadata) for the path.
|
||
* @throws {VFSError} ENOENT when the path cannot be resolved.
|
||
*/
|
||
async getEntity(path: string): Promise<VFSEntity> {
|
||
const entityId = await this.pathResolver.resolve(path)
|
||
return this.getEntityById(entityId)
|
||
}
|
||
|
||
/**
|
||
* Resolve a path to its normalized form
|
||
* Returns the normalized absolute path (e.g., '/foo/bar/file.txt')
|
||
*/
|
||
async resolvePath(path: string, from?: string): Promise<string> {
|
||
// Handle relative paths
|
||
if (!path.startsWith('/') && from) {
|
||
path = `${from}/${path}`
|
||
}
|
||
|
||
// Normalize path: remove multiple slashes, trailing slashes
|
||
return path.replace(/\/+/g, '/').replace(/\/$/, '') || '/'
|
||
}
|
||
|
||
/**
|
||
* Resolve a path to its entity ID
|
||
* Returns the UUID of the entity representing this path
|
||
*/
|
||
async resolvePathToId(path: string, from?: string): Promise<string> {
|
||
// Handle relative paths
|
||
if (!path.startsWith('/') && from) {
|
||
path = `${from}/${path}`
|
||
}
|
||
|
||
// Normalize path
|
||
const normalizedPath = path.replace(/\/+/g, '/').replace(/\/$/, '') || '/'
|
||
|
||
// Special case for root
|
||
if (normalizedPath === '/') {
|
||
return this.rootEntityId!
|
||
}
|
||
|
||
// Resolve the path to an entity ID
|
||
return await this.pathResolver.resolve(normalizedPath)
|
||
}
|
||
} |