diff --git a/docs/architecture/service-filesystem-boundary.json b/docs/architecture/service-filesystem-boundary.json index bb63c112..6def3964 100644 --- a/docs/architecture/service-filesystem-boundary.json +++ b/docs/architecture/service-filesystem-boundary.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "maximumEntries": 31, + "maximumEntries": 29, "entries": [ { "path": "server/src/services/attachment-service.ts", @@ -80,12 +80,6 @@ "owner": "#1188", "rationale": "Managed-content storage migration is tracked in issue #1188." }, - { - "path": "server/src/services/metrics/telemetry-reader.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, { "path": "server/src/services/notification-service.ts", "category": "authoritative-persistence", @@ -152,12 +146,6 @@ "owner": "#1187", "rationale": "Operational evidence storage migration is tracked in issue #1187." }, - { - "path": "server/src/services/telemetry-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, { "path": "server/src/services/template-service.ts", "category": "compatibility-debt", diff --git a/server/src/services/metrics/telemetry-reader.ts b/server/src/services/metrics/telemetry-reader.ts index 5877f455..bc9e78e4 100644 --- a/server/src/services/metrics/telemetry-reader.ts +++ b/server/src/services/metrics/telemetry-reader.ts @@ -2,15 +2,11 @@ * Telemetry file I/O utilities. * Handles reading NDJSON event files (plain and gzipped) with streaming support. */ -import { createReadStream } from '../../storage/fs-helpers.js'; -import fs from 'fs/promises'; -import path from 'path'; -import readline from 'readline'; -import { createGunzip } from 'zlib'; import type { SQLInputValue } from 'node:sqlite'; import type { AnyTelemetryEvent, TelemetryEventType, StreamEventHandler } from './types.js'; import { createLogger } from '../../lib/logger.js'; import { SqliteDatabase } from '../../storage/sqlite/database.js'; +import { TelemetryFileRepository } from '../../storage/telemetry-file-repository.js'; const log = createLogger('telemetry-reader'); interface SqliteTelemetryPayloadRow { @@ -27,14 +23,15 @@ interface SqliteTelemetryStartRow { */ export async function getEventFiles(telemetryDir: string, since: string | null): Promise { try { - const files = await fs.readdir(telemetryDir); + const repository = new TelemetryFileRepository(telemetryDir); + const files = await repository.listFiles(); const eventFiles = files.filter( (f) => f.startsWith('events-') && (f.endsWith('.ndjson') || f.endsWith('.ndjson.gz')) ); if (!since) { // Return all event files (for 'all' period) - return eventFiles.map((f) => path.join(telemetryDir, f)); + return eventFiles.map((f) => repository.eventPath(f)); } const sinceDate = since.slice(0, 10); @@ -44,7 +41,7 @@ export async function getEventFiles(telemetryDir: string, since: string | null): if (!match) return false; return match[1] >= sinceDate; }) - .map((f) => path.join(telemetryDir, f)); + .map((f) => repository.eventPath(f)); } catch (error: any) { if (error.code === 'ENOENT') { return []; @@ -56,21 +53,8 @@ export async function getEventFiles(telemetryDir: string, since: string | null): /** * Create a readline interface for an event file (handles both .ndjson and .ndjson.gz) */ -export function createLineReader(filePath: string): readline.Interface { - if (filePath.endsWith('.gz')) { - const fileStream = createReadStream(filePath); - const gunzip = createGunzip(); - const decompressed = fileStream.pipe(gunzip); - return readline.createInterface({ - input: decompressed, - crlfDelay: Infinity, - }); - } - const fileStream = createReadStream(filePath, { encoding: 'utf-8' }); - return readline.createInterface({ - input: fileStream, - crlfDelay: Infinity, - }); +export function createLineReader(filePath: string) { + return TelemetryFileRepository.createLineReader(filePath); } /** diff --git a/server/src/services/telemetry-service.ts b/server/src/services/telemetry-service.ts index 84d9b908..3142a67d 100644 --- a/server/src/services/telemetry-service.ts +++ b/server/src/services/telemetry-service.ts @@ -1,10 +1,4 @@ -import fs from 'fs/promises'; -import { createReadStream, createWriteStream } from '../storage/fs-helpers.js'; -import path from 'path'; import { getTelemetryDir } from '../utils/paths.js'; -import { createGzip, createGunzip } from 'zlib'; -import { pipeline } from 'stream/promises'; -import readline from 'readline'; import { nanoid } from 'nanoid'; import type { TelemetryEvent, @@ -17,6 +11,7 @@ import { createLogger } from '../lib/logger.js'; import type { TelemetryRepository } from '../storage/interfaces.js'; import { SqliteDatabase, type SqliteConnectionOptions } from '../storage/sqlite/database.js'; import { SqliteTelemetryRepository } from '../storage/sqlite/telemetry-repository.js'; +import { TelemetryFileRepository } from '../storage/telemetry-file-repository.js'; const log = createLogger('telemetry-service'); // Default paths - resolve via shared paths helper (respects DATA_DIR/VERITAS_DATA_DIR) @@ -53,9 +48,11 @@ export class TelemetryService { private repository: TelemetryRepository | null = null; private sqliteDatabase: SqliteDatabase | null = null; private ownsSqliteDatabase = false; + private readonly fileRepository: TelemetryFileRepository; constructor(options: TelemetryServiceOptions = {}) { this.telemetryDir = options.telemetryDir || TELEMETRY_DIR; + this.fileRepository = new TelemetryFileRepository(this.telemetryDir); // Read retention from env var, falling back to options, then default const envRetention = process.env.TELEMETRY_RETENTION_DAYS; @@ -102,7 +99,7 @@ export class TelemetryService { return; } - await fs.mkdir(this.telemetryDir, { recursive: true }); + await this.fileRepository.ensureReady(); await this.cleanupOldEvents(); this.initialized = true; } @@ -372,11 +369,11 @@ export class TelemetryService { } await this.init(); - const files = await fs.readdir(this.telemetryDir); + const files = await this.fileRepository.listFiles(); for (const file of files) { if (file.endsWith('.ndjson') || file.endsWith('.ndjson.gz')) { - await fs.unlink(path.join(this.telemetryDir, file)); + await this.fileRepository.remove(file); } } } @@ -482,10 +479,9 @@ export class TelemetryService { */ private async writeEvent(event: TelemetryEvent): Promise { const filename = this.getFilenameForDate(new Date(event.timestamp)); - const filepath = path.join(this.telemetryDir, filename); const line = JSON.stringify(event) + '\n'; - await fs.appendFile(filepath, line, 'utf-8'); + await this.fileRepository.append(filename, line); } /** @@ -497,49 +493,15 @@ export class TelemetryService { filename: string, callback: (event: AnyTelemetryEvent) => boolean | void ): Promise { - const filepath = path.join(this.telemetryDir, filename); - const isGzipped = filename.endsWith('.gz'); - - try { - await fs.access(filepath); - } catch { - return; // File doesn't exist - } - - return new Promise((resolve, reject) => { - let stream = createReadStream(filepath); - - if (isGzipped) { - const gunzip = createGunzip(); - stream = stream.pipe(gunzip) as unknown as ReturnType; + if (!(await this.fileRepository.exists(filename))) return; + await this.fileRepository.streamLines(filename, (line) => { + if (!line.trim()) return; + try { + const event = JSON.parse(line) as AnyTelemetryEvent; + return callback(event); + } catch { + log.error({ err: line }, '[Telemetry] Failed to parse line'); } - - const rl = readline.createInterface({ - input: stream as NodeJS.ReadableStream, - crlfDelay: Infinity, - }); - - let stopped = false; - - rl.on('line', (line) => { - if (stopped || !line.trim()) return; - - try { - const event = JSON.parse(line) as AnyTelemetryEvent; - const shouldContinue = callback(event); - if (shouldContinue === false) { - stopped = true; - rl.close(); - stream.destroy(); - } - } catch { - log.error({ err: line }, '[Telemetry] Failed to parse line'); - } - }); - - rl.on('close', () => resolve()); - rl.on('error', reject); - stream.on('error', reject); }); } @@ -559,7 +521,7 @@ export class TelemetryService { * Get list of event files within a date range (includes both .ndjson and .ndjson.gz) */ private async getEventFiles(since?: string, until?: string): Promise { - const files = await fs.readdir(this.telemetryDir); + const files = await this.fileRepository.listFiles(); const eventFiles = files.filter( (f) => f.startsWith('events-') && (f.endsWith('.ndjson') || f.endsWith('.ndjson.gz')) ); @@ -607,7 +569,7 @@ export class TelemetryService { compressCutoff.setDate(compressCutoff.getDate() - this.compressAfterDays); const compressCutoffStr = compressCutoff.toISOString().slice(0, 10); - const files = await fs.readdir(this.telemetryDir); + const files = await this.fileRepository.listFiles(); let deleted = 0; let compressed = 0; @@ -617,11 +579,9 @@ export class TelemetryService { const fileDate = match[1]; const isCompressed = !!match[2]; - const filepath = path.join(this.telemetryDir, filename); - // Delete files older than retention period if (fileDate < retentionCutoffStr) { - await fs.unlink(filepath); + await this.fileRepository.remove(filename); deleted++; continue; } @@ -629,7 +589,7 @@ export class TelemetryService { // Compress uncompressed files older than compress threshold if (this.compressAfterDays > 0 && !isCompressed && fileDate < compressCutoffStr) { try { - await this.compressFile(filepath); + await this.fileRepository.compress(filename); compressed++; } catch (err) { log.error({ err: err }, `[Telemetry] Failed to compress ${filename}`); @@ -645,15 +605,6 @@ export class TelemetryService { } } - /** - * Compress an NDJSON file to gzip and remove the original. - */ - private async compressFile(filepath: string): Promise { - const gzPath = filepath + '.gz'; - await pipeline(createReadStream(filepath), createGzip(), createWriteStream(gzPath)); - await fs.unlink(filepath); - } - dispose(): void { if (this.ownsSqliteDatabase) { this.sqliteDatabase?.close(); diff --git a/server/src/storage/telemetry-file-repository.ts b/server/src/storage/telemetry-file-repository.ts new file mode 100644 index 00000000..a5d51bff --- /dev/null +++ b/server/src/storage/telemetry-file-repository.ts @@ -0,0 +1,75 @@ +import { access, appendFile, mkdir, readdir, unlink } from 'node:fs/promises'; +import path from 'node:path'; +import readline from 'node:readline'; +import { pipeline } from 'node:stream/promises'; +import { createGunzip, createGzip } from 'node:zlib'; +import { ensureWithinBase } from '../utils/sanitize.js'; +import { createReadStream, createWriteStream } from './fs-helpers.js'; + +export class TelemetryFileRepository { + private readonly directory: string; + + constructor(directory: string) { + this.directory = path.resolve(directory); + } + + ensureReady(): Promise { + return mkdir(this.directory, { recursive: true }).then(() => undefined); + } + + async listFiles(): Promise { + return readdir(this.directory); + } + + eventPath(filename: string): string { + const safeFilename = path.basename(filename); + if (safeFilename !== filename) throw new Error('Telemetry filename must be a path segment'); + return ensureWithinBase(this.directory, path.join(this.directory, safeFilename)); + } + + async append(filename: string, content: string): Promise { + await appendFile(this.eventPath(filename), content, 'utf8'); + } + + async remove(filename: string): Promise { + await unlink(this.eventPath(filename)); + } + + async exists(filename: string): Promise { + try { + await access(this.eventPath(filename)); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return false; + throw error; + } + } + + async streamLines(filename: string, visitor: (line: string) => boolean | void): Promise { + const reader = TelemetryFileRepository.createLineReader(this.eventPath(filename)); + for await (const line of reader) { + if (visitor(line) === false) { + reader.close(); + break; + } + } + } + + async compress(filename: string): Promise { + const source = this.eventPath(filename); + const destination = this.eventPath(`${filename}.gz`); + await pipeline(createReadStream(source), createGzip(), createWriteStream(destination)); + await unlink(source); + } + + static createLineReader(filePath: string): readline.Interface { + if (filePath.endsWith('.gz')) { + const decompressed = createReadStream(filePath).pipe(createGunzip()); + return readline.createInterface({ input: decompressed, crlfDelay: Infinity }); + } + return readline.createInterface({ + input: createReadStream(filePath, { encoding: 'utf8' }), + crlfDelay: Infinity, + }); + } +}