up
This commit is contained in:
@@ -0,0 +1,513 @@
|
||||
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(() => {})
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user