469 lines
12 KiB
TypeScript
469 lines
12 KiB
TypeScript
import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
|
|
import Redis from 'ioredis';
|
|
|
|
/**
|
|
* Cache statistics interface
|
|
*/
|
|
export interface CacheStats {
|
|
hits: number;
|
|
misses: number;
|
|
sets: number;
|
|
deletes: number;
|
|
hitRate: number;
|
|
}
|
|
|
|
/**
|
|
* Cache options interface
|
|
*/
|
|
export interface CacheOptions {
|
|
ttl?: number; // Time to live in seconds
|
|
}
|
|
|
|
/**
|
|
* KnowledgeCacheService - Caching service for knowledge module using Valkey
|
|
*
|
|
* Provides caching operations for:
|
|
* - Entry details by slug
|
|
* - Search results
|
|
* - Graph query results
|
|
* - Cache statistics and metrics
|
|
*/
|
|
@Injectable()
|
|
export class KnowledgeCacheService implements OnModuleInit, OnModuleDestroy {
|
|
private readonly logger = new Logger(KnowledgeCacheService.name);
|
|
private client!: Redis;
|
|
|
|
// Cache key prefixes
|
|
private readonly ENTRY_PREFIX = 'knowledge:entry:';
|
|
private readonly SEARCH_PREFIX = 'knowledge:search:';
|
|
private readonly GRAPH_PREFIX = 'knowledge:graph:';
|
|
|
|
// Default TTL from environment (default: 5 minutes)
|
|
private readonly DEFAULT_TTL: number;
|
|
|
|
// Cache enabled flag
|
|
private readonly cacheEnabled: boolean;
|
|
|
|
// Stats tracking
|
|
private stats: CacheStats = {
|
|
hits: 0,
|
|
misses: 0,
|
|
sets: 0,
|
|
deletes: 0,
|
|
hitRate: 0,
|
|
};
|
|
|
|
constructor() {
|
|
this.DEFAULT_TTL = parseInt(process.env.KNOWLEDGE_CACHE_TTL || '300', 10);
|
|
this.cacheEnabled = process.env.KNOWLEDGE_CACHE_ENABLED !== 'false';
|
|
|
|
if (!this.cacheEnabled) {
|
|
this.logger.warn('Knowledge cache is DISABLED via environment configuration');
|
|
}
|
|
}
|
|
|
|
async onModuleInit() {
|
|
if (!this.cacheEnabled) {
|
|
return;
|
|
}
|
|
|
|
const valkeyUrl = process.env.VALKEY_URL || 'redis://localhost:6379';
|
|
|
|
this.logger.log(`Connecting to Valkey at ${valkeyUrl} for knowledge cache`);
|
|
|
|
this.client = new Redis(valkeyUrl, {
|
|
maxRetriesPerRequest: 3,
|
|
retryStrategy: (times) => {
|
|
const delay = Math.min(times * 50, 2000);
|
|
this.logger.warn(`Valkey connection retry attempt ${times}, waiting ${delay}ms`);
|
|
return delay;
|
|
},
|
|
reconnectOnError: (err) => {
|
|
this.logger.error('Valkey connection error:', err.message);
|
|
return true;
|
|
},
|
|
});
|
|
|
|
this.client.on('connect', () => {
|
|
this.logger.log('Knowledge cache connected to Valkey');
|
|
});
|
|
|
|
this.client.on('error', (err) => {
|
|
this.logger.error('Knowledge cache Valkey error:', err.message);
|
|
});
|
|
|
|
try {
|
|
await this.client.ping();
|
|
this.logger.log('Knowledge cache health check passed');
|
|
} catch (error) {
|
|
const errorMessage = error instanceof Error ? error.message : String(error);
|
|
this.logger.error('Knowledge cache health check failed:', errorMessage);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async onModuleDestroy() {
|
|
if (this.client) {
|
|
this.logger.log('Disconnecting knowledge cache from Valkey');
|
|
await this.client.quit();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get entry from cache by workspace and slug
|
|
*/
|
|
async getEntry<T = unknown>(workspaceId: string, slug: string): Promise<T | null> {
|
|
if (!this.cacheEnabled) return null;
|
|
|
|
try {
|
|
const key = this.getEntryKey(workspaceId, slug);
|
|
const cached = await this.client.get(key);
|
|
|
|
if (cached) {
|
|
this.stats.hits++;
|
|
this.updateHitRate();
|
|
this.logger.debug(`Cache HIT: ${key}`);
|
|
return JSON.parse(cached) as T;
|
|
}
|
|
|
|
this.stats.misses++;
|
|
this.updateHitRate();
|
|
this.logger.debug(`Cache MISS: ${key}`);
|
|
return null;
|
|
} catch (error) {
|
|
this.logger.error('Error getting entry from cache:', error);
|
|
return null; // Fail gracefully
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Set entry in cache
|
|
*/
|
|
async setEntry<T = unknown>(
|
|
workspaceId: string,
|
|
slug: string,
|
|
data: T,
|
|
options?: CacheOptions
|
|
): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
const key = this.getEntryKey(workspaceId, slug);
|
|
const ttl = options?.ttl ?? this.DEFAULT_TTL;
|
|
|
|
await this.client.setex(key, ttl, JSON.stringify(data));
|
|
|
|
this.stats.sets++;
|
|
this.logger.debug(`Cache SET: ${key} (TTL: ${ttl}s)`);
|
|
} catch (error) {
|
|
this.logger.error('Error setting entry in cache:', error);
|
|
// Don't throw - cache failures shouldn't break the app
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Invalidate entry cache
|
|
*/
|
|
async invalidateEntry(workspaceId: string, slug: string): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
const key = this.getEntryKey(workspaceId, slug);
|
|
await this.client.del(key);
|
|
|
|
this.stats.deletes++;
|
|
this.logger.debug(`Cache INVALIDATE: ${key}`);
|
|
} catch (error) {
|
|
this.logger.error('Error invalidating entry cache:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get search results from cache
|
|
*/
|
|
async getSearch<T = unknown>(
|
|
workspaceId: string,
|
|
query: string,
|
|
filters: Record<string, unknown>
|
|
): Promise<T | null> {
|
|
if (!this.cacheEnabled) return null;
|
|
|
|
try {
|
|
const key = this.getSearchKey(workspaceId, query, filters);
|
|
const cached = await this.client.get(key);
|
|
|
|
if (cached) {
|
|
this.stats.hits++;
|
|
this.updateHitRate();
|
|
this.logger.debug(`Cache HIT: ${key}`);
|
|
return JSON.parse(cached) as T;
|
|
}
|
|
|
|
this.stats.misses++;
|
|
this.updateHitRate();
|
|
this.logger.debug(`Cache MISS: ${key}`);
|
|
return null;
|
|
} catch (error) {
|
|
this.logger.error('Error getting search from cache:', error);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Set search results in cache
|
|
*/
|
|
async setSearch<T = unknown>(
|
|
workspaceId: string,
|
|
query: string,
|
|
filters: Record<string, unknown>,
|
|
data: T,
|
|
options?: CacheOptions
|
|
): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
const key = this.getSearchKey(workspaceId, query, filters);
|
|
const ttl = options?.ttl ?? this.DEFAULT_TTL;
|
|
|
|
await this.client.setex(key, ttl, JSON.stringify(data));
|
|
|
|
this.stats.sets++;
|
|
this.logger.debug(`Cache SET: ${key} (TTL: ${ttl}s)`);
|
|
} catch (error) {
|
|
this.logger.error('Error setting search in cache:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Invalidate all search caches for a workspace
|
|
*/
|
|
async invalidateSearches(workspaceId: string): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
const pattern = `${this.SEARCH_PREFIX}${workspaceId}:*`;
|
|
await this.deleteByPattern(pattern);
|
|
|
|
this.logger.debug(`Cache INVALIDATE: search caches for workspace ${workspaceId}`);
|
|
} catch (error) {
|
|
this.logger.error('Error invalidating search caches:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get graph query results from cache
|
|
*/
|
|
async getGraph<T = unknown>(
|
|
workspaceId: string,
|
|
entryId: string,
|
|
maxDepth: number
|
|
): Promise<T | null> {
|
|
if (!this.cacheEnabled) return null;
|
|
|
|
try {
|
|
const key = this.getGraphKey(workspaceId, entryId, maxDepth);
|
|
const cached = await this.client.get(key);
|
|
|
|
if (cached) {
|
|
this.stats.hits++;
|
|
this.updateHitRate();
|
|
this.logger.debug(`Cache HIT: ${key}`);
|
|
return JSON.parse(cached) as T;
|
|
}
|
|
|
|
this.stats.misses++;
|
|
this.updateHitRate();
|
|
this.logger.debug(`Cache MISS: ${key}`);
|
|
return null;
|
|
} catch (error) {
|
|
this.logger.error('Error getting graph from cache:', error);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Set graph query results in cache
|
|
*/
|
|
async setGraph<T = unknown>(
|
|
workspaceId: string,
|
|
entryId: string,
|
|
maxDepth: number,
|
|
data: T,
|
|
options?: CacheOptions
|
|
): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
const key = this.getGraphKey(workspaceId, entryId, maxDepth);
|
|
const ttl = options?.ttl ?? this.DEFAULT_TTL;
|
|
|
|
await this.client.setex(key, ttl, JSON.stringify(data));
|
|
|
|
this.stats.sets++;
|
|
this.logger.debug(`Cache SET: ${key} (TTL: ${ttl}s)`);
|
|
} catch (error) {
|
|
this.logger.error('Error setting graph in cache:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Invalidate all graph caches for a workspace
|
|
*/
|
|
async invalidateGraphs(workspaceId: string): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
const pattern = `${this.GRAPH_PREFIX}${workspaceId}:*`;
|
|
await this.deleteByPattern(pattern);
|
|
|
|
this.logger.debug(`Cache INVALIDATE: graph caches for workspace ${workspaceId}`);
|
|
} catch (error) {
|
|
this.logger.error('Error invalidating graph caches:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Invalidate graph caches that include a specific entry
|
|
*/
|
|
async invalidateGraphsForEntry(workspaceId: string, entryId: string): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
// We need to invalidate graphs centered on this entry
|
|
// and potentially graphs that include this entry as a node
|
|
// For simplicity, we'll invalidate all graphs in the workspace
|
|
// In a more optimized version, we could track which graphs include which entries
|
|
await this.invalidateGraphs(workspaceId);
|
|
|
|
this.logger.debug(`Cache INVALIDATE: graphs for entry ${entryId}`);
|
|
} catch (error) {
|
|
this.logger.error('Error invalidating graphs for entry:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get cache statistics
|
|
*/
|
|
getStats(): CacheStats {
|
|
return { ...this.stats };
|
|
}
|
|
|
|
/**
|
|
* Reset cache statistics
|
|
*/
|
|
resetStats(): void {
|
|
this.stats = {
|
|
hits: 0,
|
|
misses: 0,
|
|
sets: 0,
|
|
deletes: 0,
|
|
hitRate: 0,
|
|
};
|
|
this.logger.log('Cache statistics reset');
|
|
}
|
|
|
|
/**
|
|
* Clear all knowledge caches for a workspace
|
|
*/
|
|
async clearWorkspaceCache(workspaceId: string): Promise<void> {
|
|
if (!this.cacheEnabled) return;
|
|
|
|
try {
|
|
const patterns = [
|
|
`${this.ENTRY_PREFIX}${workspaceId}:*`,
|
|
`${this.SEARCH_PREFIX}${workspaceId}:*`,
|
|
`${this.GRAPH_PREFIX}${workspaceId}:*`,
|
|
];
|
|
|
|
for (const pattern of patterns) {
|
|
await this.deleteByPattern(pattern);
|
|
}
|
|
|
|
this.logger.log(`Cleared all caches for workspace ${workspaceId}`);
|
|
} catch (error) {
|
|
this.logger.error('Error clearing workspace cache:', error);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Generate cache key for entry
|
|
*/
|
|
private getEntryKey(workspaceId: string, slug: string): string {
|
|
return `${this.ENTRY_PREFIX}${workspaceId}:${slug}`;
|
|
}
|
|
|
|
/**
|
|
* Generate cache key for search
|
|
*/
|
|
private getSearchKey(
|
|
workspaceId: string,
|
|
query: string,
|
|
filters: Record<string, unknown>
|
|
): string {
|
|
const filterHash = this.hashObject(filters);
|
|
return `${this.SEARCH_PREFIX}${workspaceId}:${query}:${filterHash}`;
|
|
}
|
|
|
|
/**
|
|
* Generate cache key for graph
|
|
*/
|
|
private getGraphKey(
|
|
workspaceId: string,
|
|
entryId: string,
|
|
maxDepth: number
|
|
): string {
|
|
return `${this.GRAPH_PREFIX}${workspaceId}:${entryId}:${maxDepth}`;
|
|
}
|
|
|
|
/**
|
|
* Hash an object to create a consistent string representation
|
|
*/
|
|
private hashObject(obj: Record<string, unknown>): string {
|
|
return JSON.stringify(obj, Object.keys(obj).sort());
|
|
}
|
|
|
|
/**
|
|
* Update hit rate calculation
|
|
*/
|
|
private updateHitRate(): void {
|
|
const total = this.stats.hits + this.stats.misses;
|
|
this.stats.hitRate = total > 0 ? this.stats.hits / total : 0;
|
|
}
|
|
|
|
/**
|
|
* Delete keys matching a pattern
|
|
*/
|
|
private async deleteByPattern(pattern: string): Promise<void> {
|
|
if (!this.client) return;
|
|
|
|
let cursor = '0';
|
|
let deletedCount = 0;
|
|
|
|
do {
|
|
const [newCursor, keys] = await this.client.scan(
|
|
cursor,
|
|
'MATCH',
|
|
pattern,
|
|
'COUNT',
|
|
100
|
|
);
|
|
cursor = newCursor;
|
|
|
|
if (keys.length > 0) {
|
|
await this.client.del(...keys);
|
|
deletedCount += keys.length;
|
|
this.stats.deletes += keys.length;
|
|
}
|
|
} while (cursor !== '0');
|
|
|
|
this.logger.debug(`Deleted ${deletedCount} keys matching pattern: ${pattern}`);
|
|
}
|
|
|
|
/**
|
|
* Check if cache is enabled
|
|
*/
|
|
isEnabled(): boolean {
|
|
return this.cacheEnabled;
|
|
}
|
|
}
|