feat(recording): consolidate recording pipeline on server B

apps/worker-recording/src/main.ts

@@ -1,13 +1,15 @@
 import { createLogger } from '@lisglosips/observability';
 import { PrismaClient } from '@lisglosips/database';
-import { PrismaRecordingStore, RecordingTransferService, SshRecordingClient } from './transfer.js';
+import { LocalRecordingClient, PrismaRecordingStore, RecordingTransferService, SshRecordingClient } from './transfer.js';
 
 const serviceName = process.env.LISGLOSIPS_SERVICE_NAME ?? 'worker-recording';
 const logger = createLogger(serviceName, process.env.LISGLOSIPS_LOG_LEVEL ?? 'info');
 
+const sourceMode = process.env.RECORDING_SOURCE_MODE ?? 'ssh';
 const remoteHost = process.env.RECORDING_REMOTE_HOST ?? 'lisglosips-a';
 const sshConfig = process.env.RECORDING_SSH_CONFIG;
 const remoteReadyDir = process.env.RECORDING_REMOTE_READY_DIR ?? '/dev/shm/voip_rec/ready';
+const localReadyDir = process.env.RECORDING_LOCAL_READY_DIR ?? '/dev/shm/voip_rec/ready';
 const localRoot = process.env.RECORDING_LOCAL_ROOT ?? '/data/recordings';
 const scanIntervalMs = Number.parseInt(process.env.RECORDING_SCAN_INTERVAL_MS ?? '10000', 10);
 const maxFilesPerScan = Number.parseInt(process.env.RECORDING_MAX_FILES_PER_SCAN ?? '50', 10);
@@ -17,8 +19,14 @@ let shuttingDown = false;
 
 async function main(): Promise<void> {
   const prisma = new PrismaClient();
-  const remote = new SshRecordingClient({ host: remoteHost, sshConfig, remoteReadyDir });
-  const service = new RecordingTransferService(remote, new PrismaRecordingStore(prisma), {
+  if (sourceMode !== 'ssh' && sourceMode !== 'local') {
+    throw new Error(`Unsupported RECORDING_SOURCE_MODE: ${sourceMode}`);
+  }
+  const source =
+    sourceMode === 'local'
+      ? new LocalRecordingClient({ readyDir: localReadyDir })
+      : new SshRecordingClient({ host: remoteHost, sshConfig, remoteReadyDir });
+  const service = new RecordingTransferService(source, new PrismaRecordingStore(prisma), {
     localRoot,
     deleteSourceAfterCopy,
     maxFilesPerScan
@@ -27,8 +35,9 @@ async function main(): Promise<void> {
   await prisma.$connect();
   logger.info(
     {
-      remoteHost,
-      remoteReadyDir,
+      sourceMode,
+      sourceDir: sourceMode === 'local' ? localReadyDir : remoteReadyDir,
+      ...(sourceMode === 'ssh' ? { remoteHost } : {}),
       localRoot,
       scanIntervalMs,
       maxFilesPerScan,

apps/worker-recording/src/transfer.spec.ts

@@ -1,4 +1,4 @@
-import { mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises';
+import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises';
 import os from 'node:os';
 import path from 'node:path';
 
@@ -6,6 +6,7 @@ import { describe, expect, it } from 'vitest';
 
 import {
   callIdFromStorageKey,
+  LocalRecordingClient,
   normalizeRemoteRelativePath,
   type PulledRecordingFile,
   type RecordingStore,
@@ -90,6 +91,35 @@ describe('recording transfer', () => {
     }
   });
 
+  it('lists, copies, and deletes ready files from a local source directory', async () => {
+    const root = await mkdtemp(path.join(os.tmpdir(), 'lisglosips-local-rec-'));
+    const readyDir = path.join(root, 'ready');
+    const outputDir = path.join(root, 'recordings');
+    const sourcePath = path.join(readyDir, '2026', '08', '27', 'call.wav.ready');
+    const client = new LocalRecordingClient({ readyDir });
+    const store = new MemoryStore();
+    const service = new RecordingTransferService(client, store, {
+      localRoot: outputDir,
+      deleteSourceAfterCopy: true,
+      maxFilesPerScan: 10
+    });
+
+    try {
+      await mkdir(path.dirname(sourcePath), { recursive: true });
+      await writeFile(sourcePath, 'local-audio');
+      await writeFile(path.join(readyDir, 'ignored.part'), 'incomplete');
+
+      await expect(client.listReadyFiles()).resolves.toEqual([
+        { relativePath: '2026/08/27/call.wav.ready', bytes: 11 }
+      ]);
+      await expect(service.scanOnce()).resolves.toEqual({ moved: 1, skipped: 0, failed: 0 });
+      await expect(readFile(path.join(outputDir, '2026', '08', '27', 'call.wav'), 'utf8')).resolves.toBe('local-audio');
+      await expect(stat(sourcePath)).rejects.toThrow();
+    } finally {
+      await rm(root, { recursive: true, force: true });
+    }
+  });
+
   it('keeps source file when local size verification fails', async () => {
     const root = await mkdtemp(path.join(os.tmpdir(), 'lisglosips-rec-'));
     const remote = new MemoryRemote(new Map([['bad.wav.ready', Buffer.from('abc')]]));

apps/worker-recording/src/transfer.ts

@@ -1,6 +1,6 @@
 import { createHash, randomUUID } from 'node:crypto';
 import { createReadStream, createWriteStream } from 'node:fs';
-import { mkdir, rename, rm, stat } from 'node:fs/promises';
+import { mkdir, readdir, rename, rm, stat } from 'node:fs/promises';
 import path from 'node:path';
 import { spawn } from 'node:child_process';
 import { pipeline } from 'node:stream/promises';
@@ -149,6 +149,31 @@ export interface SshRecordingClientOptions {
   remoteReadyDir: string;
 }
 
+export interface LocalRecordingClientOptions {
+  readyDir: string;
+}
+
+/** Reads finalized recordings from a directory on the same host as the worker. */
+export class LocalRecordingClient implements RemoteRecordingClient {
+  constructor(private readonly options: LocalRecordingClientOptions) {}
+
+  async listReadyFiles(): Promise<RemoteRecordingFile[]> {
+    const files: RemoteRecordingFile[] = [];
+    await walkReadyFiles(this.options.readyDir, '', files);
+    return files.sort((left, right) => left.relativePath.localeCompare(right.relativePath));
+  }
+
+  async pullFile(file: RemoteRecordingFile, localPath: string): Promise<void> {
+    const sourcePath = safeJoin(this.options.readyDir, normalizeRemoteRelativePath(file.relativePath));
+    await pipeline(createReadStream(sourcePath), createWriteStream(localPath, { mode: 0o640 }));
+  }
+
+  async deleteFile(file: RemoteRecordingFile): Promise<void> {
+    const sourcePath = safeJoin(this.options.readyDir, normalizeRemoteRelativePath(file.relativePath));
+    await rm(sourcePath);
+  }
+}
+
 export class SshRecordingClient implements RemoteRecordingClient {
   constructor(private readonly options: SshRecordingClientOptions) {}
 
@@ -225,6 +250,33 @@ function safeJoin(root: string, relativePath: string): string {
   return resolvedPath;
 }
 
+async function walkReadyFiles(root: string, relativeDir: string, files: RemoteRecordingFile[]): Promise<void> {
+  const directory = relativeDir ? safeJoin(root, relativeDir) : path.resolve(root);
+  let entries;
+  try {
+    entries = await readdir(directory, { withFileTypes: true });
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
+      return;
+    }
+    throw error;
+  }
+
+  for (const entry of entries) {
+    const relativePath = relativeDir ? `${relativeDir}/${entry.name}` : entry.name;
+    if (entry.isDirectory()) {
+      await walkReadyFiles(root, relativePath, files);
+      continue;
+    }
+    if (!entry.isFile() || !entry.name.endsWith('.ready')) {
+      continue;
+    }
+    const normalized = normalizeRemoteRelativePath(relativePath);
+    const fileStat = await stat(safeJoin(root, normalized));
+    files.push({ relativePath: normalized, bytes: fileStat.size });
+  }
+}
+
 function remotePathFor(baseDir: string, relativePath: string): string {
   return `${baseDir.replace(/\/+$/, '')}/${normalizeRemoteRelativePath(relativePath)}`;
 }