diff --git a/docs/architecture/service-filesystem-boundary.json b/docs/architecture/service-filesystem-boundary.json index 6def3964..c9f5eb66 100644 --- a/docs/architecture/service-filesystem-boundary.json +++ b/docs/architecture/service-filesystem-boundary.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "maximumEntries": 29, + "maximumEntries": 16, "entries": [ { "path": "server/src/services/attachment-service.ts", @@ -8,12 +8,6 @@ "owner": "#1188", "rationale": "Managed-content storage migration is tracked in issue #1188." }, - { - "path": "server/src/services/audit-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, { "path": "server/src/services/clawdbot-agent-service.ts", "category": "transient-process-io", @@ -38,12 +32,6 @@ "owner": "#1188", "rationale": "Managed-content storage migration is tracked in issue #1188." }, - { - "path": "server/src/services/external-tracker-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, { "path": "server/src/services/file-lock.ts", "category": "transient-process-io", @@ -56,42 +44,12 @@ "owner": "#1189", "rationale": "Remaining process I/O migration is tracked in issue #1189." }, - { - "path": "server/src/services/github-sync-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, - { - "path": "server/src/services/integrity-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, - { - "path": "server/src/services/maintenance-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, { "path": "server/src/services/managed-list-service.ts", "category": "compatibility-debt", "owner": "#1188", "rationale": "Managed-content storage migration is tracked in issue #1188." }, - { - "path": "server/src/services/notification-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, - { - "path": "server/src/services/outbound-integration-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, { "path": "server/src/services/pdf-report-service.ts", "category": "transient-process-io", @@ -104,24 +62,6 @@ "owner": "#1188", "rationale": "Managed-content storage migration is tracked in issue #1188." }, - { - "path": "server/src/services/queue-intake-monitor-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, - { - "path": "server/src/services/run-session-share-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, - { - "path": "server/src/services/search-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, { "path": "server/src/services/shared-resources-service.ts", "category": "compatibility-debt", @@ -140,12 +80,6 @@ "owner": "#1189", "rationale": "Remaining process I/O migration is tracked in issue #1189." }, - { - "path": "server/src/services/task-identity-diagnostics.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", @@ -163,18 +97,6 @@ "category": "compatibility-debt", "owner": "#1188", "rationale": "Managed-content storage migration is tracked in issue #1188." - }, - { - "path": "server/src/services/trace-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." - }, - { - "path": "server/src/services/work-product-service.ts", - "category": "authoritative-persistence", - "owner": "#1187", - "rationale": "Operational evidence storage migration is tracked in issue #1187." } ] } diff --git a/server/src/services/audit-service.ts b/server/src/services/audit-service.ts index 7d98162d..9b6f1cb1 100644 --- a/server/src/services/audit-service.ts +++ b/server/src/services/audit-service.ts @@ -7,14 +7,11 @@ * Log files are stored as JSONL (one JSON object per line) with monthly rotation: * {dataDir}/audit/audit-{YYYY-MM}.log */ -import fs from 'fs/promises'; -import { createReadStream } from 'fs'; -import path from 'path'; import crypto from 'crypto'; -import readline from 'readline'; import { createLogger } from '../lib/logger.js'; import { SqliteDatabase } from '../storage/sqlite/database.js'; import { SqliteAuditRepository } from '../storage/sqlite/audit-policy-repositories.js'; +import { AuditFileRepository } from '../storage/audit-file-repository.js'; const log = createLogger('audit'); @@ -57,6 +54,7 @@ import { getAuditDir } from '../utils/paths.js'; const AUDIT_DIR = getAuditDir(); const SQLITE_AUDIT_LOG_PATH = 'sqlite://audit/current'; +const auditFileRepository = new AuditFileRepository(AUDIT_DIR); // --------------------------------------------------------------------------- // Internal State @@ -89,21 +87,19 @@ function logFilePath(date: Date = new Date()): string { return SQLITE_AUDIT_LOG_PATH; } - const yyyy = date.getFullYear(); - const mm = String(date.getMonth() + 1).padStart(2, '0'); - const month = `${yyyy}-${mm}`; + const month = `${date.getFullYear()}-${String(date.getMonth() + 1).padStart(2, '0')}`; // Cache to avoid path.join on every write if (month !== currentMonth) { currentMonth = month; - currentLogPath = path.join(AUDIT_DIR, `audit-${month}.log`); + currentLogPath = auditFileRepository.getMonthlyLogPath(date); } return currentLogPath; } /** Ensure the audit directory exists. */ async function ensureAuditDir(): Promise { - await fs.mkdir(AUDIT_DIR, { recursive: true }); + await auditFileRepository.ensureReady(); } function isSqliteAuditEnabled(): boolean { @@ -131,22 +127,7 @@ async function seedLastHash(filePath: string): Promise { return; } - try { - const content = await fs.readFile(filePath, 'utf8'); - const lines = content.trimEnd().split('\n').filter(Boolean); - if (lines.length > 0) { - lastHash = sha256(lines[lines.length - 1]); - } else { - lastHash = ''; - } - } catch (err: unknown) { - // File doesn't exist yet — first entry - if ((err as NodeJS.ErrnoException).code === 'ENOENT') { - lastHash = ''; - } else { - throw err; - } - } + lastHash = await auditFileRepository.getLastHash(filePath); } /** Track whether we've seeded for the current file. */ @@ -198,7 +179,7 @@ async function writeEntry(event: AuditEvent): Promise { if (isSqliteAuditEnabled()) { getAuditRepository().save(entry, line); } else { - await fs.appendFile(filePath, line + '\n', 'utf8'); + await auditFileRepository.append(filePath, line); } // Update the running hash @@ -216,71 +197,7 @@ export async function verifyAuditLog(filePath: string): Promise { return getAuditRepository().verify(); } - // Check if file exists - try { - await fs.access(filePath); - } catch (err: unknown) { - if ((err as NodeJS.ErrnoException).code === 'ENOENT') { - return { valid: true, entries: 0 }; - } - throw err; - } - - return new Promise((resolve, reject) => { - const stream = createReadStream(filePath, { encoding: 'utf8' }); - const rl = readline.createInterface({ - input: stream, - crlfDelay: Infinity, - }); - - let prevHash = ''; - let lineIndex = 0; - let totalLines = 0; - let invalidResult: VerifyResult | null = null; - - rl.on('line', (line) => { - if (invalidResult) return; // Already found an error - - const trimmed = line.trim(); - if (!trimmed) { - lineIndex++; - return; - } - - totalLines++; - - let entry: AuditEntry; - try { - entry = JSON.parse(trimmed) as AuditEntry; - } catch { - invalidResult = { valid: false, entries: totalLines, firstBroken: lineIndex }; - rl.close(); - stream.destroy(); - return; - } - - if (entry.integrity !== prevHash) { - invalidResult = { valid: false, entries: totalLines, firstBroken: lineIndex }; - rl.close(); - stream.destroy(); - return; - } - - prevHash = sha256(trimmed); - lineIndex++; - }); - - rl.on('close', () => { - if (invalidResult) { - resolve(invalidResult); - } else { - resolve({ valid: true, entries: totalLines }); - } - }); - - rl.on('error', reject); - stream.on('error', reject); - }); + return auditFileRepository.verify(filePath); } /** @@ -293,20 +210,7 @@ export async function readRecentAuditEntries(limit = 100): Promise return getAuditRepository().readRecent(limit) as AuditEntry[]; } - let content: string; - try { - content = await fs.readFile(filePath, 'utf8'); - } catch (err: unknown) { - if ((err as NodeJS.ErrnoException).code === 'ENOENT') { - return []; - } - throw err; - } - - const lines = content.trimEnd().split('\n').filter(Boolean); - // Take the last `limit` entries and reverse for newest-first - const slice = lines.slice(-limit).reverse(); - return slice.map((line) => JSON.parse(line) as AuditEntry); + return auditFileRepository.readRecent(filePath, limit); } /** diff --git a/server/src/services/external-tracker-service.ts b/server/src/services/external-tracker-service.ts index 99ac4675..6c1c0d4f 100644 --- a/server/src/services/external-tracker-service.ts +++ b/server/src/services/external-tracker-service.ts @@ -1,4 +1,3 @@ -import fs from 'fs/promises'; import path from 'path'; import { nanoid } from 'nanoid'; import type { @@ -24,9 +23,9 @@ import type { } from '@veritas-kanban/shared'; import { auditLog, type AuditEvent } from './audit-service.js'; import { activityService, type ActivityService } from './activity-service.js'; -import { withFileLock } from './file-lock.js'; import { getTaskService } from './task-service.js'; import { ConflictError, NotFoundError, ValidationError } from '../middleware/error-handler.js'; +import { JsonFileRepository } from '../storage/json-file-repository.js'; import { getRuntimeDir } from '../utils/paths.js'; import { stripHtml, validatePathSegment } from '../utils/sanitize.js'; @@ -400,7 +399,7 @@ class MockExternalTrackerAdapter implements ExternalTrackerAdapter { } export class ExternalTrackerService { - private readonly storageDir: string; + private readonly repository: JsonFileRepository; private readonly persist: boolean; private readonly audit: (event: AuditEvent) => Promise; private readonly taskService: ExternalTrackerTaskService; @@ -410,7 +409,8 @@ export class ExternalTrackerService { private state: ExternalTrackerState = this.emptyState(); constructor(options: ExternalTrackerServiceOptions = {}) { - this.storageDir = options.storageDir ?? path.join(getRuntimeDir(), 'external-trackers'); + const storageDir = options.storageDir ?? path.join(getRuntimeDir(), 'external-trackers'); + this.repository = new JsonFileRepository(path.join(storageDir, STATE_FILE)); this.persist = options.persist ?? process.env.VITEST !== 'true'; this.audit = options.audit ?? auditLog; this.taskService = options.taskService ?? getTaskService(); @@ -946,10 +946,6 @@ export class ExternalTrackerService { }; } - private get stateFile(): string { - return path.join(this.storageDir, STATE_FILE); - } - private async ensureLoaded(): Promise { if (this.loaded) return; if (!this.persist) { @@ -960,10 +956,8 @@ export class ExternalTrackerService { this.loaded = true; return; } - await fs.mkdir(this.storageDir, { recursive: true }); try { - const content = await fs.readFile(this.stateFile, 'utf8'); - this.state = { ...this.emptyState(), ...JSON.parse(content) }; + this.state = { ...this.emptyState(), ...(await this.repository.read()) }; } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; this.state = this.emptyState(); @@ -977,10 +971,7 @@ export class ExternalTrackerService { private async saveState(): Promise { if (!this.persist) return; - await fs.mkdir(this.storageDir, { recursive: true }); - await withFileLock(this.stateFile, async () => { - await fs.writeFile(this.stateFile, JSON.stringify(this.state, null, 2), 'utf8'); - }); + await this.repository.write(this.state); } } diff --git a/server/src/services/github-sync-service.ts b/server/src/services/github-sync-service.ts index cce4c96e..601de532 100644 --- a/server/src/services/github-sync-service.ts +++ b/server/src/services/github-sync-service.ts @@ -8,11 +8,10 @@ * Persists config to `.veritas-kanban/integrations.json` and sync state to * `.veritas-kanban/github-sync.json`. */ -import { readFile, writeFile, mkdir } from 'fs/promises'; import { join } from 'path'; import { execFile } from 'child_process'; import { promisify } from 'util'; -import { fileExists } from '../storage/fs-helpers.js'; +import { JsonFileRepository } from '../storage/json-file-repository.js'; import { getBreaker } from './circuit-registry.js'; import { getTaskService, type TaskService } from './task-service.js'; import { createLogger } from '../lib/logger.js'; @@ -68,6 +67,10 @@ interface GhIssue { const DATA_DIR = getRuntimeDir(); const INTEGRATIONS_FILE = join(DATA_DIR, 'integrations.json'); const SYNC_STATE_FILE = join(DATA_DIR, 'github-sync.json'); +const integrationsRepository = new JsonFileRepository>( + INTEGRATIONS_FILE +); +const syncStateRepository = new JsonFileRepository>(SYNC_STATE_FILE); const DEFAULT_CONFIG: IntegrationsConfig = { github: { @@ -97,13 +100,8 @@ export class GitHubSyncService { // ── Config persistence ──────────────────────────────────── async getConfig(): Promise { - await mkdir(DATA_DIR, { recursive: true }); - if (!(await fileExists(INTEGRATIONS_FILE))) { - return { ...DEFAULT_CONFIG }; - } try { - const raw = await readFile(INTEGRATIONS_FILE, 'utf-8'); - const parsed = JSON.parse(raw) as Partial; + const parsed = await integrationsRepository.read(); return { github: { ...DEFAULT_CONFIG.github, ...parsed.github }, }; @@ -115,21 +113,15 @@ export class GitHubSyncService { async updateConfig(patch: Partial): Promise { const current = await this.getConfig(); current.github = { ...current.github, ...patch }; - await mkdir(DATA_DIR, { recursive: true }); - await writeFile(INTEGRATIONS_FILE, JSON.stringify(current, null, 2), 'utf-8'); + await integrationsRepository.write(current); return current; } // ── Sync-state persistence ──────────────────────────────── async getSyncState(): Promise { - await mkdir(DATA_DIR, { recursive: true }); - if (!(await fileExists(SYNC_STATE_FILE))) { - return { ...DEFAULT_STATE, issueMappings: {} }; - } try { - const raw = await readFile(SYNC_STATE_FILE, 'utf-8'); - const parsed = JSON.parse(raw) as Partial; + const parsed = await syncStateRepository.read(); return { lastSyncAt: parsed.lastSyncAt ?? null, issueMappings: parsed.issueMappings ?? {}, @@ -140,8 +132,7 @@ export class GitHubSyncService { } private async saveSyncState(state: SyncState): Promise { - await mkdir(DATA_DIR, { recursive: true }); - await writeFile(SYNC_STATE_FILE, JSON.stringify(state, null, 2), 'utf-8'); + await syncStateRepository.write(state); } // ── gh CLI helpers ──────────────────────────────────────── diff --git a/server/src/services/integrity-service.ts b/server/src/services/integrity-service.ts index 6aaafb8d..85c5ce44 100644 --- a/server/src/services/integrity-service.ts +++ b/server/src/services/integrity-service.ts @@ -1,7 +1,7 @@ -import fs from 'fs/promises'; import path from 'path'; import matter, { type MarkdownFrontmatterFile } from '../utils/frontmatter.js'; import { createLogger } from '../lib/logger.js'; +import * as fs from '../storage/fs-helpers.js'; const log = createLogger('integrity'); diff --git a/server/src/services/maintenance-service.ts b/server/src/services/maintenance-service.ts index dace7414..670ecda7 100644 --- a/server/src/services/maintenance-service.ts +++ b/server/src/services/maintenance-service.ts @@ -1,4 +1,3 @@ -import fs from 'fs/promises'; import path from 'path'; import { createHash } from 'node:crypto'; import { @@ -33,6 +32,7 @@ import { import { redactString } from '../lib/redact.js'; import { getSqliteStorageDiagnostics } from '../storage/sqlite/database.js'; import { getStorage } from '../storage/index.js'; +import * as fs from '../storage/fs-helpers.js'; import { getRunPhaseAuthorityService } from './run-phase-authority-service.js'; import { getAdmissionControlService } from './admission-control-service.js'; @@ -413,9 +413,7 @@ export class MaintenanceService { (!artifact.retention.activeLeaseUntil || Date.parse(artifact.retention.activeLeaseUntil) <= current) ).length, - lastUsedAt: this.latestDate( - artifacts.map((artifact) => artifact.redaction.validatedAt) - ), + lastUsedAt: this.latestDate(artifacts.map((artifact) => artifact.redaction.validatedAt)), }; } catch { return { bytes: 0, itemCount: 0, cleanupEligibleCount: 0 }; diff --git a/server/src/services/notification-service.ts b/server/src/services/notification-service.ts index b41d9b15..f8b8506e 100644 --- a/server/src/services/notification-service.ts +++ b/server/src/services/notification-service.ts @@ -6,12 +6,10 @@ */ import { createLogger } from '../lib/logger.js'; -import * as fs from 'node:fs/promises'; -import * as path from 'node:path'; -import { withFileLock } from './file-lock.js'; import { SqliteDatabase, type SqliteConnectionOptions } from '../storage/sqlite/database.js'; import { SqliteNotificationRepository } from '../storage/sqlite/notification-repository.js'; import { getRuntimeDir } from '../utils/paths.js'; +import { NotificationFileRepository } from '../storage/notification-file-repository.js'; const DATA_DIR = getRuntimeDir(); @@ -98,17 +96,18 @@ export class NotificationService { private notifications: Notification[] = []; private subscriptions: ThreadSubscription[] = []; private loaded = false; - private readonly notificationsFile: string; - private readonly subscriptionsFile: string; + private readonly fileRepository: NotificationFileRepository; private readonly repository: SqliteNotificationRepository | null = null; private readonly sqliteDatabase: SqliteDatabase | null = null; private readonly ownsSqliteDatabase: boolean = false; constructor(options: NotificationServiceOptions = {}) { const dataDir = options.dataDir ?? DATA_DIR; - this.notificationsFile = options.notificationsFile ?? path.join(dataDir, 'notifications.json'); - this.subscriptionsFile = - options.subscriptionsFile ?? path.join(dataDir, 'thread-subscriptions.json'); + this.fileRepository = new NotificationFileRepository({ + dataDir, + notificationsFile: options.notificationsFile, + subscriptionsFile: options.subscriptionsFile, + }); const storageType = options.storageType ?? (process.env.VERITAS_STORAGE === 'sqlite' ? 'sqlite' : 'file'); @@ -130,18 +129,8 @@ export class NotificationService { return; } - try { - const nData = await fs.readFile(this.notificationsFile, 'utf-8'); - this.notifications = JSON.parse(nData); - } catch { - this.notifications = []; - } - try { - const sData = await fs.readFile(this.subscriptionsFile, 'utf-8'); - this.subscriptions = JSON.parse(sData); - } catch { - this.subscriptions = []; - } + this.notifications = await this.fileRepository.loadNotifications(); + this.subscriptions = await this.fileRepository.loadSubscriptions(); this.loaded = true; } @@ -151,9 +140,7 @@ export class NotificationService { return; } - await withFileLock(this.notificationsFile, async () => { - await fs.writeFile(this.notificationsFile, JSON.stringify(this.notifications, null, 2)); - }); + await this.fileRepository.saveNotifications(this.notifications); } private async saveSubscriptions(): Promise { @@ -162,9 +149,7 @@ export class NotificationService { return; } - await withFileLock(this.subscriptionsFile, async () => { - await fs.writeFile(this.subscriptionsFile, JSON.stringify(this.subscriptions, null, 2)); - }); + await this.fileRepository.saveSubscriptions(this.subscriptions); } /** diff --git a/server/src/services/outbound-integration-service.ts b/server/src/services/outbound-integration-service.ts index 44b1fb6f..6bbd1abf 100644 --- a/server/src/services/outbound-integration-service.ts +++ b/server/src/services/outbound-integration-service.ts @@ -1,4 +1,3 @@ -import fs from 'fs/promises'; import path from 'path'; import { nanoid } from 'nanoid'; import type { FeatureSettings } from '@veritas-kanban/shared'; @@ -9,6 +8,7 @@ import { type UrlValidationOptions, } from '../utils/url-validation.js'; import { getRuntimeDir } from '../utils/paths.js'; +import { JsonFileRepository } from '../storage/json-file-repository.js'; import { defaultDependencyCircuitExecutionService, integrationDependencyIdentity, @@ -201,7 +201,8 @@ async function readLimitedResponseText( } export class OutboundIntegrationService { - private readonly storageDir: string; + private readonly endpointsRepository: JsonFileRepository; + private readonly deliveriesRepository: JsonFileRepository; private readonly persist: boolean; private readonly audit: (event: AuditEvent) => Promise; private readonly dependencyExecution: DependencyCircuitExecutionService; @@ -210,7 +211,9 @@ export class OutboundIntegrationService { private deliveries: OutboundDeliveryAttempt[] = []; constructor(options: OutboundIntegrationServiceOptions = {}) { - this.storageDir = options.storageDir || path.join(getRuntimeDir(), 'outbound-integrations'); + const storageDir = options.storageDir || path.join(getRuntimeDir(), 'outbound-integrations'); + this.endpointsRepository = new JsonFileRepository(path.join(storageDir, 'endpoints.json')); + this.deliveriesRepository = new JsonFileRepository(path.join(storageDir, 'deliveries.json')); this.persist = options.persist ?? process.env.VITEST !== 'true'; this.audit = options.audit || auditLog; this.dependencyExecution = @@ -388,8 +391,7 @@ export class OutboundIntegrationService { const message = this.sanitizeError(err, endpoint.url); const circuitRejected = err instanceof DependencyRouteUnavailableError; const policyBlocked = err instanceof OutboundPolicyBlockError; - const dependencyResponse = - err instanceof OutboundDependencyResponseError ? err : undefined; + const dependencyResponse = err instanceof OutboundDependencyResponseError ? err : undefined; const status: OutboundDeliveryStatus = circuitRejected || policyBlocked ? 'blocked' @@ -412,7 +414,7 @@ export class OutboundIntegrationService { ? 'Dependency circuit rejected outbound delivery.' : policyBlocked ? err.message - : message, + : message, }); return { ok: false, @@ -518,19 +520,15 @@ export class OutboundIntegrationService { return; } - await fs.mkdir(this.storageDir, { recursive: true }); - try { - const raw = await fs.readFile(this.endpointsPath, 'utf-8'); - const parsed = JSON.parse(raw) as OutboundEndpointRecord[]; + const parsed = await this.endpointsRepository.read(); this.endpoints = new Map(parsed.map((endpoint) => [endpoint.id, endpoint])); } catch { this.endpoints = new Map(); } try { - const raw = await fs.readFile(this.deliveriesPath, 'utf-8'); - this.deliveries = JSON.parse(raw) as OutboundDeliveryAttempt[]; + this.deliveries = await this.deliveriesRepository.read(); if (this.deliveries.length > MAX_DELIVERIES) { this.deliveries = this.deliveries.slice(-MAX_DELIVERIES); } @@ -541,28 +539,14 @@ export class OutboundIntegrationService { this.loaded = true; } - private get endpointsPath(): string { - return path.join(this.storageDir, 'endpoints.json'); - } - - private get deliveriesPath(): string { - return path.join(this.storageDir, 'deliveries.json'); - } - private async saveEndpoints(): Promise { if (!this.persist) return; - await fs.mkdir(this.storageDir, { recursive: true }); - await fs.writeFile( - this.endpointsPath, - JSON.stringify(Array.from(this.endpoints.values()), null, 2), - 'utf-8' - ); + await this.endpointsRepository.write(Array.from(this.endpoints.values())); } private async saveDeliveries(): Promise { if (!this.persist) return; - await fs.mkdir(this.storageDir, { recursive: true }); - await fs.writeFile(this.deliveriesPath, JSON.stringify(this.deliveries, null, 2), 'utf-8'); + await this.deliveriesRepository.write(this.deliveries); } private async recordDelivery(input: { diff --git a/server/src/services/queue-intake-monitor-service.ts b/server/src/services/queue-intake-monitor-service.ts index 1c5c8a77..a30bc4cc 100644 --- a/server/src/services/queue-intake-monitor-service.ts +++ b/server/src/services/queue-intake-monitor-service.ts @@ -1,5 +1,4 @@ import { execFile } from 'node:child_process'; -import fs from 'node:fs/promises'; import path from 'node:path'; import { promisify } from 'node:util'; import { nanoid } from 'nanoid'; @@ -28,6 +27,7 @@ import type { import { createLogger } from '../lib/logger.js'; import { NotFoundError, ValidationError } from '../middleware/error-handler.js'; import { getRuntimeDir } from '../utils/paths.js'; +import { JsonFileRepository } from '../storage/json-file-repository.js'; import { getAgentBudgetService, type AgentBudgetService } from './agent-budget-service.js'; import { getBreaker } from './circuit-registry.js'; import { @@ -121,7 +121,7 @@ interface GateContext { } export class QueueIntakeMonitorService { - private readonly storeFile: string; + private readonly repository: JsonFileRepository; private readonly githubExec: (args: string[]) => Promise; private readonly watcherPolicyService: WatcherPolicyService; private readonly sandboxPolicyService: SandboxPolicyService; @@ -136,7 +136,8 @@ export class QueueIntakeMonitorService { private readonly runningMonitors = new Set(); constructor(options: QueueIntakeMonitorServiceOptions = {}) { - this.storeFile = options.storeFile ?? path.join(getRuntimeDir(), 'queue-monitors.json'); + const storeFile = options.storeFile ?? path.join(getRuntimeDir(), 'queue-monitors.json'); + this.repository = new JsonFileRepository(storeFile); this.githubExec = options.githubExec ?? defaultGhExec; this.watcherPolicyService = options.watcherPolicyService ?? new WatcherPolicyService(); this.sandboxPolicyService = options.sandboxPolicyService ?? getSandboxPolicyService(); @@ -965,8 +966,7 @@ export class QueueIntakeMonitorService { private async ensureLoaded(): Promise { if (this.store) return; try { - const raw = await fs.readFile(this.storeFile, 'utf-8'); - const parsed = JSON.parse(raw) as Partial; + const parsed = await this.repository.read(); this.store = normalizeStore(parsed); } catch { this.store = defaultStore(); @@ -979,8 +979,7 @@ export class QueueIntakeMonitorService { } private async saveStore(): Promise { - await fs.mkdir(path.dirname(this.storeFile), { recursive: true }); - await fs.writeFile(this.storeFile, JSON.stringify(this.store, null, 2), 'utf-8'); + await this.repository.write(this.currentStore()); } } diff --git a/server/src/services/run-session-share-service.ts b/server/src/services/run-session-share-service.ts index e31d01ff..2edbf997 100644 --- a/server/src/services/run-session-share-service.ts +++ b/server/src/services/run-session-share-service.ts @@ -1,4 +1,3 @@ -import fs from 'node:fs/promises'; import path from 'node:path'; import { nanoid } from 'nanoid'; import type { @@ -19,6 +18,7 @@ import type { UpdateRunSessionShareInput, } from '@veritas-kanban/shared'; import { getDataDir } from '../utils/paths.js'; +import { JsonFileRepository } from '../storage/json-file-repository.js'; import { validatePathSegment } from '../utils/sanitize.js'; import { redactString } from '../lib/redact.js'; import { ConflictError, ForbiddenError, NotFoundError } from '../middleware/error-handler.js'; @@ -55,15 +55,16 @@ const MAX_EVENT_HISTORY = 5000; const LOG_CONTEXT_LIMIT = 4000; export class RunSessionShareService { - private readonly filePath: string; + private readonly repository: JsonFileRepository; private readonly taskService: TaskService; private readonly agentService: typeof clawdbotAgentService; private readonly approvalBroker: RunApprovalBrokerService; private state: RunSessionShareState | null = null; constructor(options: RunSessionShareServiceOptions = {}) { - this.filePath = + const filePath = options.filePath ?? path.join(getDataDir(), 'storage', 'run-session-shares.json'); + this.repository = new JsonFileRepository(filePath, { trailingNewline: true }); this.taskService = options.taskService ?? getTaskService(); this.agentService = options.agentService ?? clawdbotAgentService; this.approvalBroker = options.approvalBroker ?? getRunApprovalBrokerService(); @@ -479,8 +480,7 @@ export class RunSessionShareService { private async loadState(): Promise { if (this.state) return this.state; try { - const raw = await fs.readFile(this.filePath, 'utf8'); - this.state = JSON.parse(raw) as RunSessionShareState; + this.state = await this.repository.read(); } catch { this.state = { shares: [], events: [], forks: [] }; } @@ -489,8 +489,7 @@ export class RunSessionShareService { private async saveState(state: RunSessionShareState): Promise { state.events = state.events.slice(-MAX_EVENT_HISTORY); - await fs.mkdir(path.dirname(this.filePath), { recursive: true }); - await fs.writeFile(this.filePath, `${JSON.stringify(state, null, 2)}\n`, 'utf8'); + await this.repository.write(state); this.state = state; } } diff --git a/server/src/services/search-service.ts b/server/src/services/search-service.ts index c6f17bbd..aba8320a 100644 --- a/server/src/services/search-service.ts +++ b/server/src/services/search-service.ts @@ -1,5 +1,4 @@ import { execFile } from 'node:child_process'; -import fs from 'node:fs/promises'; import path from 'node:path'; import type { AnyTelemetryEvent } from '@veritas-kanban/shared'; import { createLogger } from '../lib/logger.js'; @@ -13,6 +12,10 @@ import { getWorkProductService, WorkProductService } from './work-product-servic import { getWorkflowService, WorkflowService } from './workflow-service.js'; import { getWorkflowRunService, WorkflowRunService } from './workflow-run-service.js'; import { getStorageRoot } from '../utils/paths.js'; +import { + SearchFileRepository, + type SearchFileDescriptor, +} from '../storage/search-file-repository.js'; const log = createLogger('search-service'); @@ -80,10 +83,7 @@ interface SearchSource { extensions: readonly string[]; } -interface SearchFile { - path: string; - mtimeMs: number; -} +type SearchFile = SearchFileDescriptor; interface ScoreDetails { score: number; @@ -118,6 +118,7 @@ const MAX_TELEMETRY_SEARCH_EVENTS = : 2_000; class SearchService { + private readonly fileRepository = new SearchFileRepository(); async search(request: SearchRequest): Promise { const started = Date.now(); const query = request.query.trim(); @@ -417,7 +418,7 @@ class SearchService { ): Promise { let content: string; try { - content = await fs.readFile(file.path, 'utf-8'); + content = await this.fileRepository.readText(file.path); } catch { return null; } @@ -1130,45 +1131,11 @@ class SearchService { dir: string, extensions: readonly string[] = DEFAULT_FILE_EXTENSIONS ): Promise { - const files: SearchFile[] = []; - const allowedExtensions = new Set(extensions.map((extension) => extension.toLowerCase())); - - const visit = async (currentDir: string): Promise => { - if (files.length >= MAX_FILES_PER_SOURCE) return; - - let entries; - try { - entries = await fs.readdir(currentDir, { withFileTypes: true }); - } catch { - return; - } - - for (const entry of entries) { - if (files.length >= MAX_FILES_PER_SOURCE) return; - if (SKIPPED_DIRECTORIES.has(entry.name)) continue; - - const fullPath = path.join(currentDir, entry.name); - if (entry.isDirectory()) { - await visit(fullPath); - continue; - } - - if (!entry.isFile()) continue; - - const extension = path.extname(entry.name).toLowerCase(); - if (!allowedExtensions.has(extension)) continue; - - try { - const stats = await fs.stat(fullPath); - files.push({ path: fullPath, mtimeMs: stats.mtimeMs }); - } catch { - continue; - } - } - }; - - await visit(dir); - return files.sort((a, b) => b.mtimeMs - a.mtimeMs).slice(0, MAX_FILES_PER_SOURCE); + return this.fileRepository.listFiles(dir, { + extensions, + maxFiles: MAX_FILES_PER_SOURCE, + skippedDirectories: SKIPPED_DIRECTORIES, + }); } private extractTitle( diff --git a/server/src/services/task-identity-diagnostics.ts b/server/src/services/task-identity-diagnostics.ts index 6e1dc629..eb4fed4f 100644 --- a/server/src/services/task-identity-diagnostics.ts +++ b/server/src/services/task-identity-diagnostics.ts @@ -1,6 +1,6 @@ -import fs from 'fs/promises'; import path from 'path'; import matter from '../utils/frontmatter.js'; +import { TaskIdentityFileRepository } from '../storage/task-identity-file-repository.js'; export type TaskIdentityLocation = 'active' | 'backlog' | 'archive'; export type TaskIdentityConflictKind = 'task-id' | 'business-id'; @@ -60,6 +60,7 @@ const EMPTY_DIAGNOSTICS: TaskIdentityDiagnostics = { conflictCount: 0, conflicts: [], }; +const fileRepository = new TaskIdentityFileRepository(); function taskIdFromFilename(filename: string): string { return filename.replace(/\.md$/, '').split('-')[0] ?? ''; @@ -121,22 +122,11 @@ async function readMarkdownSources( source: TaskIdentityScanSource, rootDir: string ): Promise { - let files: string[]; - try { - files = await fs.readdir(source.dir); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') { - return []; - } - throw error; - } - - const markdownFiles = files.filter((filename) => filename.endsWith('.md')).sort(); const results: TaskIdentitySource[] = []; - for (const filename of markdownFiles) { - const filepath = path.join(source.dir, filename); - const content = await fs.readFile(filepath, 'utf-8'); + for (const { filename, absolutePath, content } of await fileRepository.readMarkdownFiles( + source.dir + )) { const parsed = matter(content); const frontmatter = parsed.data as Record; const taskId = @@ -148,7 +138,7 @@ async function readMarkdownSources( results.push({ location: source.location, - path: path.relative(rootDir, filepath), + path: path.relative(rootDir, absolutePath), filename, taskId, title: typeof frontmatter.title === 'string' ? frontmatter.title : undefined, diff --git a/server/src/services/trace-service.ts b/server/src/services/trace-service.ts index 7f9de05f..9428f199 100644 --- a/server/src/services/trace-service.ts +++ b/server/src/services/trace-service.ts @@ -1,5 +1,3 @@ -import fs from 'fs/promises'; -import path from 'path'; import type { AgentRunTrace, AgentRunTraceMetadata, @@ -8,8 +6,9 @@ import type { AgentType, } from '@veritas-kanban/shared'; import { getTelemetryService } from './telemetry-service.js'; -import { validatePathSegment, ensureWithinBase } from '../utils/sanitize.js'; +import { validatePathSegment } from '../utils/sanitize.js'; import { getTracesDir } from '../utils/paths.js'; +import { TraceFileRepository } from '../storage/trace-file-repository.js'; const TRACES_DIR = getTracesDir(); @@ -21,11 +20,11 @@ export type Trace = AgentRunTrace; const activeTraces = new Map(); export class TraceService { - private tracesDir: string; + private readonly repository: TraceFileRepository; private enabled: boolean = false; constructor() { - this.tracesDir = TRACES_DIR; + this.repository = new TraceFileRepository(TRACES_DIR); this.init(); } @@ -36,7 +35,7 @@ export class TraceService { this.enabled = config.traces ?? false; if (this.enabled) { - await fs.mkdir(this.tracesDir, { recursive: true }); + await this.repository.ensureReady(); } } @@ -54,7 +53,7 @@ export class TraceService { this.enabled = enabled; if (enabled) { // Intentionally silent: best-effort directory creation - fs.mkdir(this.tracesDir, { recursive: true }).catch(() => {}); + this.repository.ensureReady().catch(() => {}); } } @@ -180,10 +179,7 @@ export class TraceService { // Try to load from disk try { - const filepath = path.join(this.tracesDir, `${attemptId}.json`); - ensureWithinBase(this.tracesDir, filepath); - const content = await fs.readFile(filepath, 'utf-8'); - return JSON.parse(content) as Trace; + return await this.repository.read(attemptId); } catch { // Intentionally silent: trace file may not exist on disk return null; @@ -208,20 +204,9 @@ export class TraceService { // Load completed traces from disk try { - const files = await fs.readdir(this.tracesDir); - for (const file of files) { - if (!file.endsWith('.json')) continue; - - try { - const filepath = path.join(this.tracesDir, file); - ensureWithinBase(this.tracesDir, filepath); - const content = await fs.readFile(filepath, 'utf-8'); - const trace = JSON.parse(content) as Trace; - if (trace.taskId === taskId && !activeTraces.has(trace.traceId)) { - traces.push(trace); - } - } catch { - // Skip invalid files + for (const trace of await this.repository.list()) { + if (trace.taskId === taskId && !activeTraces.has(trace.traceId)) { + traces.push(trace); } } } catch { @@ -236,13 +221,7 @@ export class TraceService { * Save a trace to disk */ private async saveTrace(trace: Trace): Promise { - // Validate traceId to prevent path traversal - validatePathSegment(trace.traceId); - - await fs.mkdir(this.tracesDir, { recursive: true }); - const filepath = path.join(this.tracesDir, `${trace.traceId}.json`); - ensureWithinBase(this.tracesDir, filepath); - await fs.writeFile(filepath, JSON.stringify(trace, null, 2), 'utf-8'); + await this.repository.write(trace); } } diff --git a/server/src/services/work-product-service.ts b/server/src/services/work-product-service.ts index 525edaa7..63d19051 100644 --- a/server/src/services/work-product-service.ts +++ b/server/src/services/work-product-service.ts @@ -1,4 +1,3 @@ -import fs from 'node:fs/promises'; import path from 'node:path'; import { randomUUID } from 'node:crypto'; import type { @@ -20,6 +19,7 @@ import type { import { ForbiddenError } from '../middleware/error-handler.js'; import { SqliteDatabase, type SqliteConnectionOptions } from '../storage/sqlite/database.js'; import { SqliteWorkProductRepository } from '../storage/sqlite/work-product-repository.js'; +import { JsonFileRepository } from '../storage/json-file-repository.js'; import { getRuntimeDir } from '../utils/paths.js'; const DATA_DIR = getRuntimeDir(); @@ -40,7 +40,7 @@ export interface WorkProductServiceOptions { } export class WorkProductService { - private readonly filePath: string; + private readonly fileRepository: JsonFileRepository; private readonly versionLimit: number; private readonly repository: SqliteWorkProductRepository | null = null; private readonly sqliteDatabase: SqliteDatabase | null = null; @@ -50,7 +50,8 @@ export class WorkProductService { constructor(options: WorkProductServiceOptions = {}) { const dataDir = options.dataDir ?? DATA_DIR; - this.filePath = options.filePath ?? path.join(dataDir, 'work-products.json'); + const filePath = options.filePath ?? path.join(dataDir, 'work-products.json'); + this.fileRepository = new JsonFileRepository(filePath); this.versionLimit = options.versionLimit ?? DEFAULT_VERSION_LIMIT; const storageType = options.storageType ?? (process.env.VERITAS_STORAGE === 'sqlite' ? 'sqlite' : 'file'); @@ -809,8 +810,7 @@ export class WorkProductService { if (this.loaded) return; try { - const raw = await fs.readFile(this.filePath, 'utf-8'); - const parsed = JSON.parse(raw) as WorkProductFileState | WorkProduct[]; + const parsed = await this.fileRepository.read(); this.fileState = Array.isArray(parsed) ? { products: parsed, versions: [] } : parsed; } catch { this.fileState = { products: [], versions: [] }; @@ -820,8 +820,7 @@ export class WorkProductService { } private async saveFileState(): Promise { - await fs.mkdir(path.dirname(this.filePath), { recursive: true }); - await fs.writeFile(this.filePath, JSON.stringify(this.fileState, null, 2)); + await this.fileRepository.write(this.fileState); } private createVersion( diff --git a/server/src/storage/audit-file-repository.ts b/server/src/storage/audit-file-repository.ts new file mode 100644 index 00000000..5f524826 --- /dev/null +++ b/server/src/storage/audit-file-repository.ts @@ -0,0 +1,108 @@ +import crypto from 'node:crypto'; +import { appendFile, mkdir, readFile } from 'node:fs/promises'; +import path from 'node:path'; +import readline from 'node:readline'; +import { createReadStream } from './fs-helpers.js'; + +export interface StoredAuditEntry { + timestamp: string; + action: string; + actor: string; + resource?: string; + details?: Record; + integrity: string; +} + +export interface AuditFileVerifyResult { + valid: boolean; + entries: number; + firstBroken?: number; +} + +function sha256(data: string): string { + return crypto.createHash('sha256').update(data, 'utf8').digest('hex'); +} + +export class AuditFileRepository { + constructor(private readonly directory: string) {} + + getMonthlyLogPath(date: Date): string { + const yyyy = date.getFullYear(); + const mm = String(date.getMonth() + 1).padStart(2, '0'); + return path.join(this.directory, `audit-${yyyy}-${mm}.log`); + } + + ensureReady(): Promise { + return mkdir(this.directory, { recursive: true }).then(() => undefined); + } + + async getLastHash(filePath: string): Promise { + try { + const content = await readFile(filePath, 'utf8'); + const lines = content.trimEnd().split('\n').filter(Boolean); + return lines.length > 0 ? sha256(lines[lines.length - 1]) : ''; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return ''; + throw error; + } + } + + append(filePath: string, line: string): Promise { + return appendFile(filePath, `${line}\n`, 'utf8'); + } + + async verify(filePath: string): Promise { + const stream = createReadStream(filePath, { encoding: 'utf8' }); + const reader = readline.createInterface({ input: stream, crlfDelay: Infinity }); + let previousHash = ''; + let lineIndex = 0; + let totalEntries = 0; + + try { + for await (const line of reader) { + const trimmed = line.trim(); + if (!trimmed) { + lineIndex += 1; + continue; + } + + totalEntries += 1; + let entry: StoredAuditEntry; + try { + entry = JSON.parse(trimmed) as StoredAuditEntry; + } catch { + return { valid: false, entries: totalEntries, firstBroken: lineIndex }; + } + + if (entry.integrity !== previousHash) { + return { valid: false, entries: totalEntries, firstBroken: lineIndex }; + } + previousHash = sha256(trimmed); + lineIndex += 1; + } + return { valid: true, entries: totalEntries }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return { valid: true, entries: 0 }; + throw error; + } finally { + reader.close(); + stream.destroy(); + } + } + + async readRecent(filePath: string, limit: number): Promise { + try { + const content = await readFile(filePath, 'utf8'); + return content + .trimEnd() + .split('\n') + .filter(Boolean) + .slice(-limit) + .reverse() + .map((line) => JSON.parse(line) as StoredAuditEntry); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw error; + } + } +} diff --git a/server/src/storage/fs-helpers.ts b/server/src/storage/fs-helpers.ts index 8f91e99a..cb073808 100644 --- a/server/src/storage/fs-helpers.ts +++ b/server/src/storage/fs-helpers.ts @@ -14,12 +14,15 @@ import type { FSWatcher } from 'node:fs'; import { EventEmitter } from 'node:events'; import { access, + copyFile as copyFileAsync, lstat as lstatAsync, mkdir as mkdirAsync, readFile as readFileAsync, readdir as readdirAsync, rename as renameAsync, rm as rmAsync, + stat as statAsync, + statfs as statfsAsync, unlink as unlinkAsync, writeFile as writeFileAsync, } from 'node:fs/promises'; @@ -70,12 +73,16 @@ export const createWriteStream = fs.createWriteStream; // --------------------------------------------------------------------------- export const mkdir = mkdirAsync; +export { access }; +export const copyFile = copyFileAsync; export const lstat = lstatAsync; export const readFile = readFileAsync; export const readdir = readdirAsync; export const realpath = fs.promises.realpath; export const rename = renameAsync; export const rm = rmAsync; +export const stat = statAsync; +export const statfs = statfsAsync; export const unlink = unlinkAsync; export const writeFile = writeFileAsync; diff --git a/server/src/storage/json-file-repository.ts b/server/src/storage/json-file-repository.ts new file mode 100644 index 00000000..097c05f9 --- /dev/null +++ b/server/src/storage/json-file-repository.ts @@ -0,0 +1,35 @@ +import path from 'node:path'; +import { withFileLock } from '../services/file-lock.js'; +import { atomicWriteFile, mkdir, readFile } from './fs-helpers.js'; + +export interface JsonFileRepositoryOptions { + trailingNewline?: boolean; +} + +/** + * Small persistence boundary for JSON-backed service state. + * + * Reads intentionally surface filesystem and parse errors so each service can + * retain its existing fallback policy. Writes are serialized across processes + * and replace the destination atomically. + */ +export class JsonFileRepository { + private readonly trailingNewline: boolean; + + constructor( + private readonly filePath: string, + options: JsonFileRepositoryOptions = {} + ) { + this.trailingNewline = options.trailingNewline ?? false; + } + + async read(): Promise { + return JSON.parse(await readFile(this.filePath, 'utf8')) as T; + } + + async write(value: T): Promise { + await mkdir(path.dirname(this.filePath), { recursive: true }); + const serialized = JSON.stringify(value, null, 2) + (this.trailingNewline ? '\n' : ''); + await withFileLock(this.filePath, () => atomicWriteFile(this.filePath, serialized, 'utf8')); + } +} diff --git a/server/src/storage/notification-file-repository.ts b/server/src/storage/notification-file-repository.ts new file mode 100644 index 00000000..af99bec3 --- /dev/null +++ b/server/src/storage/notification-file-repository.ts @@ -0,0 +1,49 @@ +import { readFile, writeFile } from 'node:fs/promises'; +import path from 'node:path'; +import { withFileLock } from '../services/file-lock.js'; + +export interface NotificationFileRepositoryOptions { + dataDir: string; + notificationsFile?: string; + subscriptionsFile?: string; +} + +export class NotificationFileRepository { + private readonly notificationsFile: string; + private readonly subscriptionsFile: string; + + constructor(options: NotificationFileRepositoryOptions) { + this.notificationsFile = + options.notificationsFile ?? path.join(options.dataDir, 'notifications.json'); + this.subscriptionsFile = + options.subscriptionsFile ?? path.join(options.dataDir, 'thread-subscriptions.json'); + } + + loadNotifications(): Promise { + return this.loadArray(this.notificationsFile); + } + + loadSubscriptions(): Promise { + return this.loadArray(this.subscriptionsFile); + } + + saveNotifications(notifications: T[]): Promise { + return this.saveArray(this.notificationsFile, notifications); + } + + saveSubscriptions(subscriptions: T[]): Promise { + return this.saveArray(this.subscriptionsFile, subscriptions); + } + + private async loadArray(filePath: string): Promise { + try { + return JSON.parse(await readFile(filePath, 'utf8')) as T[]; + } catch { + return []; + } + } + + private saveArray(filePath: string, values: T[]): Promise { + return withFileLock(filePath, () => writeFile(filePath, JSON.stringify(values, null, 2))); + } +} diff --git a/server/src/storage/search-file-repository.ts b/server/src/storage/search-file-repository.ts new file mode 100644 index 00000000..6e79fb43 --- /dev/null +++ b/server/src/storage/search-file-repository.ts @@ -0,0 +1,61 @@ +import path from 'node:path'; +import { readFile, readdir, lstat } from './fs-helpers.js'; + +export interface SearchFileDescriptor { + path: string; + mtimeMs: number; +} + +export interface SearchFileListOptions { + extensions: readonly string[]; + maxFiles: number; + skippedDirectories: ReadonlySet; +} + +export class SearchFileRepository { + readText(filePath: string): Promise { + return readFile(filePath, 'utf8'); + } + + async listFiles(dir: string, options: SearchFileListOptions): Promise { + const files: SearchFileDescriptor[] = []; + const allowedExtensions = new Set( + options.extensions.map((extension) => extension.toLowerCase()) + ); + + const visit = async (currentDir: string): Promise => { + if (files.length >= options.maxFiles) return; + + let entries; + try { + entries = await readdir(currentDir, { withFileTypes: true }); + } catch { + return; + } + + for (const entry of entries) { + if (files.length >= options.maxFiles) return; + if (options.skippedDirectories.has(entry.name)) continue; + + const fullPath = path.join(currentDir, entry.name); + if (entry.isDirectory()) { + await visit(fullPath); + continue; + } + if (!entry.isFile() || !allowedExtensions.has(path.extname(entry.name).toLowerCase())) { + continue; + } + + try { + const stats = await lstat(fullPath); + files.push({ path: fullPath, mtimeMs: stats.mtimeMs }); + } catch { + // File may disappear while the index is being refreshed. + } + } + }; + + await visit(dir); + return files.sort((a, b) => b.mtimeMs - a.mtimeMs).slice(0, options.maxFiles); + } +} diff --git a/server/src/storage/task-identity-file-repository.ts b/server/src/storage/task-identity-file-repository.ts new file mode 100644 index 00000000..88cca644 --- /dev/null +++ b/server/src/storage/task-identity-file-repository.ts @@ -0,0 +1,31 @@ +import path from 'node:path'; +import { readFile, readdir } from './fs-helpers.js'; + +export interface MarkdownSourceFile { + absolutePath: string; + filename: string; + content: string; +} + +export class TaskIdentityFileRepository { + async readMarkdownFiles(directory: string): Promise { + let files: string[]; + try { + files = await readdir(directory); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw error; + } + + const results: MarkdownSourceFile[] = []; + for (const filename of files.filter((entry) => entry.endsWith('.md')).sort()) { + const absolutePath = path.join(directory, filename); + results.push({ + absolutePath, + filename, + content: await readFile(absolutePath, 'utf8'), + }); + } + return results; + } +} diff --git a/server/src/storage/trace-file-repository.ts b/server/src/storage/trace-file-repository.ts new file mode 100644 index 00000000..2ed95978 --- /dev/null +++ b/server/src/storage/trace-file-repository.ts @@ -0,0 +1,47 @@ +import path from 'node:path'; +import type { AgentRunTrace } from '@veritas-kanban/shared'; +import { ensureWithinBase, validatePathSegment } from '../utils/sanitize.js'; +import { mkdir, readdir } from './fs-helpers.js'; +import { JsonFileRepository } from './json-file-repository.js'; + +export class TraceFileRepository { + constructor(private readonly tracesDir: string) {} + + ensureReady(): Promise { + return mkdir(this.tracesDir, { recursive: true }).then(() => undefined); + } + + read(traceId: string): Promise { + return this.repositoryFor(traceId).read(); + } + + async list(): Promise { + let files: string[]; + try { + files = await readdir(this.tracesDir); + } catch { + return []; + } + + const traces: AgentRunTrace[] = []; + for (const file of files) { + if (!file.endsWith('.json')) continue; + try { + traces.push(await this.repositoryFor(file.slice(0, -'.json'.length)).read()); + } catch { + // Ignore invalid or concurrently removed trace files. + } + } + return traces; + } + + write(trace: AgentRunTrace): Promise { + return this.repositoryFor(trace.traceId).write(trace); + } + + private repositoryFor(traceId: string): JsonFileRepository { + validatePathSegment(traceId); + const filePath = ensureWithinBase(this.tracesDir, path.join(this.tracesDir, `${traceId}.json`)); + return new JsonFileRepository(filePath); + } +}