refactor(8.0): remove dead, unreachable, and unwired modules
Pre-GA dead-code sweep. A deterministic import-reachability walk from the
package's real entry points (the exports map, the CLI bin, the conversation
surface) found 34 source modules that nothing reachable imports — they
compile and ship as dist output that no consumer or internal path can ever
reach. All were superseded duplicates, abandoned parallel implementations,
or built-but-never-wired features:
- superseded duplicates of live modules: an older "unified" entry, a
standalone neural-import variant, a static NLP processor + its matcher,
a parallel API-types module, a duplicate progress-types module
- an abandoned import path (orchestrator + entity deduplicator + barrels)
left behind when ingestion moved to the coordinator
- unwired feature modules (instance pool, import presets, cached
embeddings, relationship-confidence scorer) reachable only from tests
- dead leaf utilities (write buffer, deleted-items index, bounded
registry, two crypto shims, a cache manager, a structured logger, a
stale v5 type-migration helper, a browser-only FS type shim) and
several dead re-export barrels
- a CLI catalog command wired into no command
Also removed two tests that only exercised deleted modules, repointed two
VFS tests off a deleted barrel onto the implementation module, and deleted
one example that demoed removed features.
Verified: clean build (both tsc passes), unit 1505 + integration 607 green,
and a re-run of the reachability walk reports zero remaining orphans.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
03d654061f
commit
bf0afe8563
39 changed files with 2 additions and 10497 deletions
|
|
@ -1,62 +0,0 @@
|
|||
/**
|
||||
* Bounded Registry with LRU eviction
|
||||
* Prevents unbounded memory growth in production
|
||||
*/
|
||||
export class BoundedRegistry<T> {
|
||||
private items = new Map<T, number>() // item -> last accessed timestamp
|
||||
private readonly maxSize: number
|
||||
|
||||
constructor(maxSize: number = 10000) {
|
||||
this.maxSize = maxSize
|
||||
}
|
||||
|
||||
/**
|
||||
* Add item to registry, evicting oldest if at capacity
|
||||
*/
|
||||
add(item: T): void {
|
||||
// Update timestamp if already exists
|
||||
if (this.items.has(item)) {
|
||||
this.items.delete(item) // Remove to re-add at end
|
||||
this.items.set(item, Date.now())
|
||||
return
|
||||
}
|
||||
|
||||
// Evict oldest if at capacity
|
||||
if (this.items.size >= this.maxSize) {
|
||||
const oldest = this.items.entries().next().value
|
||||
if (oldest) {
|
||||
this.items.delete(oldest[0])
|
||||
}
|
||||
}
|
||||
|
||||
this.items.set(item, Date.now())
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if item exists
|
||||
*/
|
||||
has(item: T): boolean {
|
||||
return this.items.has(item)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all items
|
||||
*/
|
||||
getAll(): T[] {
|
||||
return Array.from(this.items.keys())
|
||||
}
|
||||
|
||||
/**
|
||||
* Get size
|
||||
*/
|
||||
get size(): number {
|
||||
return this.items.size
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all items
|
||||
*/
|
||||
clear(): void {
|
||||
this.items.clear()
|
||||
}
|
||||
}
|
||||
|
|
@ -1,47 +0,0 @@
|
|||
/**
|
||||
* Cross-platform crypto utilities
|
||||
* Provides hashing functions that work in both Node.js and browser environments
|
||||
*/
|
||||
|
||||
/**
|
||||
* Simple string hash function that works in all environments
|
||||
* Uses djb2 algorithm - fast and good distribution
|
||||
* @param str - String to hash
|
||||
* @returns Positive integer hash
|
||||
*/
|
||||
export function hashString(str: string): number {
|
||||
let hash = 5381
|
||||
for (let i = 0; i < str.length; i++) {
|
||||
const char = str.charCodeAt(i)
|
||||
hash = ((hash << 5) + hash) + char // hash * 33 + char
|
||||
}
|
||||
// Ensure positive number
|
||||
return Math.abs(hash)
|
||||
}
|
||||
|
||||
/**
|
||||
* Alternative: FNV-1a hash algorithm
|
||||
* Good distribution and fast
|
||||
* @param str - String to hash
|
||||
* @returns Positive integer hash
|
||||
*/
|
||||
export function fnv1aHash(str: string): number {
|
||||
let hash = 2166136261
|
||||
for (let i = 0; i < str.length; i++) {
|
||||
hash ^= str.charCodeAt(i)
|
||||
hash = (hash * 16777619) >>> 0
|
||||
}
|
||||
return hash
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate a deterministic hash for partitioning
|
||||
* Uses the most appropriate algorithm for the environment
|
||||
* @param input - Input string to hash
|
||||
* @returns Positive integer hash suitable for modulo operations
|
||||
*/
|
||||
export function getPartitionHash(input: string): number {
|
||||
// Use djb2 by default as it's fast and has good distribution
|
||||
// This ensures consistent partitioning across all environments
|
||||
return hashString(input)
|
||||
}
|
||||
|
|
@ -1,106 +0,0 @@
|
|||
/**
|
||||
* Dedicated index for tracking soft-deleted items
|
||||
* This is MUCH more efficient than checking every item in the database
|
||||
*
|
||||
* Performance characteristics:
|
||||
* - Add deleted item: O(1)
|
||||
* - Remove deleted item: O(1)
|
||||
* - Check if deleted: O(1)
|
||||
* - Get all deleted: O(d) where d = number of deleted items << total items
|
||||
*/
|
||||
|
||||
export class DeletedItemsIndex {
|
||||
private deletedIds: Set<string> = new Set()
|
||||
private deletedCount: number = 0
|
||||
|
||||
/**
|
||||
* Mark an item as deleted
|
||||
*/
|
||||
markDeleted(id: string): void {
|
||||
if (!this.deletedIds.has(id)) {
|
||||
this.deletedIds.add(id)
|
||||
this.deletedCount++
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark an item as not deleted (restored)
|
||||
*/
|
||||
markRestored(id: string): void {
|
||||
if (this.deletedIds.delete(id)) {
|
||||
this.deletedCount--
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if an item is deleted - O(1)
|
||||
*/
|
||||
isDeleted(id: string): boolean {
|
||||
return this.deletedIds.has(id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all deleted item IDs - O(d)
|
||||
*/
|
||||
getAllDeleted(): string[] {
|
||||
return Array.from(this.deletedIds)
|
||||
}
|
||||
|
||||
/**
|
||||
* Filter out deleted items from results - O(k) where k = result count
|
||||
*/
|
||||
filterDeleted<T extends { id?: string }>(items: T[]): T[] {
|
||||
if (this.deletedCount === 0) {
|
||||
// Fast path - no deleted items
|
||||
return items
|
||||
}
|
||||
|
||||
return items.filter(item => {
|
||||
const id = item.id
|
||||
return id ? !this.deletedIds.has(id) : true
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Get statistics
|
||||
*/
|
||||
getStats() {
|
||||
return {
|
||||
deletedCount: this.deletedCount,
|
||||
memoryUsage: this.deletedCount * 100 // Rough estimate: 100 bytes per ID
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all deleted items (for testing)
|
||||
*/
|
||||
clear(): void {
|
||||
this.deletedIds.clear()
|
||||
this.deletedCount = 0
|
||||
}
|
||||
|
||||
/**
|
||||
* Serialize for persistence
|
||||
*/
|
||||
serialize(): string {
|
||||
return JSON.stringify(Array.from(this.deletedIds))
|
||||
}
|
||||
|
||||
/**
|
||||
* Deserialize from persistence
|
||||
*/
|
||||
deserialize(data: string): void {
|
||||
try {
|
||||
const ids = JSON.parse(data)
|
||||
this.deletedIds = new Set(ids)
|
||||
this.deletedCount = this.deletedIds.size
|
||||
} catch (e) {
|
||||
console.warn('Failed to deserialize deleted items index')
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Global singleton for deleted items tracking
|
||||
*/
|
||||
export const deletedItemsIndex = new DeletedItemsIndex()
|
||||
|
|
@ -1,88 +0,0 @@
|
|||
/**
|
||||
* Utility to ensure all metadata has the deleted field set properly
|
||||
* This is CRITICAL for O(1) soft delete filtering performance
|
||||
*
|
||||
* Uses _brainy namespace to avoid conflicts with user metadata
|
||||
*/
|
||||
|
||||
const BRAINY_NAMESPACE = '_brainy'
|
||||
|
||||
/**
|
||||
* Ensure metadata has internal Brainy fields set
|
||||
* @param metadata The metadata object (could be null/undefined)
|
||||
* @param preserveExisting If true, preserve existing deleted value
|
||||
* @returns Metadata with internal fields guaranteed
|
||||
*/
|
||||
export function ensureDeletedField(metadata: any, preserveExisting: boolean = true): any {
|
||||
// Handle null/undefined metadata
|
||||
if (!metadata) {
|
||||
return {
|
||||
[BRAINY_NAMESPACE]: {
|
||||
deleted: false,
|
||||
version: 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Clone to avoid mutation
|
||||
const result = { ...metadata }
|
||||
|
||||
// Ensure _brainy namespace exists
|
||||
if (!result[BRAINY_NAMESPACE]) {
|
||||
result[BRAINY_NAMESPACE] = {}
|
||||
}
|
||||
|
||||
// Set deleted field if not present
|
||||
if (!('deleted' in result[BRAINY_NAMESPACE])) {
|
||||
result[BRAINY_NAMESPACE].deleted = false
|
||||
} else if (!preserveExisting) {
|
||||
// Force to false if not preserving
|
||||
result[BRAINY_NAMESPACE].deleted = false
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark an item as soft deleted
|
||||
* @param metadata The metadata object
|
||||
* @returns Metadata with _brainy.deleted=true
|
||||
*/
|
||||
export function markAsDeleted(metadata: any): any {
|
||||
const result = ensureDeletedField(metadata)
|
||||
result[BRAINY_NAMESPACE].deleted = true
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark an item as restored (not deleted)
|
||||
* @param metadata The metadata object
|
||||
* @returns Metadata with _brainy.deleted=false
|
||||
*/
|
||||
export function markAsRestored(metadata: any): any {
|
||||
const result = ensureDeletedField(metadata)
|
||||
result[BRAINY_NAMESPACE].deleted = false
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if an item is deleted
|
||||
* @param metadata The metadata object
|
||||
* @returns true if deleted, false otherwise (including if field missing)
|
||||
*/
|
||||
export function isDeleted(metadata: any): boolean {
|
||||
return metadata?.[BRAINY_NAMESPACE]?.deleted === true
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if an item is active (not deleted)
|
||||
* @param metadata The metadata object
|
||||
* @returns true if not deleted (default), false if deleted
|
||||
*/
|
||||
export function isActive(metadata: any): boolean {
|
||||
// If no deleted field or deleted=false, item is active
|
||||
return !isDeleted(metadata)
|
||||
}
|
||||
|
||||
// Export the namespace constant for use in queries
|
||||
export const BRAINY_DELETED_FIELD = `${BRAINY_NAMESPACE}.deleted`
|
||||
|
|
@ -1,255 +0,0 @@
|
|||
/**
|
||||
* Clean Metadata Architecture for Brainy 2.2
|
||||
* No backward compatibility - doing it RIGHT from the start!
|
||||
*/
|
||||
|
||||
// Namespace constants
|
||||
export const BRAINY_NS = '_brainy' as const
|
||||
export const AUG_NS = '_augmentations' as const
|
||||
export const AUDIT_NS = '_audit' as const
|
||||
|
||||
// Field paths for O(1) indexing
|
||||
export const DELETED_FIELD = `${BRAINY_NS}.deleted` as const
|
||||
export const INDEXED_FIELD = `${BRAINY_NS}.indexed` as const
|
||||
export const VERSION_FIELD = `${BRAINY_NS}.version` as const
|
||||
|
||||
/**
|
||||
* Internal Brainy metadata structure
|
||||
* These fields are ALWAYS present and indexed for O(1) access
|
||||
*/
|
||||
export interface BrainyInternalMetadata {
|
||||
deleted: boolean // ALWAYS boolean, enables O(1) soft delete
|
||||
indexed: boolean // Whether in search index
|
||||
version: number // Schema version
|
||||
created: number // Unix timestamp
|
||||
updated: number // Unix timestamp
|
||||
|
||||
// Optional internal fields
|
||||
domain?: string // Domain classification
|
||||
priority?: number // Query priority hint
|
||||
ttl?: number // Time to live
|
||||
}
|
||||
|
||||
/**
|
||||
* Complete metadata structure with namespaces
|
||||
*/
|
||||
export interface NamespacedMetadata<T = any> {
|
||||
// User metadata - any fields they want
|
||||
[key: string]: any
|
||||
|
||||
// Internal metadata - our fields
|
||||
[BRAINY_NS]: BrainyInternalMetadata
|
||||
|
||||
// Augmentation metadata - isolated per augmentation
|
||||
[AUG_NS]?: {
|
||||
[augmentationName: string]: any
|
||||
}
|
||||
|
||||
// Audit trail - optional
|
||||
[AUDIT_NS]?: Array<{
|
||||
timestamp: number
|
||||
augmentation: string
|
||||
field: string
|
||||
oldValue: any
|
||||
newValue: any
|
||||
}>
|
||||
}
|
||||
|
||||
/**
|
||||
* Create properly namespaced metadata
|
||||
* This is called for EVERY noun/verb creation
|
||||
*/
|
||||
export function createNamespacedMetadata<T = any>(
|
||||
userMetadata?: T
|
||||
): NamespacedMetadata<T> {
|
||||
const now = Date.now()
|
||||
|
||||
// Start with user metadata or empty object
|
||||
const result: any = userMetadata ? { ...userMetadata } : {}
|
||||
|
||||
// ALWAYS add internal namespace with required fields
|
||||
result[BRAINY_NS] = {
|
||||
deleted: false, // CRITICAL: Always false for new items
|
||||
indexed: true, // New items are indexed
|
||||
version: 1, // Current schema version
|
||||
created: now,
|
||||
updated: now
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Update metadata while preserving namespaces
|
||||
*/
|
||||
export function updateNamespacedMetadata<T = any>(
|
||||
existing: NamespacedMetadata<T>,
|
||||
updates: Partial<T>
|
||||
): NamespacedMetadata<T> {
|
||||
const now = Date.now()
|
||||
|
||||
// Merge user fields
|
||||
const result: any = {
|
||||
...existing,
|
||||
...updates
|
||||
}
|
||||
|
||||
// Preserve internal namespace but update timestamp
|
||||
result[BRAINY_NS] = {
|
||||
...existing[BRAINY_NS],
|
||||
updated: now
|
||||
}
|
||||
|
||||
// Preserve augmentation namespace
|
||||
if (existing[AUG_NS]) {
|
||||
result[AUG_NS] = existing[AUG_NS]
|
||||
}
|
||||
|
||||
// Preserve audit trail
|
||||
if (existing[AUDIT_NS]) {
|
||||
result[AUDIT_NS] = existing[AUDIT_NS]
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Soft delete a noun (O(1) operation)
|
||||
*/
|
||||
export function markDeleted<T = any>(
|
||||
metadata: NamespacedMetadata<T>
|
||||
): NamespacedMetadata<T> {
|
||||
return {
|
||||
...metadata,
|
||||
[BRAINY_NS]: {
|
||||
...metadata[BRAINY_NS],
|
||||
deleted: true,
|
||||
updated: Date.now()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Restore a soft-deleted noun (O(1) operation)
|
||||
*/
|
||||
export function markRestored<T = any>(
|
||||
metadata: NamespacedMetadata<T>
|
||||
): NamespacedMetadata<T> {
|
||||
return {
|
||||
...metadata,
|
||||
[BRAINY_NS]: {
|
||||
...metadata[BRAINY_NS],
|
||||
deleted: false,
|
||||
updated: Date.now()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a noun is deleted (O(1) check)
|
||||
*/
|
||||
export function isDeleted<T = any>(
|
||||
metadata: NamespacedMetadata<T>
|
||||
): boolean {
|
||||
return metadata[BRAINY_NS]?.deleted === true
|
||||
}
|
||||
|
||||
/**
|
||||
* Get user metadata without internal fields
|
||||
* Used by augmentations to get clean user data
|
||||
*/
|
||||
export function getUserMetadata<T = any>(
|
||||
metadata: NamespacedMetadata<T>
|
||||
): T {
|
||||
const { [BRAINY_NS]: _, [AUG_NS]: __, [AUDIT_NS]: ___, ...userMeta } = metadata
|
||||
return userMeta as T
|
||||
}
|
||||
|
||||
/**
|
||||
* Set augmentation data in isolated namespace
|
||||
*/
|
||||
export function setAugmentationData<T = any>(
|
||||
metadata: NamespacedMetadata<T>,
|
||||
augmentationName: string,
|
||||
data: any
|
||||
): NamespacedMetadata<T> {
|
||||
const result = { ...metadata }
|
||||
|
||||
if (!result[AUG_NS]) {
|
||||
result[AUG_NS] = {}
|
||||
}
|
||||
|
||||
result[AUG_NS][augmentationName] = data
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Add audit entry for tracking
|
||||
*/
|
||||
export function addAuditEntry<T = any>(
|
||||
metadata: NamespacedMetadata<T>,
|
||||
entry: {
|
||||
augmentation: string
|
||||
field: string
|
||||
oldValue: any
|
||||
newValue: any
|
||||
}
|
||||
): NamespacedMetadata<T> {
|
||||
const result = { ...metadata }
|
||||
|
||||
if (!result[AUDIT_NS]) {
|
||||
result[AUDIT_NS] = []
|
||||
}
|
||||
|
||||
result[AUDIT_NS].push({
|
||||
...entry,
|
||||
timestamp: Date.now()
|
||||
})
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* INDEXING EXPLANATION:
|
||||
*
|
||||
* The MetadataIndex flattens nested objects into dot-notation keys:
|
||||
*
|
||||
* Input metadata:
|
||||
* {
|
||||
* name: "Django",
|
||||
* _brainy: {
|
||||
* deleted: false,
|
||||
* indexed: true
|
||||
* }
|
||||
* }
|
||||
*
|
||||
* Creates index entries:
|
||||
* - "name" -> "django" -> Set([id1, id2...])
|
||||
* - "_brainy.deleted" -> "false" -> Set([id1, id2...]) // O(1) lookup!
|
||||
* - "_brainy.indexed" -> "true" -> Set([id1, id2...])
|
||||
*
|
||||
* Query: { "_brainy.deleted": false }
|
||||
* Lookup: index["_brainy.deleted"]["false"] -> Set of IDs in O(1)
|
||||
*
|
||||
* This is why namespacing doesn't hurt performance - it's all flattened!
|
||||
*/
|
||||
|
||||
/**
|
||||
* Fields that should ALWAYS be indexed for O(1) access
|
||||
*/
|
||||
export const ALWAYS_INDEXED_FIELDS = [
|
||||
DELETED_FIELD, // For soft delete filtering
|
||||
INDEXED_FIELD, // For index management
|
||||
VERSION_FIELD // For schema versioning
|
||||
]
|
||||
|
||||
/**
|
||||
* Fields that should use sorted index for O(log n) range queries
|
||||
*/
|
||||
export const SORTED_INDEX_FIELDS = [
|
||||
`${BRAINY_NS}.created`,
|
||||
`${BRAINY_NS}.updated`,
|
||||
`${BRAINY_NS}.priority`,
|
||||
`${BRAINY_NS}.ttl`
|
||||
]
|
||||
|
|
@ -1,545 +0,0 @@
|
|||
/**
|
||||
* Enhanced Structured Logging System for Brainy
|
||||
* Provides production-ready logging with structured output, context preservation,
|
||||
* performance tracking, and multiple transport support
|
||||
*/
|
||||
|
||||
import { performance } from 'perf_hooks'
|
||||
import { randomUUID } from '../universal/crypto.js'
|
||||
|
||||
export enum LogLevel {
|
||||
SILENT = -1,
|
||||
FATAL = 0,
|
||||
ERROR = 1,
|
||||
WARN = 2,
|
||||
INFO = 3,
|
||||
DEBUG = 4,
|
||||
TRACE = 5
|
||||
}
|
||||
|
||||
export interface LogContext {
|
||||
requestId?: string
|
||||
userId?: string
|
||||
operation?: string
|
||||
entityId?: string
|
||||
entityType?: string
|
||||
[key: string]: any
|
||||
}
|
||||
|
||||
export interface LogEntry {
|
||||
timestamp: string
|
||||
level: string
|
||||
levelNumeric: number
|
||||
module: string
|
||||
message: string
|
||||
context?: LogContext
|
||||
data?: any
|
||||
error?: {
|
||||
name: string
|
||||
message: string
|
||||
stack?: string
|
||||
code?: string
|
||||
}
|
||||
performance?: {
|
||||
duration?: number
|
||||
memory?: {
|
||||
used: number
|
||||
total: number
|
||||
}
|
||||
}
|
||||
host?: string
|
||||
pid: number
|
||||
version?: string
|
||||
}
|
||||
|
||||
export interface LogTransport {
|
||||
name: string
|
||||
log(entry: LogEntry): void | Promise<void>
|
||||
flush?(): Promise<void>
|
||||
}
|
||||
|
||||
export interface StructuredLoggerConfig {
|
||||
level: LogLevel
|
||||
modules?: Record<string, LogLevel>
|
||||
format: 'json' | 'pretty' | 'simple'
|
||||
transports: LogTransport[]
|
||||
context?: LogContext
|
||||
includeHost?: boolean
|
||||
includeMemory?: boolean
|
||||
bufferSize?: number
|
||||
flushInterval?: number
|
||||
version?: string
|
||||
}
|
||||
|
||||
class ConsoleTransport implements LogTransport {
|
||||
name = 'console'
|
||||
private format: 'json' | 'pretty' | 'simple'
|
||||
|
||||
constructor(format: 'json' | 'pretty' | 'simple' = 'json') {
|
||||
this.format = format
|
||||
}
|
||||
|
||||
log(entry: LogEntry): void {
|
||||
const method = this.getConsoleMethod(entry.levelNumeric)
|
||||
|
||||
if (this.format === 'json') {
|
||||
method(JSON.stringify(entry))
|
||||
} else if (this.format === 'pretty') {
|
||||
const color = this.getColor(entry.levelNumeric)
|
||||
const prefix = `${entry.timestamp} ${color}[${entry.level}]\\x1b[0m [${entry.module}]`
|
||||
const message = entry.message
|
||||
|
||||
if (entry.error) {
|
||||
method(`${prefix} ${message}`, entry.error)
|
||||
} else if (entry.data) {
|
||||
method(`${prefix} ${message}`, entry.data)
|
||||
} else {
|
||||
method(`${prefix} ${message}`)
|
||||
}
|
||||
} else {
|
||||
// Simple format
|
||||
method(`[${entry.level}] ${entry.message}`)
|
||||
}
|
||||
}
|
||||
|
||||
private getConsoleMethod(level: number): (...args: any[]) => void {
|
||||
switch (level) {
|
||||
case LogLevel.FATAL:
|
||||
case LogLevel.ERROR:
|
||||
return console.error
|
||||
case LogLevel.WARN:
|
||||
return console.warn
|
||||
case LogLevel.INFO:
|
||||
return console.info
|
||||
default:
|
||||
return console.log
|
||||
}
|
||||
}
|
||||
|
||||
private getColor(level: number): string {
|
||||
switch (level) {
|
||||
case LogLevel.FATAL:
|
||||
return '\\x1b[35m' // Magenta
|
||||
case LogLevel.ERROR:
|
||||
return '\\x1b[31m' // Red
|
||||
case LogLevel.WARN:
|
||||
return '\\x1b[33m' // Yellow
|
||||
case LogLevel.INFO:
|
||||
return '\\x1b[36m' // Cyan
|
||||
case LogLevel.DEBUG:
|
||||
return '\\x1b[32m' // Green
|
||||
case LogLevel.TRACE:
|
||||
return '\\x1b[90m' // Gray
|
||||
default:
|
||||
return '\\x1b[0m' // Reset
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class BufferedTransport implements LogTransport {
|
||||
name = 'buffered'
|
||||
private buffer: LogEntry[] = []
|
||||
private innerTransport: LogTransport
|
||||
private bufferSize: number
|
||||
private flushTimer?: NodeJS.Timeout
|
||||
|
||||
constructor(innerTransport: LogTransport, bufferSize: number = 100, flushInterval: number = 5000) {
|
||||
this.innerTransport = innerTransport
|
||||
this.bufferSize = bufferSize
|
||||
|
||||
if (flushInterval > 0) {
|
||||
this.flushTimer = setInterval(() => this.flush(), flushInterval)
|
||||
}
|
||||
}
|
||||
|
||||
log(entry: LogEntry): void {
|
||||
this.buffer.push(entry)
|
||||
|
||||
if (this.buffer.length >= this.bufferSize) {
|
||||
this.flush()
|
||||
}
|
||||
}
|
||||
|
||||
async flush(): Promise<void> {
|
||||
const entries = this.buffer.splice(0)
|
||||
for (const entry of entries) {
|
||||
await this.innerTransport.log(entry)
|
||||
}
|
||||
|
||||
if (this.innerTransport.flush) {
|
||||
await this.innerTransport.flush()
|
||||
}
|
||||
}
|
||||
|
||||
destroy(): void {
|
||||
if (this.flushTimer) {
|
||||
clearInterval(this.flushTimer)
|
||||
}
|
||||
this.flush()
|
||||
}
|
||||
}
|
||||
|
||||
export class StructuredLogger {
|
||||
private static instance: StructuredLogger
|
||||
private config: StructuredLoggerConfig
|
||||
private defaultContext: LogContext = {}
|
||||
private performanceMarks = new Map<string, number>()
|
||||
|
||||
private constructor() {
|
||||
const isDevelopment = process.env.NODE_ENV !== 'production'
|
||||
const format = isDevelopment ? 'pretty' : 'json'
|
||||
|
||||
this.config = {
|
||||
level: isDevelopment ? LogLevel.DEBUG : LogLevel.INFO,
|
||||
format,
|
||||
transports: [new ConsoleTransport(format)],
|
||||
includeHost: !isDevelopment,
|
||||
includeMemory: false,
|
||||
bufferSize: 100,
|
||||
flushInterval: 5000,
|
||||
version: process.env.npm_package_version
|
||||
}
|
||||
|
||||
// Load from environment
|
||||
this.loadEnvironmentConfig()
|
||||
}
|
||||
|
||||
private loadEnvironmentConfig(): void {
|
||||
const envLevel = process.env.BRAINY_LOG_LEVEL
|
||||
if (envLevel) {
|
||||
const level = LogLevel[envLevel.toUpperCase() as keyof typeof LogLevel]
|
||||
if (level !== undefined) {
|
||||
this.config.level = level
|
||||
}
|
||||
}
|
||||
|
||||
const envFormat = process.env.BRAINY_LOG_FORMAT
|
||||
if (envFormat && ['json', 'pretty', 'simple'].includes(envFormat)) {
|
||||
this.config.format = envFormat as 'json' | 'pretty' | 'simple'
|
||||
}
|
||||
|
||||
const moduleConfig = process.env.BRAINY_MODULE_LOG_LEVELS
|
||||
if (moduleConfig) {
|
||||
try {
|
||||
this.config.modules = JSON.parse(moduleConfig)
|
||||
} catch {
|
||||
// Ignore parse errors
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static getInstance(): StructuredLogger {
|
||||
if (!StructuredLogger.instance) {
|
||||
StructuredLogger.instance = new StructuredLogger()
|
||||
}
|
||||
return StructuredLogger.instance
|
||||
}
|
||||
|
||||
configure(config: Partial<StructuredLoggerConfig>): void {
|
||||
this.config = { ...this.config, ...config }
|
||||
}
|
||||
|
||||
setContext(context: LogContext): void {
|
||||
this.defaultContext = { ...this.defaultContext, ...context }
|
||||
}
|
||||
|
||||
clearContext(): void {
|
||||
this.defaultContext = {}
|
||||
}
|
||||
|
||||
withContext(context: LogContext): StructuredLogger {
|
||||
const contextualLogger = Object.create(this)
|
||||
contextualLogger.defaultContext = { ...this.defaultContext, ...context }
|
||||
return contextualLogger
|
||||
}
|
||||
|
||||
startTimer(label: string): void {
|
||||
this.performanceMarks.set(label, performance.now())
|
||||
}
|
||||
|
||||
endTimer(label: string): number | undefined {
|
||||
const start = this.performanceMarks.get(label)
|
||||
if (start === undefined) return undefined
|
||||
|
||||
const duration = performance.now() - start
|
||||
this.performanceMarks.delete(label)
|
||||
return duration
|
||||
}
|
||||
|
||||
private shouldLog(level: LogLevel, module: string): boolean {
|
||||
if (this.config.modules?.[module] !== undefined) {
|
||||
return level <= this.config.modules[module]
|
||||
}
|
||||
return level <= this.config.level
|
||||
}
|
||||
|
||||
private createLogEntry(
|
||||
level: LogLevel,
|
||||
module: string,
|
||||
message: string,
|
||||
context?: LogContext,
|
||||
data?: any,
|
||||
error?: Error
|
||||
): LogEntry {
|
||||
const entry: LogEntry = {
|
||||
timestamp: new Date().toISOString(),
|
||||
level: LogLevel[level],
|
||||
levelNumeric: level,
|
||||
module,
|
||||
message,
|
||||
pid: process.pid,
|
||||
version: this.config.version
|
||||
}
|
||||
|
||||
// Merge contexts
|
||||
const mergedContext = { ...this.defaultContext, ...context }
|
||||
if (Object.keys(mergedContext).length > 0) {
|
||||
entry.context = mergedContext
|
||||
}
|
||||
|
||||
if (data !== undefined) {
|
||||
entry.data = data
|
||||
}
|
||||
|
||||
if (error) {
|
||||
entry.error = {
|
||||
name: error.name,
|
||||
message: error.message,
|
||||
stack: error.stack,
|
||||
code: (error as Error & { code?: string }).code
|
||||
}
|
||||
}
|
||||
|
||||
if (this.config.includeHost) {
|
||||
try {
|
||||
const os = require('node:os')
|
||||
entry.host = os.hostname()
|
||||
} catch {
|
||||
entry.host = 'unknown'
|
||||
}
|
||||
}
|
||||
|
||||
if (this.config.includeMemory && typeof process !== 'undefined' && process.memoryUsage) {
|
||||
const mem = process.memoryUsage()
|
||||
entry.performance = {
|
||||
memory: {
|
||||
used: Math.round(mem.heapUsed / 1024 / 1024),
|
||||
total: Math.round(mem.heapTotal / 1024 / 1024)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return entry
|
||||
}
|
||||
|
||||
private log(
|
||||
level: LogLevel,
|
||||
module: string,
|
||||
message: string,
|
||||
contextOrData?: LogContext | any,
|
||||
data?: any
|
||||
): void {
|
||||
if (!this.shouldLog(level, module)) {
|
||||
return
|
||||
}
|
||||
|
||||
// Handle overloaded parameters
|
||||
let context: LogContext | undefined
|
||||
let logData: any
|
||||
|
||||
if (contextOrData && typeof contextOrData === 'object' && !Array.isArray(contextOrData)) {
|
||||
// Check if it looks like a context object
|
||||
const hasContextKeys = ['requestId', 'userId', 'operation', 'entityId', 'entityType']
|
||||
.some(key => key in contextOrData)
|
||||
|
||||
if (hasContextKeys) {
|
||||
context = contextOrData
|
||||
logData = data
|
||||
} else {
|
||||
logData = contextOrData
|
||||
}
|
||||
} else {
|
||||
logData = contextOrData
|
||||
}
|
||||
|
||||
// Extract error if present
|
||||
let error: Error | undefined
|
||||
if (logData instanceof Error) {
|
||||
error = logData
|
||||
logData = undefined
|
||||
} else if (logData?.error instanceof Error) {
|
||||
error = logData.error
|
||||
delete logData.error
|
||||
}
|
||||
|
||||
const entry = this.createLogEntry(level, module, message, context, logData, error)
|
||||
|
||||
// Send to all transports
|
||||
for (const transport of this.config.transports) {
|
||||
try {
|
||||
transport.log(entry)
|
||||
} catch (err) {
|
||||
// Fallback to console.error if transport fails
|
||||
console.error('Logger transport error:', err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fatal(module: string, message: string, contextOrData?: LogContext | any, data?: any): void {
|
||||
this.log(LogLevel.FATAL, module, message, contextOrData, data)
|
||||
}
|
||||
|
||||
error(module: string, message: string, contextOrData?: LogContext | any, data?: any): void {
|
||||
this.log(LogLevel.ERROR, module, message, contextOrData, data)
|
||||
}
|
||||
|
||||
warn(module: string, message: string, contextOrData?: LogContext | any, data?: any): void {
|
||||
this.log(LogLevel.WARN, module, message, contextOrData, data)
|
||||
}
|
||||
|
||||
info(module: string, message: string, contextOrData?: LogContext | any, data?: any): void {
|
||||
this.log(LogLevel.INFO, module, message, contextOrData, data)
|
||||
}
|
||||
|
||||
debug(module: string, message: string, contextOrData?: LogContext | any, data?: any): void {
|
||||
this.log(LogLevel.DEBUG, module, message, contextOrData, data)
|
||||
}
|
||||
|
||||
trace(module: string, message: string, contextOrData?: LogContext | any, data?: any): void {
|
||||
this.log(LogLevel.TRACE, module, message, contextOrData, data)
|
||||
}
|
||||
|
||||
createModuleLogger(module: string) {
|
||||
const self = this
|
||||
return {
|
||||
fatal: (message: string, contextOrData?: LogContext | any, data?: any) =>
|
||||
self.fatal(module, message, contextOrData, data),
|
||||
error: (message: string, contextOrData?: LogContext | any, data?: any) =>
|
||||
self.error(module, message, contextOrData, data),
|
||||
warn: (message: string, contextOrData?: LogContext | any, data?: any) =>
|
||||
self.warn(module, message, contextOrData, data),
|
||||
info: (message: string, contextOrData?: LogContext | any, data?: any) =>
|
||||
self.info(module, message, contextOrData, data),
|
||||
debug: (message: string, contextOrData?: LogContext | any, data?: any) =>
|
||||
self.debug(module, message, contextOrData, data),
|
||||
trace: (message: string, contextOrData?: LogContext | any, data?: any) =>
|
||||
self.trace(module, message, contextOrData, data),
|
||||
withContext: (context: LogContext) => {
|
||||
const contextual = self.withContext(context)
|
||||
return contextual.createModuleLogger(module)
|
||||
},
|
||||
startTimer: (label: string) => self.startTimer(`${module}:${label}`),
|
||||
endTimer: (label: string) => self.endTimer(`${module}:${label}`)
|
||||
}
|
||||
}
|
||||
|
||||
async flush(): Promise<void> {
|
||||
const flushPromises = this.config.transports
|
||||
.filter(t => t.flush)
|
||||
.map(t => t.flush!())
|
||||
|
||||
await Promise.all(flushPromises)
|
||||
}
|
||||
|
||||
addTransport(transport: LogTransport): void {
|
||||
this.config.transports.push(transport)
|
||||
}
|
||||
|
||||
removeTransport(name: string): void {
|
||||
this.config.transports = this.config.transports.filter(t => t.name !== name)
|
||||
}
|
||||
|
||||
child(context: LogContext): StructuredLogger {
|
||||
return this.withContext(context)
|
||||
}
|
||||
}
|
||||
|
||||
// Singleton instance
|
||||
export const structuredLogger = StructuredLogger.getInstance()
|
||||
|
||||
// Convenience functions
|
||||
export function createModuleLogger(module: string) {
|
||||
return structuredLogger.createModuleLogger(module)
|
||||
}
|
||||
|
||||
export function setLogContext(context: LogContext) {
|
||||
structuredLogger.setContext(context)
|
||||
}
|
||||
|
||||
export function withLogContext(context: LogContext) {
|
||||
return structuredLogger.withContext(context)
|
||||
}
|
||||
|
||||
// Correlation ID middleware helper
|
||||
export function createCorrelationId(): string {
|
||||
return randomUUID()
|
||||
}
|
||||
|
||||
// Performance logging helper
|
||||
export function logPerformance(
|
||||
logger: ReturnType<typeof createModuleLogger>,
|
||||
operation: string,
|
||||
fn: () => any
|
||||
): any {
|
||||
logger.startTimer(operation)
|
||||
try {
|
||||
const result = fn()
|
||||
if (result && typeof result.then === 'function') {
|
||||
return result.finally(() => {
|
||||
const duration = logger.endTimer(operation)
|
||||
logger.debug(`${operation} completed`, { duration })
|
||||
})
|
||||
}
|
||||
const duration = logger.endTimer(operation)
|
||||
logger.debug(`${operation} completed`, { duration })
|
||||
return result
|
||||
} catch (error) {
|
||||
const duration = logger.endTimer(operation)
|
||||
logger.error(`${operation} failed`, { duration, error })
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
// Backward compatibility wrapper for existing logger
|
||||
export class LoggerCompatibilityWrapper {
|
||||
private moduleLogger: ReturnType<typeof createModuleLogger>
|
||||
|
||||
constructor(module: string = 'legacy') {
|
||||
this.moduleLogger = createModuleLogger(module)
|
||||
}
|
||||
|
||||
error(module: string, message: string, ...args: any[]): void {
|
||||
this.moduleLogger.error(message, { module, data: args })
|
||||
}
|
||||
|
||||
warn(module: string, message: string, ...args: any[]): void {
|
||||
this.moduleLogger.warn(message, { module, data: args })
|
||||
}
|
||||
|
||||
info(module: string, message: string, ...args: any[]): void {
|
||||
this.moduleLogger.info(message, { module, data: args })
|
||||
}
|
||||
|
||||
debug(module: string, message: string, ...args: any[]): void {
|
||||
this.moduleLogger.debug(message, { module, data: args })
|
||||
}
|
||||
|
||||
trace(module: string, message: string, ...args: any[]): void {
|
||||
this.moduleLogger.trace(message, { module, data: args })
|
||||
}
|
||||
|
||||
createModuleLogger(module: string) {
|
||||
const logger = createModuleLogger(module)
|
||||
return {
|
||||
error: (message: string, ...args: any[]) => logger.error(message, { data: args }),
|
||||
warn: (message: string, ...args: any[]) => logger.warn(message, { data: args }),
|
||||
info: (message: string, ...args: any[]) => logger.info(message, { data: args }),
|
||||
debug: (message: string, ...args: any[]) => logger.debug(message, { data: args }),
|
||||
trace: (message: string, ...args: any[]) => logger.trace(message, { data: args })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Export types for external use
|
||||
export type ModuleLogger = ReturnType<typeof createModuleLogger>
|
||||
// Types are already exported above, no need to re-export
|
||||
|
|
@ -1,411 +0,0 @@
|
|||
/**
|
||||
* Write Buffer
|
||||
* Accumulates writes and flushes them in bulk to reduce S3 operations
|
||||
* Implements intelligent deduplication and compression
|
||||
*/
|
||||
|
||||
import { HNSWNoun, HNSWVerb } from '../coreTypes.js'
|
||||
import { createModuleLogger } from './logger.js'
|
||||
import { getGlobalBackpressure } from './adaptiveBackpressure.js'
|
||||
|
||||
interface BufferedWrite<T> {
|
||||
id: string
|
||||
data: T
|
||||
timestamp: number
|
||||
type: 'noun' | 'verb' | 'metadata'
|
||||
retryCount: number
|
||||
}
|
||||
|
||||
interface FlushResult {
|
||||
successful: number
|
||||
failed: number
|
||||
duration: number
|
||||
}
|
||||
|
||||
/**
|
||||
* High-performance write buffer for bulk operations
|
||||
*/
|
||||
export class WriteBuffer<T> {
|
||||
private logger = createModuleLogger('WriteBuffer')
|
||||
|
||||
// Buffer storage
|
||||
private buffer = new Map<string, BufferedWrite<T>>()
|
||||
|
||||
// Configuration - More aggressive for high volume
|
||||
private maxBufferSize = 2000 // Allow larger buffers
|
||||
private flushInterval = 500 // Flush more frequently (0.5 seconds)
|
||||
private minFlushSize = 50 // Lower minimum to flush sooner
|
||||
private maxRetries = 3 // Maximum retry attempts
|
||||
|
||||
// State
|
||||
private flushTimer: NodeJS.Timeout | null = null
|
||||
private isFlushing = false
|
||||
private lastFlush = Date.now()
|
||||
private pendingFlush: Promise<FlushResult> | null = null
|
||||
|
||||
// Statistics
|
||||
private totalWrites = 0
|
||||
private totalFlushes = 0
|
||||
private failedWrites = 0
|
||||
private duplicatesRemoved = 0
|
||||
|
||||
// Write function
|
||||
private writeFunction: (items: Map<string, T>) => Promise<void>
|
||||
private type: 'noun' | 'verb' | 'metadata'
|
||||
|
||||
// Backpressure integration
|
||||
private backpressure = getGlobalBackpressure()
|
||||
|
||||
constructor(
|
||||
type: 'noun' | 'verb' | 'metadata',
|
||||
writeFunction: (items: Map<string, T>) => Promise<void>,
|
||||
options?: {
|
||||
maxBufferSize?: number
|
||||
flushInterval?: number
|
||||
minFlushSize?: number
|
||||
}
|
||||
) {
|
||||
this.type = type
|
||||
this.writeFunction = writeFunction
|
||||
|
||||
if (options) {
|
||||
this.maxBufferSize = options.maxBufferSize || this.maxBufferSize
|
||||
this.flushInterval = options.flushInterval || this.flushInterval
|
||||
this.minFlushSize = options.minFlushSize || this.minFlushSize
|
||||
}
|
||||
|
||||
// Start periodic flush
|
||||
this.startPeriodicFlush()
|
||||
}
|
||||
|
||||
/**
|
||||
* Add item to buffer
|
||||
*/
|
||||
public async add(id: string, data: T): Promise<void> {
|
||||
// Check if we're already at capacity
|
||||
if (this.buffer.size >= this.maxBufferSize) {
|
||||
// Wait for current flush to complete
|
||||
if (this.pendingFlush) {
|
||||
await this.pendingFlush
|
||||
}
|
||||
|
||||
// Force flush if still at capacity
|
||||
if (this.buffer.size >= this.maxBufferSize) {
|
||||
await this.flush('capacity')
|
||||
}
|
||||
}
|
||||
|
||||
// Check for duplicate and update if newer
|
||||
const existing = this.buffer.get(id)
|
||||
if (existing) {
|
||||
// Update with newer data
|
||||
existing.data = data
|
||||
existing.timestamp = Date.now()
|
||||
this.duplicatesRemoved++
|
||||
} else {
|
||||
// Add new item
|
||||
this.buffer.set(id, {
|
||||
id,
|
||||
data,
|
||||
timestamp: Date.now(),
|
||||
type: this.type,
|
||||
retryCount: 0
|
||||
})
|
||||
}
|
||||
|
||||
this.totalWrites++
|
||||
|
||||
// Log buffer growth periodically
|
||||
if (this.totalWrites % 100 === 0) {
|
||||
this.logger.info(`📈 BUFFER GROWTH: ${this.buffer.size} ${this.type} items buffered (${this.totalWrites} total writes, ${this.duplicatesRemoved} deduplicated)`)
|
||||
}
|
||||
|
||||
// Check if we should flush
|
||||
this.checkFlush()
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if we should flush
|
||||
*/
|
||||
private checkFlush(): void {
|
||||
const bufferSize = this.buffer.size
|
||||
const timeSinceFlush = Date.now() - this.lastFlush
|
||||
|
||||
// Immediate flush conditions
|
||||
if (bufferSize >= this.maxBufferSize) {
|
||||
this.flush('size')
|
||||
return
|
||||
}
|
||||
|
||||
// Time-based flush with minimum size
|
||||
if (timeSinceFlush >= this.flushInterval && bufferSize >= this.minFlushSize) {
|
||||
this.flush('time')
|
||||
return
|
||||
}
|
||||
|
||||
// Adaptive flush based on system load
|
||||
const backpressureStatus = this.backpressure.getStatus()
|
||||
if (backpressureStatus.queueLength > 1000 && bufferSize > 10) {
|
||||
// System under pressure - flush smaller batches more frequently
|
||||
this.flush('pressure')
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush buffer to storage
|
||||
*/
|
||||
public async flush(reason: string = 'manual'): Promise<FlushResult> {
|
||||
// Prevent concurrent flushes
|
||||
if (this.isFlushing) {
|
||||
if (this.pendingFlush) {
|
||||
return this.pendingFlush
|
||||
}
|
||||
return { successful: 0, failed: 0, duration: 0 }
|
||||
}
|
||||
|
||||
// Nothing to flush
|
||||
if (this.buffer.size === 0) {
|
||||
return { successful: 0, failed: 0, duration: 0 }
|
||||
}
|
||||
|
||||
this.isFlushing = true
|
||||
const startTime = Date.now()
|
||||
|
||||
// Create flush promise
|
||||
this.pendingFlush = this.doFlush(reason, startTime)
|
||||
|
||||
try {
|
||||
const result = await this.pendingFlush
|
||||
return result
|
||||
} finally {
|
||||
this.isFlushing = false
|
||||
this.pendingFlush = null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform the actual flush
|
||||
*/
|
||||
private async doFlush(reason: string, startTime: number): Promise<FlushResult> {
|
||||
const itemsToFlush = new Map<string, T>()
|
||||
const flushingItems = new Map<string, BufferedWrite<T>>()
|
||||
|
||||
// Take items from buffer
|
||||
let count = 0
|
||||
for (const [id, item] of this.buffer.entries()) {
|
||||
itemsToFlush.set(id, item.data)
|
||||
flushingItems.set(id, item)
|
||||
count++
|
||||
|
||||
// Limit batch size for better performance
|
||||
if (count >= 500) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Remove from buffer
|
||||
for (const id of itemsToFlush.keys()) {
|
||||
this.buffer.delete(id)
|
||||
}
|
||||
|
||||
this.logger.warn(`🔄 BUFFERING: Flushing ${itemsToFlush.size} ${this.type} items (buffer had ${this.buffer.size + itemsToFlush.size}) - reason: ${reason}`)
|
||||
|
||||
try {
|
||||
// Request permission from backpressure system
|
||||
const opId = `flush-${Date.now()}`
|
||||
await this.backpressure.requestPermission(opId, 2) // Higher priority
|
||||
|
||||
try {
|
||||
// Perform bulk write
|
||||
await this.writeFunction(itemsToFlush)
|
||||
|
||||
// Success
|
||||
this.backpressure.releasePermission(opId, true)
|
||||
this.totalFlushes++
|
||||
this.lastFlush = Date.now()
|
||||
|
||||
const duration = Date.now() - startTime
|
||||
this.logger.warn(`🚀 BATCH FLUSH: ${itemsToFlush.size} ${this.type} items → 1 bulk S3 operation (${duration}ms, reason: ${reason})`)
|
||||
|
||||
return {
|
||||
successful: itemsToFlush.size,
|
||||
failed: 0,
|
||||
duration
|
||||
}
|
||||
} catch (error) {
|
||||
// Release with error
|
||||
this.backpressure.releasePermission(opId, false)
|
||||
throw error
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Flush failed: ${error}`)
|
||||
|
||||
// Put items back with retry count
|
||||
for (const [id, item] of flushingItems.entries()) {
|
||||
item.retryCount++
|
||||
|
||||
if (item.retryCount < this.maxRetries) {
|
||||
// Put back for retry
|
||||
this.buffer.set(id, item)
|
||||
} else {
|
||||
// Max retries exceeded
|
||||
this.failedWrites++
|
||||
this.logger.error(`Max retries exceeded for ${this.type} ${id}`)
|
||||
}
|
||||
}
|
||||
|
||||
const duration = Date.now() - startTime
|
||||
|
||||
return {
|
||||
successful: 0,
|
||||
failed: itemsToFlush.size,
|
||||
duration
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start periodic flush timer
|
||||
*/
|
||||
private startPeriodicFlush(): void {
|
||||
if (this.flushTimer) {
|
||||
return
|
||||
}
|
||||
|
||||
this.flushTimer = setInterval(() => {
|
||||
if (this.buffer.size > 0) {
|
||||
const timeSinceFlush = Date.now() - this.lastFlush
|
||||
|
||||
// Flush if we have items and enough time has passed
|
||||
if (timeSinceFlush >= this.flushInterval) {
|
||||
this.flush('periodic').catch(error => {
|
||||
this.logger.error('Periodic flush failed:', error)
|
||||
})
|
||||
}
|
||||
}
|
||||
}, Math.min(100, this.flushInterval / 2))
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop periodic flush timer
|
||||
*/
|
||||
public stop(): void {
|
||||
if (this.flushTimer) {
|
||||
clearInterval(this.flushTimer)
|
||||
this.flushTimer = null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Force flush all pending writes
|
||||
*/
|
||||
public async forceFlush(): Promise<FlushResult> {
|
||||
// Flush everything regardless of size
|
||||
const oldMinSize = this.minFlushSize
|
||||
this.minFlushSize = 0
|
||||
|
||||
try {
|
||||
const result = await this.flush('force')
|
||||
|
||||
// Flush any remaining items
|
||||
while (this.buffer.size > 0) {
|
||||
const additionalResult = await this.flush('force-remaining')
|
||||
result.successful += additionalResult.successful
|
||||
result.failed += additionalResult.failed
|
||||
result.duration += additionalResult.duration
|
||||
}
|
||||
|
||||
return result
|
||||
} finally {
|
||||
this.minFlushSize = oldMinSize
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get buffer statistics
|
||||
*/
|
||||
public getStats(): {
|
||||
bufferSize: number
|
||||
totalWrites: number
|
||||
totalFlushes: number
|
||||
failedWrites: number
|
||||
duplicatesRemoved: number
|
||||
avgFlushSize: number
|
||||
} {
|
||||
return {
|
||||
bufferSize: this.buffer.size,
|
||||
totalWrites: this.totalWrites,
|
||||
totalFlushes: this.totalFlushes,
|
||||
failedWrites: this.failedWrites,
|
||||
duplicatesRemoved: this.duplicatesRemoved,
|
||||
avgFlushSize: this.totalFlushes > 0 ? this.totalWrites / this.totalFlushes : 0
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Adjust parameters based on load
|
||||
*/
|
||||
public adjustForLoad(pendingRequests: number): void {
|
||||
if (pendingRequests > 10000) {
|
||||
// Extreme load - buffer more aggressively
|
||||
this.maxBufferSize = 5000
|
||||
this.flushInterval = 500
|
||||
this.minFlushSize = 500
|
||||
} else if (pendingRequests > 1000) {
|
||||
// High load
|
||||
this.maxBufferSize = 2000
|
||||
this.flushInterval = 1000
|
||||
this.minFlushSize = 200
|
||||
} else if (pendingRequests > 100) {
|
||||
// Moderate load
|
||||
this.maxBufferSize = 1000
|
||||
this.flushInterval = 2000
|
||||
this.minFlushSize = 100
|
||||
} else {
|
||||
// Low load - optimize for latency
|
||||
this.maxBufferSize = 500
|
||||
this.flushInterval = 5000
|
||||
this.minFlushSize = 50
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Global write buffers
|
||||
const writeBuffers = new Map<string, WriteBuffer<any>>()
|
||||
|
||||
/**
|
||||
* Get or create a write buffer
|
||||
*/
|
||||
export function getWriteBuffer<T>(
|
||||
id: string,
|
||||
type: 'noun' | 'verb' | 'metadata',
|
||||
writeFunction: (items: Map<string, T>) => Promise<void>
|
||||
): WriteBuffer<T> {
|
||||
if (!writeBuffers.has(id)) {
|
||||
writeBuffers.set(id, new WriteBuffer<T>(type, writeFunction))
|
||||
}
|
||||
return writeBuffers.get(id)!
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush all write buffers
|
||||
*/
|
||||
export async function flushAllBuffers(): Promise<void> {
|
||||
const promises: Promise<FlushResult>[] = []
|
||||
|
||||
for (const buffer of writeBuffers.values()) {
|
||||
promises.push(buffer.forceFlush())
|
||||
}
|
||||
|
||||
await Promise.all(promises)
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear all write buffers
|
||||
*/
|
||||
export function clearWriteBuffers(): void {
|
||||
for (const buffer of writeBuffers.values()) {
|
||||
buffer.stop()
|
||||
}
|
||||
writeBuffers.clear()
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue