514 lines
19 KiB
TypeScript
514 lines
19 KiB
TypeScript
import { createHash, randomUUID } from 'node:crypto'
|
|
import { createReadStream } from 'node:fs'
|
|
import { lstat, mkdir, open, readFile, readdir, rename, rm, statfs } from 'node:fs/promises'
|
|
import { extname, join, resolve } from 'node:path'
|
|
import { setTimeout as wait } from 'node:timers/promises'
|
|
import { parseScreenPackage, type ScreenPackage } from './screen-package.js'
|
|
|
|
const digest = (value: string) => createHash('sha256').update(value).digest('hex')
|
|
const hex = /^[a-f0-9]{64}$/
|
|
const reserve = 64n * 1024n * 1024n
|
|
const mimeAliases: Record<string, string> = {
|
|
'audio/wave': 'audio/wav',
|
|
'audio/x-wav': 'audio/wav',
|
|
'application/ogg': 'audio/ogg',
|
|
'audio/mp3': 'audio/mpeg',
|
|
'image/jpg': 'image/jpeg',
|
|
}
|
|
const mimeTypes: Record<string, string> = {
|
|
'.mp3': 'audio/mpeg',
|
|
'.wav': 'audio/wav',
|
|
'.ogg': 'audio/ogg',
|
|
'.mp4': 'video/mp4',
|
|
'.webm': 'video/webm',
|
|
'.png': 'image/png',
|
|
'.jpg': 'image/jpeg',
|
|
'.jpeg': 'image/jpeg',
|
|
'.gif': 'image/gif',
|
|
'.webp': 'image/webp',
|
|
}
|
|
export interface StoredFile {
|
|
id: string
|
|
size: number
|
|
sha256: string
|
|
mime: string
|
|
}
|
|
export interface LocalPackage {
|
|
formatVersion: 1
|
|
key: string
|
|
version: string
|
|
package: ScreenPackage
|
|
files: StoredFile[]
|
|
}
|
|
export interface DownloadProgress {
|
|
stage: 'downloading' | 'verifying' | 'activating' | 'ready'
|
|
assetId: string | null
|
|
downloadedBytes: number
|
|
completedFiles: number
|
|
totalFiles: number
|
|
totalBytes: number | null
|
|
}
|
|
interface Dependencies {
|
|
fetch?: typeof fetch
|
|
freeBytes?: () => Promise<bigint>
|
|
wait?: (milliseconds: number, signal: AbortSignal) => Promise<void>
|
|
}
|
|
class RetryableError extends Error {}
|
|
export function packageKey(server: string, gameId: number): string {
|
|
return digest(JSON.stringify([server, gameId]))
|
|
}
|
|
function versionOf(pkg: ScreenPackage, files: StoredFile[]): string {
|
|
return digest(
|
|
JSON.stringify([
|
|
pkg.metadataVersion,
|
|
files.map(({ id, size, sha256, mime }) => ({ id, size, sha256, mime })),
|
|
]),
|
|
)
|
|
}
|
|
function checkHex(value: unknown): asserts value is string {
|
|
if (typeof value !== 'string' || !hex.test(value))
|
|
throw new Error('Некорректный идентификатор пакета или файла')
|
|
}
|
|
async function checksum(path: string) {
|
|
const hash = createHash('sha256')
|
|
let size = 0
|
|
for await (const chunk of createReadStream(path)) {
|
|
hash.update(chunk)
|
|
size += chunk.length
|
|
}
|
|
return { size, sha256: hash.digest('hex') }
|
|
}
|
|
|
|
export class PackageStore {
|
|
private readonly root: string
|
|
private busy = false
|
|
private cancellation: AbortController | null = null
|
|
private readonly ready = new Map<string, LocalPackage>()
|
|
constructor(
|
|
root: string,
|
|
private readonly dependencies: Dependencies = {},
|
|
) {
|
|
this.root = resolve(root)
|
|
}
|
|
|
|
// Все сегменты создаёт сервис; существующие симлинки никогда не обходятся.
|
|
private async path(...parts: string[]) {
|
|
let path = this.root
|
|
for (const part of ['', ...parts]) {
|
|
if (part) path = join(path, part)
|
|
try {
|
|
const info = await lstat(path)
|
|
if (info.isSymbolicLink()) throw new Error('Симлинк в хранилище пакетов')
|
|
} catch (error) {
|
|
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
|
|
}
|
|
}
|
|
return path
|
|
}
|
|
private async exclusive<T>(action: () => Promise<T>): Promise<T> {
|
|
if (this.busy) throw new Error('Хранилище пакетов занято')
|
|
this.busy = true
|
|
try {
|
|
return await action()
|
|
} finally {
|
|
this.busy = false
|
|
}
|
|
}
|
|
private async syncDirectory(path: string) {
|
|
// Windows не поддерживает fsync каталога; файлы синхронизируются на всех ОС.
|
|
if (process.platform === 'win32') return
|
|
const handle = await open(path, 'r')
|
|
try {
|
|
await handle.sync()
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
}
|
|
private async writeJSON(path: string, value: unknown) {
|
|
const handle = await open(path, 'wx', 0o600)
|
|
try {
|
|
await handle.writeFile(JSON.stringify(value))
|
|
await handle.sync()
|
|
} finally {
|
|
await handle.close()
|
|
}
|
|
}
|
|
private async cleanup(key: string, keep?: string) {
|
|
const dir = await this.path(key)
|
|
for (const name of await readdir(dir)) {
|
|
if (
|
|
(/^staging-[a-f0-9-]{36}$/.test(name) ||
|
|
/^v-[a-f0-9]{64}$/.test(name) ||
|
|
name === 'current.tmp') &&
|
|
name !== keep
|
|
) {
|
|
// rm удаляет сам симлинк, не его цель; содержимое по ссылке не читается.
|
|
await rm(join(dir, name), { recursive: true, force: true })
|
|
}
|
|
}
|
|
}
|
|
async initialize(): Promise<void> {
|
|
await this.exclusive(async () => {
|
|
await this.path()
|
|
await mkdir(this.root, { recursive: true })
|
|
await this.path()
|
|
this.ready.clear()
|
|
for (const key of await readdir(this.root)) {
|
|
if (!hex.test(key)) continue
|
|
try {
|
|
await this.path(key)
|
|
const manifest = await this.load(key)
|
|
this.ready.set(key, manifest)
|
|
await this.cleanup(key, `v-${manifest.version}`)
|
|
} catch {
|
|
// Повреждённый пакет остаётся для явного удаления, но не становится готовым.
|
|
try {
|
|
let keep: string | undefined
|
|
try {
|
|
const pointer: unknown = JSON.parse(
|
|
await readFile(await this.path(key, 'current.json'), 'utf8'),
|
|
)
|
|
checkHex(pointer)
|
|
keep = `v-${pointer}`
|
|
} catch {
|
|
/* Без корректного указателя готовой версии нет. */
|
|
}
|
|
await this.cleanup(key, keep)
|
|
} catch {
|
|
/* Чужие пути не обходим. */
|
|
}
|
|
}
|
|
}
|
|
})
|
|
}
|
|
private async load(key: string): Promise<LocalPackage> {
|
|
checkHex(key)
|
|
const pointer: unknown = JSON.parse(
|
|
await readFile(await this.path(key, 'current.json'), 'utf8'),
|
|
)
|
|
checkHex(pointer)
|
|
return this.verifyDirectory(key, pointer)
|
|
}
|
|
private async verifyDirectory(key: string, version: string): Promise<LocalPackage> {
|
|
const raw = JSON.parse(
|
|
await readFile(await this.path(key, `v-${version}`, 'manifest.json'), 'utf8'),
|
|
)
|
|
if (
|
|
raw?.formatVersion !== 1 ||
|
|
raw.key !== key ||
|
|
raw.version !== version ||
|
|
!Array.isArray(raw.files)
|
|
)
|
|
throw new Error('Повреждён манифест пакета')
|
|
const pkg = parseScreenPackage(raw.package)
|
|
if (packageKey(pkg.server, pkg.game.id) !== key || raw.files.length !== pkg.assets.length)
|
|
throw new Error('Неполный пакет')
|
|
const files: StoredFile[] = []
|
|
for (let i = 0; i < pkg.assets.length; i++) {
|
|
const file = raw.files[i] as StoredFile
|
|
if (
|
|
!file ||
|
|
file.id !== pkg.assets[i]!.id ||
|
|
!Number.isSafeInteger(file.size) ||
|
|
file.size <= 0 ||
|
|
!hex.test(file.sha256) ||
|
|
!Object.values(mimeTypes).includes(file.mime)
|
|
)
|
|
throw new Error('Повреждён список файлов')
|
|
const path = await this.path(key, `v-${version}`, 'assets', file.id)
|
|
if (!(await lstat(path)).isFile()) throw new Error('Ресурс не является файлом')
|
|
const actual = await checksum(path)
|
|
if (actual.size !== file.size || actual.sha256 !== file.sha256)
|
|
throw new Error('Повреждён файл пакета')
|
|
files.push({ id: file.id, ...actual, mime: file.mime })
|
|
}
|
|
if (versionOf(pkg, files) !== version) throw new Error('Повреждена версия пакета')
|
|
return { formatVersion: 1, key, version, package: pkg, files }
|
|
}
|
|
list(): LocalPackage[] {
|
|
return structuredClone([...this.ready.values()])
|
|
}
|
|
cancel(): void {
|
|
this.cancellation?.abort()
|
|
}
|
|
async verify(key: string): Promise<LocalPackage> {
|
|
return this.exclusive(async () => {
|
|
try {
|
|
const pkg = await this.load(key)
|
|
this.ready.set(key, pkg)
|
|
return structuredClone(pkg)
|
|
} catch (error) {
|
|
this.ready.delete(key)
|
|
throw error
|
|
}
|
|
})
|
|
}
|
|
async remove(key: string): Promise<void> {
|
|
checkHex(key)
|
|
await this.exclusive(async () => {
|
|
const dir = await this.path(key)
|
|
await rm(dir, { recursive: true, force: true })
|
|
this.ready.delete(key)
|
|
})
|
|
}
|
|
async localFile(key: string, assetId: string) {
|
|
checkHex(key)
|
|
checkHex(assetId)
|
|
const pkg = this.ready.get(key)
|
|
const file = pkg?.files.find((file) => file.id === assetId)
|
|
if (!pkg || !file) throw new Error('Готовый ресурс не найден')
|
|
try {
|
|
const path = await this.path(key, `v-${pkg.version}`, 'assets', assetId)
|
|
const info = await lstat(path)
|
|
if (!info.isFile() || info.size !== file.size) throw new Error('Повреждён файл пакета')
|
|
return { path, size: file.size, mime: file.mime }
|
|
} catch (error) {
|
|
this.ready.delete(key)
|
|
throw error
|
|
}
|
|
}
|
|
private async space(needed = 0) {
|
|
const available = this.dependencies.freeBytes
|
|
? await this.dependencies.freeBytes()
|
|
: await statfs(this.root, { bigint: true }).then((s) => s.bavail * s.bsize)
|
|
if (available < reserve + BigInt(needed))
|
|
throw new Error('Недостаточно места для пакета (нужен резерв 64 МиБ)')
|
|
}
|
|
|
|
async download(
|
|
value: ScreenPackage,
|
|
options: { signal?: AbortSignal; onProgress?: (progress: DownloadProgress) => void } = {},
|
|
): Promise<LocalPackage> {
|
|
return this.exclusive(async () => {
|
|
const pkg = parseScreenPackage(value)
|
|
const key = packageKey(pkg.server, pkg.game.id)
|
|
this.cancellation = new AbortController()
|
|
const signal = AbortSignal.any([
|
|
this.cancellation.signal,
|
|
...(options.signal ? [options.signal] : []),
|
|
])
|
|
const stage = `staging-${randomUUID()}`
|
|
let staged = ''
|
|
let published = ''
|
|
let activated = false
|
|
const files: StoredFile[] = []
|
|
let completedBytes = 0
|
|
const report = (stage: DownloadProgress['stage'], assetId: string | null, bytes = 0) => {
|
|
try {
|
|
options.onProgress?.({
|
|
stage,
|
|
assetId,
|
|
downloadedBytes: completedBytes + bytes,
|
|
completedFiles: files.length,
|
|
totalFiles: pkg.assets.length,
|
|
totalBytes: stage === 'ready' ? completedBytes : null,
|
|
})
|
|
} catch {
|
|
/* Наблюдатель не должен прерывать изменение хранилища. */
|
|
}
|
|
}
|
|
try {
|
|
signal.throwIfAborted()
|
|
await this.path()
|
|
await mkdir(this.root, { recursive: true })
|
|
const dir = await this.path(key)
|
|
await mkdir(dir, { recursive: true })
|
|
await this.space()
|
|
staged = await this.path(key, stage)
|
|
await mkdir(join(staged, 'assets'), { recursive: true })
|
|
for (const asset of pkg.assets) {
|
|
let file: StoredFile | undefined
|
|
for (let attempt = 0; attempt < 3; attempt++) {
|
|
signal.throwIfAborted()
|
|
const path = join(staged, 'assets', asset.id)
|
|
try {
|
|
file = await this.fetchFile(asset, path, signal, (bytes) =>
|
|
report('downloading', asset.id, bytes),
|
|
)
|
|
break
|
|
} catch (error) {
|
|
await rm(path, { force: true })
|
|
signal.throwIfAborted()
|
|
if (!(error instanceof RetryableError) || attempt === 2) throw error
|
|
const delay = attempt === 0 ? 1000 : 3000
|
|
if (this.dependencies.wait) await this.dependencies.wait(delay, signal)
|
|
else await wait(delay, undefined, { signal })
|
|
}
|
|
}
|
|
if (!file) throw new Error('Не удалось скачать файл')
|
|
report('verifying', asset.id, file.size)
|
|
const actual = await checksum(join(staged, 'assets', asset.id))
|
|
signal.throwIfAborted()
|
|
if (actual.size !== file.size || actual.sha256 !== file.sha256)
|
|
throw new Error('Ошибка проверки скачанного файла')
|
|
files.push(file)
|
|
completedBytes += file.size
|
|
}
|
|
const version = versionOf(pkg, files)
|
|
const manifest: LocalPackage = { formatVersion: 1, key, version, package: pkg, files }
|
|
await this.writeJSON(join(staged, 'manifest.json'), manifest)
|
|
await this.syncDirectory(join(staged, 'assets'))
|
|
await this.syncDirectory(staged)
|
|
report('activating', null)
|
|
signal.throwIfAborted()
|
|
const target = await this.path(key, `v-${version}`)
|
|
let exists = false
|
|
try {
|
|
await lstat(target)
|
|
exists = true
|
|
} catch (error) {
|
|
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
|
|
}
|
|
if (exists) {
|
|
try {
|
|
await this.verifyDirectory(key, version)
|
|
} catch {
|
|
// Повторная загрузка восстанавливает уже повреждённую локальную копию.
|
|
if (this.ready.get(key)?.version === version) this.ready.delete(key)
|
|
await rm(target, { recursive: true, force: true })
|
|
exists = false
|
|
}
|
|
}
|
|
if (!exists) {
|
|
await rename(staged, target)
|
|
staged = ''
|
|
published = target
|
|
}
|
|
await this.syncDirectory(dir)
|
|
const pointer = await this.path(key, 'current.tmp')
|
|
await rm(pointer, { force: true })
|
|
await this.writeJSON(pointer, version)
|
|
signal.throwIfAborted()
|
|
await rename(pointer, await this.path(key, 'current.json'))
|
|
activated = true
|
|
this.ready.set(key, manifest)
|
|
await this.syncDirectory(dir).catch(() => {})
|
|
// После переключения указателя сбой очистки не отменяет успешную активацию.
|
|
await this.cleanup(key, `v-${version}`).catch(() => {})
|
|
report('ready', null)
|
|
return structuredClone(manifest)
|
|
} catch (error) {
|
|
if (signal.aborted && !activated) throw new Error('Загрузка отменена')
|
|
if ((error as NodeJS.ErrnoException).code === 'ENOSPC')
|
|
throw new Error('Недостаточно места для пакета')
|
|
throw error
|
|
} finally {
|
|
this.cancellation = null
|
|
if (staged) await rm(staged, { recursive: true, force: true }).catch(() => {})
|
|
if (published && !activated)
|
|
await rm(published, { recursive: true, force: true }).catch(() => {})
|
|
if (!activated)
|
|
await rm(await this.path(key, 'current.tmp'), { force: true }).catch(() => {})
|
|
}
|
|
})
|
|
}
|
|
|
|
private async fetchFile(
|
|
asset: ScreenPackage['assets'][number],
|
|
path: string,
|
|
signal: AbortSignal,
|
|
progress: (bytes: number) => void,
|
|
): Promise<StoredFile> {
|
|
const idle = new AbortController()
|
|
const combined = AbortSignal.any([signal, idle.signal])
|
|
let timer: ReturnType<typeof setTimeout>
|
|
const reset = () => {
|
|
clearTimeout(timer)
|
|
timer = setTimeout(() => idle.abort(), 30_000)
|
|
}
|
|
reset()
|
|
const network = async <T>(operation: () => Promise<T>): Promise<T> => {
|
|
try {
|
|
return await operation()
|
|
} catch {
|
|
signal.throwIfAborted()
|
|
throw new RetryableError('Обрыв соединения или таймаут загрузки')
|
|
}
|
|
}
|
|
let response: Response | undefined
|
|
let reader: ReadableStreamDefaultReader<Uint8Array> | undefined
|
|
try {
|
|
let url = new URL(asset.url)
|
|
for (let redirects = 0; ; redirects++) {
|
|
if (
|
|
!['http:', 'https:'].includes(url.protocol) ||
|
|
url.username ||
|
|
url.password ||
|
|
url.hash ||
|
|
[...url.searchParams.keys()].some((key) =>
|
|
/token|signature|credential|password|authorization/i.test(key),
|
|
)
|
|
)
|
|
throw new Error('Недопустимый адрес медиа')
|
|
response = await network(() =>
|
|
(this.dependencies.fetch ?? fetch)(url.href, {
|
|
redirect: 'manual',
|
|
credentials: 'omit',
|
|
headers: { 'Accept-Encoding': 'identity' },
|
|
signal: combined,
|
|
}),
|
|
)
|
|
reset()
|
|
if (![301, 302, 303, 307, 308].includes(response.status)) break
|
|
await response.body?.cancel()
|
|
const location = response.headers.get('location')
|
|
if (!location || redirects === 5) throw new Error('Недопустимый редирект медиа')
|
|
const next = new URL(location, url)
|
|
if (url.protocol === 'https:' && next.protocol !== 'https:')
|
|
throw new Error('Небезопасный редирект медиа')
|
|
url = next
|
|
}
|
|
if ([408, 429].includes(response.status) || response.status >= 500)
|
|
throw new RetryableError(`Ошибка загрузки: HTTP ${response.status}`)
|
|
if (response.status !== 200 || !response.body)
|
|
throw new Error(`Медиа недоступно: HTTP ${response.status}`)
|
|
const encoding = response.headers.get('content-encoding')
|
|
if (encoding && encoding !== 'identity') throw new Error('Неожиданное сжатие медиа')
|
|
const header = response.headers.get('content-length')
|
|
const length = header === null ? null : Number(header)
|
|
if (length !== null && (!Number.isSafeInteger(length) || length <= 0))
|
|
throw new Error('Некорректный размер медиа')
|
|
const contentType = response.headers.get('content-type')?.split(';')[0]?.trim().toLowerCase()
|
|
const mime =
|
|
contentType && contentType !== 'application/octet-stream'
|
|
? (mimeAliases[contentType] ?? contentType)
|
|
: mimeTypes[extname(url.pathname).toLowerCase()]
|
|
if (!mime || !Object.values(mimeTypes).includes(mime))
|
|
throw new Error('Неподдерживаемый тип медиа')
|
|
await this.space(length ?? 0)
|
|
const file = await open(path, 'wx', 0o600)
|
|
let size = 0
|
|
const hash = createHash('sha256')
|
|
try {
|
|
reader = response.body.getReader()
|
|
for (;;) {
|
|
const chunk = await network(() => reader!.read())
|
|
signal.throwIfAborted()
|
|
if (chunk.done) break
|
|
reset()
|
|
if (
|
|
!Number.isSafeInteger(size + chunk.value.length) ||
|
|
(length !== null && size + chunk.value.length > length)
|
|
)
|
|
throw new RetryableError('Размер медиа не совпадает с ответом сервера')
|
|
await this.space(chunk.value.length)
|
|
await file.writeFile(chunk.value)
|
|
hash.update(chunk.value)
|
|
size += chunk.value.length
|
|
progress(size)
|
|
}
|
|
if (!size || (length !== null && size !== length))
|
|
throw new RetryableError('Файл загружен не полностью')
|
|
await file.sync()
|
|
} finally {
|
|
await file.close()
|
|
}
|
|
return { id: asset.id, size, sha256: hash.digest('hex'), mime }
|
|
} finally {
|
|
clearTimeout(timer!)
|
|
if (reader) await reader.cancel().catch(() => {})
|
|
else await response?.body?.cancel().catch(() => {})
|
|
}
|
|
}
|
|
}
|