25601018106

Committed 09 May 2026 12:22PM UTC coverage: 33.99% (-31.1%) from 65.107%

Build # 25601018106

Build Type

Pull #615

github

Committed by

web-flow

Commit Message

Merge 1ef509ec3 into 36e5af87e

Pull Request Pull Request #615: test: add unit tests for remaining app workers (#489)

Coverage Stats

788 of 3170 branches covered (24.86%)

Branch coverage included in aggregate %.

0 of 8 new or added lines in 2 files covered. (0.0%)

1822 existing lines in 87 files now uncovered.

2352 of 6068 relevant lines covered (38.76%)

13.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.26

/src/utils/compression.ts

import { createGunzip, createGzip } from 'zlib'
import { PassThrough, Transform } from 'stream'
import { cpus } from 'os'
import { extname } from 'path'
import { open } from 'fs/promises'

export enum CompressionFormat {
  GZIP = 'gzip',
  XZ = 'xz',
}

const GZIP_MAGIC = Buffer.from([0x1f, 0x8b])
const XZ_MAGIC = Buffer.from([0xfd, 0x37, 0x7a, 0x58, 0x5a, 0x00])

const DEFAULT_XZ_PRESET = 6
const DEFAULT_MAX_XZ_THREADS = 4
const MIN_XZ_PRESET = 0
const MAX_XZ_PRESET = 9

type LzmaNative = {
  createCompressor: (options?: Record<string, unknown>) => Transform
  createDecompressor: (options?: Record<string, unknown>) => Transform
}

type Environment = Record<string, string | undefined>

const getLzmaNative = (): LzmaNative => {
  try {
    return require('lzma-native') as LzmaNative
  } catch (error) {
    const reason = error instanceof Error ? error.message : String(error)

    throw new Error(`XZ support requires the "lzma-native" package. Install dependencies and try again. (${reason})`)
  }
}

const parseIntegerEnv = (
  key: string,
  env: Environment,
): number | undefined => {
  const rawValue = env[key]
  if (!rawValue || rawValue.trim() === '') {
    return undefined
  }

  if (!/^-?\d+$/.test(rawValue.trim())) {
    throw new Error(`Invalid ${key}: ${rawValue}. Expected an integer.`)
  }

  return Number(rawValue)
}

export const getXzCompressionOptions = (
  cpuCount: number,
  env: Environment = process.env,
): { preset: number; threads: number } => {
  const parsedPreset = parseIntegerEnv('NOSTREAM_XZ_PRESET', env)
  const preset = parsedPreset ?? DEFAULT_XZ_PRESET

  if (preset < MIN_XZ_PRESET || preset > MAX_XZ_PRESET) {
    throw new Error(
      `Invalid NOSTREAM_XZ_PRESET: ${preset}. Expected an integer between ${MIN_XZ_PRESET} and ${MAX_XZ_PRESET}.`,
    )
  }

  const parsedThreadCap = parseIntegerEnv('NOSTREAM_XZ_THREADS', env)
  if (parsedThreadCap !== undefined && parsedThreadCap <= 0) {
    throw new Error('Invalid NOSTREAM_XZ_THREADS: expected a positive integer.')
  }

  // Keep one core available by default to reduce contention with the running relay.
  const availableThreads = Math.max(1, Math.max(1, Math.trunc(cpuCount)) - 1)
  const maxThreads = parsedThreadCap ?? DEFAULT_MAX_XZ_THREADS

  return {
    preset,
    threads: Math.max(1, Math.min(availableThreads, maxThreads)),
  }
}

export const parseCompressionFormat = (input: string): CompressionFormat => {
  switch (input.trim().toLowerCase()) {
    case 'gzip':
    case 'gz':
      return CompressionFormat.GZIP
    case 'xz':
      return CompressionFormat.XZ
    default:
      throw new Error(`Unsupported compression format: ${input}. Use gzip|gz|xz.`)
  }
}

export const getCompressionFormatFromExtension = (
  filePath: string,
): CompressionFormat | undefined => {
  switch (extname(filePath).toLowerCase()) {
    case '.gz':
      return CompressionFormat.GZIP
    case '.xz':
      return CompressionFormat.XZ
    default:
      return undefined
  }
}

export const getCompressionFormatFromHeader = (
  header: Buffer,
): CompressionFormat | undefined => {
  if (header.length >= GZIP_MAGIC.length && header.subarray(0, GZIP_MAGIC.length).equals(GZIP_MAGIC)) {
    return CompressionFormat.GZIP
  }

  if (header.length >= XZ_MAGIC.length && header.subarray(0, XZ_MAGIC.length).equals(XZ_MAGIC)) {
    return CompressionFormat.XZ
  }

  return undefined
}

const readFileHeader = async (filePath: string, bytes = XZ_MAGIC.length): Promise<Buffer> => {
  const fileHandle = await open(filePath, 'r')

  try {
    const header = Buffer.alloc(bytes)
    const { bytesRead } = await fileHandle.read(header, 0, bytes, 0)

    return header.subarray(0, bytesRead)
  } finally {
    await fileHandle.close()
  }
}

export const detectCompressionFormat = async (
  filePath: string,
): Promise<CompressionFormat | undefined> => {
  const extensionFormat = getCompressionFormatFromExtension(filePath)
  const header = await readFileHeader(filePath)
  const headerFormat = getCompressionFormatFromHeader(header)

  if (extensionFormat && headerFormat && extensionFormat !== headerFormat) {
    throw new Error(
      `Compression mismatch for ${filePath}: extension suggests ${extensionFormat} but header is ${headerFormat}.`,
    )
  }

  return headerFormat ?? extensionFormat
}

export const createCompressionStream = (
  format?: CompressionFormat,
): Transform => {
  if (!format) {
    return new PassThrough()
  }

  switch (format) {
    case CompressionFormat.GZIP:
      return createGzip()
    case CompressionFormat.XZ: {
      const lzmaNative = getLzmaNative()
      const { preset, threads } = getXzCompressionOptions(cpus().length)

      return lzmaNative.createCompressor({
        preset,
        threads,
      })
    }
    default:
      throw new Error(`Unsupported compression format: ${String(format)}`)
  }
}

export const createDecompressionStream = (
  format?: CompressionFormat,
): Transform => {
  if (!format) {
    return new PassThrough()
  }

  switch (format) {
    case CompressionFormat.GZIP:
      return createGunzip()
    case CompressionFormat.XZ:
      return getLzmaNative().createDecompressor()
    default:
      throw new Error(`Unsupported compression format: ${String(format)}`)
  }
}

1	import { createGunzip, createGzip } from 'zlib'	4✔
2	import { PassThrough, Transform } from 'stream'	4✔
3	import { cpus } from 'os'	4✔
4	import { extname } from 'path'	4✔
5	import { open } from 'fs/promises'	4✔
6
7	export enum CompressionFormat {	4✔
8	GZIP = 'gzip',	4✔
9	XZ = 'xz',	4✔
10	}
11
12	const GZIP_MAGIC = Buffer.from([0x1f, 0x8b])	4✔
13	const XZ_MAGIC = Buffer.from([0xfd, 0x37, 0x7a, 0x58, 0x5a, 0x00])	4✔
14
15	const DEFAULT_XZ_PRESET = 6	4✔
16	const DEFAULT_MAX_XZ_THREADS = 4	4✔
17	const MIN_XZ_PRESET = 0	4✔
18	const MAX_XZ_PRESET = 9	4✔
19
20	type LzmaNative = {
21	createCompressor: (options?: Record<string, unknown>) => Transform
22	createDecompressor: (options?: Record<string, unknown>) => Transform
23	}
24
25	type Environment = Record<string, string \| undefined>
26
27	const getLzmaNative = (): LzmaNative => {	4✔
28	try {	2✔
29	return require('lzma-native') as LzmaNative	2✔
30	} catch (error) {
31	const reason = error instanceof Error ? error.message : String(error)	×
32
33	throw new Error(`XZ support requires the "lzma-native" package. Install dependencies and try again. (${reason})`)	×
34	}
35	}
36
37	const parseIntegerEnv = (	4✔
38	key: string,
39	env: Environment,
40	): number \| undefined => {
41	const rawValue = env[key]	2✔
42	if (!rawValue \|\| rawValue.trim() === '') {	2!
43	return undefined	2✔
44	}
45
UNCOV 46	if (!/^-?\d+$/.test(rawValue.trim())) {	×
UNCOV 47	throw new Error(`Invalid ${key}: ${rawValue}. Expected an integer.`)	×
48	}
49
UNCOV 50	return Number(rawValue)	×
51	}
52
53	export const getXzCompressionOptions = (	4✔
54	cpuCount: number,
55	env: Environment = process.env,	1✔
56	): { preset: number; threads: number } => {
57	const parsedPreset = parseIntegerEnv('NOSTREAM_XZ_PRESET', env)	1✔
58	const preset = parsedPreset ?? DEFAULT_XZ_PRESET	1✔
59
60	if (preset < MIN_XZ_PRESET \|\| preset > MAX_XZ_PRESET) {	1!
UNCOV 61	throw new Error(	×
62	`Invalid NOSTREAM_XZ_PRESET: ${preset}. Expected an integer between ${MIN_XZ_PRESET} and ${MAX_XZ_PRESET}.`,
63	)
64	}
65
66	const parsedThreadCap = parseIntegerEnv('NOSTREAM_XZ_THREADS', env)	1✔
67	if (parsedThreadCap !== undefined && parsedThreadCap <= 0) {	1!
UNCOV 68	throw new Error('Invalid NOSTREAM_XZ_THREADS: expected a positive integer.')	×
69	}
70
71	// Keep one core available by default to reduce contention with the running relay.
72	const availableThreads = Math.max(1, Math.max(1, Math.trunc(cpuCount)) - 1)	1✔
73	const maxThreads = parsedThreadCap ?? DEFAULT_MAX_XZ_THREADS	1✔
74
75	return {	1✔
76	preset,
77	threads: Math.max(1, Math.min(availableThreads, maxThreads)),
78	}
79	}
80
81	export const parseCompressionFormat = (input: string): CompressionFormat => {	4✔
82	switch (input.trim().toLowerCase()) {	2!
83	case 'gzip':
84	case 'gz':
85	return CompressionFormat.GZIP	1✔
86	case 'xz':
87	return CompressionFormat.XZ	1✔
88	default:
UNCOV 89	throw new Error(`Unsupported compression format: ${input}. Use gzip\|gz\|xz.`)	×
90	}
91	}
92
93	export const getCompressionFormatFromExtension = (	4✔
94	filePath: string,
95	): CompressionFormat \| undefined => {
96	switch (extname(filePath).toLowerCase()) {	2!
97	case '.gz':
98	return CompressionFormat.GZIP	1✔
99	case '.xz':
100	return CompressionFormat.XZ	1✔
101	default:
UNCOV 102	return undefined	×
103	}
104	}
105
106	export const getCompressionFormatFromHeader = (	4✔
107	header: Buffer,
108	): CompressionFormat \| undefined => {
109	if (header.length >= GZIP_MAGIC.length && header.subarray(0, GZIP_MAGIC.length).equals(GZIP_MAGIC)) {	2✔
110	return CompressionFormat.GZIP	1✔
111	}
112
113	if (header.length >= XZ_MAGIC.length && header.subarray(0, XZ_MAGIC.length).equals(XZ_MAGIC)) {	1!
114	return CompressionFormat.XZ	1✔
115	}
116
UNCOV 117	return undefined	×
118	}
119
120	const readFileHeader = async (filePath: string, bytes = XZ_MAGIC.length): Promise<Buffer> => {	4✔
121	const fileHandle = await open(filePath, 'r')	2✔
122
123	try {	2✔
124	const header = Buffer.alloc(bytes)	2✔
125	const { bytesRead } = await fileHandle.read(header, 0, bytes, 0)	2✔
126
127	return header.subarray(0, bytesRead)	2✔
128	} finally {
129	await fileHandle.close()	2✔
130	}
131	}
132
133	export const detectCompressionFormat = async (	4✔
134	filePath: string,
135	): Promise<CompressionFormat \| undefined> => {
136	const extensionFormat = getCompressionFormatFromExtension(filePath)	2✔
137	const header = await readFileHeader(filePath)	2✔
138	const headerFormat = getCompressionFormatFromHeader(header)	2✔
139
140	if (extensionFormat && headerFormat && extensionFormat !== headerFormat) {	2!
UNCOV 141	throw new Error(	×
142	`Compression mismatch for ${filePath}: extension suggests ${extensionFormat} but header is ${headerFormat}.`,
143	)
144	}
145
146	return headerFormat ?? extensionFormat	2!
147	}
148
149	export const createCompressionStream = (	4✔
150	format?: CompressionFormat,
151	): Transform => {
152	if (!format) {	2!
UNCOV 153	return new PassThrough()	×
154	}
155
156	switch (format) {	2!
157	case CompressionFormat.GZIP:
158	return createGzip()	1✔
159	case CompressionFormat.XZ: {
160	const lzmaNative = getLzmaNative()	1✔
161	const { preset, threads } = getXzCompressionOptions(cpus().length)	1✔
162
163	return lzmaNative.createCompressor({	1✔
164	preset,
165	threads,
166	})
167	}
168	default:
169	throw new Error(`Unsupported compression format: ${String(format)}`)	×
170	}
171	}
172
173	export const createDecompressionStream = (	4✔
174	format?: CompressionFormat,
175	): Transform => {
176	if (!format) {	2!
UNCOV 177	return new PassThrough()	×
178	}
179
180	switch (format) {	2!
181	case CompressionFormat.GZIP:
182	return createGunzip()	1✔
183	case CompressionFormat.XZ:
184	return getLzmaNative().createDecompressor()	1✔
185	default:
186	throw new Error(`Unsupported compression format: ${String(format)}`)	×
187	}
188	}

cameri / nostream / 25601018106

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous