68 lines
2.1 KiB
TypeScript
68 lines
2.1 KiB
TypeScript
import { Inject, Injectable } from '@nestjs/common';
|
|
import type { QueueHandle } from '@mosaicstack/queue';
|
|
import type { LogService } from '@mosaicstack/log';
|
|
import { LOG_SERVICE } from '../log/log.tokens.js';
|
|
import { REDIS } from './gc.tokens.js';
|
|
|
|
export interface GCResult {
|
|
sessionId: string;
|
|
cleaned: {
|
|
valkeyKeys?: number;
|
|
logsDemoted?: number;
|
|
tempFilesRemoved?: number;
|
|
};
|
|
}
|
|
|
|
/** Escape Redis glob metacharacters so a session identifier is always literal. */
|
|
function escapeRedisGlobLiteral(value: string): string {
|
|
return value.replace(/[\\*?\[\]]/g, '\\$&');
|
|
}
|
|
|
|
@Injectable()
|
|
export class SessionGCService {
|
|
constructor(
|
|
@Inject(REDIS) private readonly redis: QueueHandle['redis'],
|
|
@Inject(LOG_SERVICE) private readonly logService: LogService,
|
|
) {}
|
|
|
|
/**
|
|
* Scan Valkey for all keys matching a pattern using SCAN (non-blocking).
|
|
* KEYS is avoided because it blocks the Valkey event loop for the full scan
|
|
* duration, which can cause latency spikes under production key volumes.
|
|
*/
|
|
private async scanKeys(pattern: string): Promise<string[]> {
|
|
const collected: string[] = [];
|
|
let cursor = '0';
|
|
do {
|
|
const [nextCursor, keys] = await this.redis.scan(cursor, 'MATCH', pattern, 'COUNT', 100);
|
|
cursor = nextCursor;
|
|
collected.push(...keys);
|
|
} while (cursor !== '0');
|
|
return collected;
|
|
}
|
|
|
|
/**
|
|
* Immediate cleanup for a single session (call from destroySession).
|
|
*/
|
|
async collect(sessionId: string): Promise<GCResult> {
|
|
const result: GCResult = { sessionId, cleaned: {} };
|
|
|
|
// 1. Valkey: delete all session-scoped keys
|
|
const pattern = `mosaic:session:${escapeRedisGlobLiteral(sessionId)}:*`;
|
|
const valkeyKeys = await this.scanKeys(pattern);
|
|
if (valkeyKeys.length > 0) {
|
|
await this.redis.del(...valkeyKeys);
|
|
result.cleaned.valkeyKeys = valkeyKeys.length;
|
|
}
|
|
|
|
// 2. PG: demote hot-tier agent logs for this session only.
|
|
const cutoff = new Date();
|
|
const logsDemoted = await this.logService.logs.promoteSessionToWarm(sessionId, cutoff);
|
|
if (logsDemoted > 0) {
|
|
result.cleaned.logsDemoted = logsDemoted;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
}
|