scriptoriumadmin/veritas-kanban
1
0
Fork 0
mirror of https://github.com/BradGroux/veritas-kanban.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

refactor: isolate telemetry file storage

Browse source
This commit is contained in:
Brad Groux 2026-08-23 20:34:38 -05:00
parent bab205757a
commit 94c3b569d5
4 changed files with 102 additions and 104 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
docs/architecture/service-filesystem-boundary.json
View file

@ -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",

30
server/src/services/metrics/telemetry-reader.ts
View file

@ -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<string[]> {
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);
}
/**

87
server/src/services/telemetry-service.ts
View file

@ -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<void> {
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<void> {
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<typeof createReadStream>;
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<string[]> {
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<void> {
const gzPath = filepath + '.gz';
await pipeline(createReadStream(filepath), createGzip(), createWriteStream(gzPath));
await fs.unlink(filepath);
}
dispose(): void {
if (this.ownsSqliteDatabase) {
this.sqliteDatabase?.close();

75
server/src/storage/telemetry-file-repository.ts Normal file
View file

@ -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<void> {
return mkdir(this.directory, { recursive: true }).then(() => undefined);
}
async listFiles(): Promise<string[]> {
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<void> {
await appendFile(this.eventPath(filename), content, 'utf8');
}
async remove(filename: string): Promise<void> {
await unlink(this.eventPath(filename));
}
async exists(filename: string): Promise<boolean> {
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<void> {
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<void> {
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,
});
}
}