Function streamReadableToFile

src/daemon/artifact-download.ts:43–108

(
  source: NodeJS.ReadableStream,
  destPath: string,
)

}

export function streamReadableToFile(
  source: NodeJS.ReadableStream,
  destPath: string,
): Promise<void> {
  return new Promise((resolve, reject) => {
    let settled = false;
    let bytesWritten = 0;
    let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
    const output = fs.createWriteStream(destPath);

    const settle = (error?: unknown) => {
      if (settled) return;
      settled = true;
      if (timeoutHandle) clearTimeout(timeoutHandle);
      if (error) {
        void removePartialFile(output, destPath).finally(() => reject(error));
        return;
      }
      resolve();
    };
    const armTimeout = () => {
      if (timeoutHandle) clearTimeout(timeoutHandle);
      timeoutHandle = setTimeout(() => {
        const error = new AppError(
          'COMMAND_FAILED',
          'Artifact transfer timed out due to inactivity',
          {
            timeoutMs: REQUEST_IDLE_TIMEOUT_MS,
          },
        );
        if ('destroy' in source && typeof source.destroy === 'function') {
          source.destroy(error);
        }
        byteLimit.destroy(error);
        settle(error);
      }, REQUEST_IDLE_TIMEOUT_MS);
    };

    const byteLimit = new Transform({
      transform(chunk: Buffer | string, encoding, callback) {
        armTimeout();
        const size = Buffer.isBuffer(chunk) ? chunk.length : Buffer.byteLength(chunk, encoding);
        bytesWritten += size;
        if (bytesWritten > MAX_ARTIFACT_BYTES) {
          callback(
            new AppError(
              'INVALID_ARGS',
              `Upload exceeds maximum size of ${MAX_ARTIFACT_BYTES} bytes`,
            ),
          );
          return;
        }
        callback(null, chunk);
      },
    });

    source.on('aborted', () => {
      settle(new AppError('COMMAND_FAILED', 'Artifact transfer was interrupted'));