253 lines
8.8 KiB
JavaScript
253 lines
8.8 KiB
JavaScript
import { createHash, randomUUID } from 'node:crypto';
|
|
import {
|
|
mkdirSync,
|
|
readFileSync,
|
|
writeFileSync,
|
|
renameSync,
|
|
rmSync,
|
|
readdirSync,
|
|
existsSync,
|
|
statSync,
|
|
} from 'node:fs';
|
|
import { open, stat, truncate } from 'node:fs/promises';
|
|
import { createReadStream } from 'node:fs';
|
|
import { resolve } from 'node:path';
|
|
import { eq } from 'drizzle-orm';
|
|
import * as tables from './db/schema.js';
|
|
import { fail } from './security.js';
|
|
|
|
export const CHUNK_BYTES = 8 * 1024 * 1024;
|
|
export class ResumableUploads {
|
|
constructor(service) {
|
|
this.service = service;
|
|
this.directory = resolve(service.cfg.uploadDir, 'chunks');
|
|
mkdirSync(this.directory, { recursive: true, mode: 0o700 });
|
|
this.busy = new Set();
|
|
}
|
|
path(id, extension) {
|
|
return resolve(this.directory, `${id}.${extension}`);
|
|
}
|
|
metadata(id) {
|
|
try {
|
|
return JSON.parse(readFileSync(this.path(id, 'json'), 'utf8'));
|
|
} catch (error) {
|
|
if (error.code === 'ENOENT') return null;
|
|
throw error;
|
|
}
|
|
}
|
|
save(id, value) {
|
|
writeFileSync(this.path(id, 'next'), JSON.stringify(value), { mode: 0o600, flush: true });
|
|
renameSync(this.path(id, 'next'), this.path(id, 'json'));
|
|
}
|
|
remove(id) {
|
|
for (const extension of ['part', 'json', 'next'])
|
|
rmSync(this.path(id, extension), { force: true });
|
|
}
|
|
manifest(subject, requestId, position) {
|
|
return this.service.storage.uploadManifest(subject, requestId, position);
|
|
}
|
|
file(subject, id) {
|
|
const row = this.service.db.select().from(tables.files).where(eq(tables.files.id, id)).get();
|
|
if (!row) fail(404, 'UPLOAD_NOT_FOUND', '업로드를 찾을 수 없습니다.');
|
|
return this.manifest(subject, row.requestId, row.position);
|
|
}
|
|
async lock(id, action) {
|
|
if (this.busy.has(id))
|
|
fail(409, 'UPLOAD_ACTIVE', '업로드 요청이 처리 중입니다. 잠시 후 다시 시도해 주세요.');
|
|
this.busy.add(id);
|
|
try {
|
|
return await action();
|
|
} finally {
|
|
this.busy.delete(id);
|
|
}
|
|
}
|
|
async state(subject, file, fingerprint) {
|
|
if (file.status === 'stored')
|
|
return {
|
|
uploadId: file.id,
|
|
offset: file.size,
|
|
size: file.size,
|
|
chunkSize: CHUNK_BYTES,
|
|
status: 'stored',
|
|
};
|
|
let meta = this.metadata(file.id);
|
|
if (!meta) {
|
|
if (!fingerprint) fail(404, 'UPLOAD_NOT_STARTED', '먼저 업로드를 시작해 주세요.');
|
|
meta = {
|
|
requestId: file.requestId,
|
|
subjectId: subject.id,
|
|
sessionId: subject.sessionId,
|
|
key: randomUUID(),
|
|
fingerprint,
|
|
hashes: [],
|
|
offset: 0,
|
|
updatedAt: Date.now(),
|
|
};
|
|
writeFileSync(this.path(file.id, 'part'), '', { flag: 'w', mode: 0o600 });
|
|
this.save(file.id, meta);
|
|
}
|
|
if (meta.subjectId !== subject.id || meta.sessionId !== subject.sessionId)
|
|
fail(409, 'UPLOAD_SESSION_CHANGED', '처음 업로드한 세션으로 다시 시도해 주세요.');
|
|
if (fingerprint && meta.fingerprint !== fingerprint)
|
|
fail(
|
|
409,
|
|
'FILE_CHANGED',
|
|
'처음 선택한 파일과 내용이 다릅니다. 같은 파일로 다시 시도해 주세요.',
|
|
);
|
|
const length = (await stat(this.path(file.id, 'part'))).size;
|
|
if (length < meta.offset) fail(409, 'UPLOAD_CORRUPTED', '업로드 자료를 다시 선택해 주세요.');
|
|
// Recover a write interrupted between appending bytes and committing metadata.
|
|
if (length > meta.offset) await truncate(this.path(file.id, 'part'), meta.offset);
|
|
this.manifest(subject, file.requestId, file.position);
|
|
this.service.storage.reserve(subject, file.requestId);
|
|
return {
|
|
uploadId: file.id,
|
|
offset: meta.offset,
|
|
size: file.size,
|
|
chunkSize: CHUNK_BYTES,
|
|
status: 'waiting',
|
|
};
|
|
}
|
|
async start(subject, requestId, position, fingerprint) {
|
|
const file = this.manifest(subject, requestId, position);
|
|
return this.lock(file.id, () => this.state(subject, file, fingerprint));
|
|
}
|
|
async status(subject, id) {
|
|
return this.lock(id, () => this.state(subject, this.file(subject, id)));
|
|
}
|
|
async append(subject, id, offset, bytes) {
|
|
return this.lock(id, async () => {
|
|
const file = this.file(subject, id);
|
|
if (file.status === 'stored') fail(409, 'FILE_STORED', '이미 저장된 파일입니다.');
|
|
const current = await this.state(subject, file),
|
|
meta = this.metadata(id);
|
|
if (
|
|
!Buffer.isBuffer(bytes) ||
|
|
offset % CHUNK_BYTES !== 0 ||
|
|
bytes.length !== Math.min(CHUNK_BYTES, file.size - offset) ||
|
|
!bytes.length
|
|
)
|
|
fail(413, 'CHUNK_SIZE_INVALID', '업로드 조각 크기가 올바르지 않습니다.');
|
|
const handle = await open(this.path(id, 'part'), 'r+');
|
|
try {
|
|
if (offset < current.offset && offset + bytes.length <= current.offset) {
|
|
const previous = Buffer.alloc(bytes.length);
|
|
const result = await handle.read(previous, 0, previous.length, offset);
|
|
if (result.bytesRead !== bytes.length || !previous.equals(bytes))
|
|
fail(409, 'CHUNK_CONFLICT', '이미 받은 조각과 내용이 다릅니다.');
|
|
return current;
|
|
}
|
|
if (offset !== current.offset)
|
|
fail(409, 'UPLOAD_OFFSET', '서버가 받은 위치부터 다시 보내 주세요.');
|
|
let written = 0;
|
|
while (written < bytes.length) {
|
|
const result = await handle.write(
|
|
bytes,
|
|
written,
|
|
bytes.length - written,
|
|
offset + written,
|
|
);
|
|
if (!result.bytesWritten) throw Error('Unable to write upload chunk');
|
|
written += result.bytesWritten;
|
|
}
|
|
await handle.sync();
|
|
this.file(subject, id);
|
|
if (!this.metadata(id)) fail(409, 'UPLOAD_CANCELLED', '전송이 취소되었습니다.');
|
|
meta.offset += bytes.length;
|
|
meta.hashes.push(createHash('sha256').update(bytes).digest('hex'));
|
|
meta.updatedAt = Date.now();
|
|
this.save(id, meta);
|
|
return { ...current, offset: meta.offset };
|
|
} finally {
|
|
await handle.close();
|
|
}
|
|
});
|
|
}
|
|
async complete(subject, id, request) {
|
|
return this.lock(id, async () => {
|
|
const file = this.file(subject, id);
|
|
if (file.status === 'stored') return { fileId: id, status: 'stored' };
|
|
const current = await this.state(subject, file),
|
|
meta = this.metadata(id);
|
|
if (current.offset !== file.size)
|
|
fail(409, 'UPLOAD_INCOMPLETE', '파일을 모두 보낸 후 완료할 수 있습니다.');
|
|
if (
|
|
createHash('sha256')
|
|
.update(Buffer.from(meta.hashes.join(''), 'hex'))
|
|
.digest('hex') !== meta.fingerprint
|
|
)
|
|
fail(409, 'FILE_CHECKSUM', '파일 내용 검증에 실패했습니다.');
|
|
const path = this.path(id, 'part');
|
|
const adapter = {
|
|
raw: request.raw,
|
|
async *parts() {
|
|
yield { type: 'file', filename: file.name, file: createReadStream(path) };
|
|
},
|
|
};
|
|
try {
|
|
const result = await this.service.storage.upload(
|
|
subject,
|
|
file.requestId,
|
|
file.position,
|
|
adapter,
|
|
meta.key,
|
|
path,
|
|
);
|
|
this.remove(id);
|
|
return result;
|
|
} catch (error) {
|
|
if (!existsSync(path)) this.remove(id);
|
|
throw error;
|
|
}
|
|
});
|
|
}
|
|
abort(requestId) {
|
|
for (const file of this.service.storage.fileRows(requestId)) this.remove(file.id);
|
|
}
|
|
abortSession(sessionId) {
|
|
for (const name of readdirSync(this.directory)) {
|
|
if (!name.endsWith('.json')) continue;
|
|
const id = name.slice(0, -5);
|
|
if (this.metadata(id)?.sessionId === sessionId) this.remove(id);
|
|
}
|
|
}
|
|
abortUser(userId) {
|
|
for (const name of readdirSync(this.directory)) {
|
|
if (name.endsWith('.json') && this.metadata(name.slice(0, -5))?.subjectId === userId)
|
|
this.remove(name.slice(0, -5));
|
|
}
|
|
}
|
|
cleanup() {
|
|
const now = Date.now();
|
|
for (const name of readdirSync(this.directory)) {
|
|
if (!name.endsWith('.json')) continue;
|
|
const id = name.slice(0, -5),
|
|
meta = this.metadata(id);
|
|
const row = this.service.db
|
|
.select()
|
|
.from(tables.requests)
|
|
.where(eq(tables.requests.id, meta.requestId))
|
|
.get();
|
|
if (
|
|
!this.busy.has(id) &&
|
|
(row?.status !== 'accepted' ||
|
|
!row.startBy ||
|
|
row.startBy <= now ||
|
|
meta.updatedAt < now - this.service.cfg.orphanHours * 3600000)
|
|
)
|
|
this.remove(id);
|
|
}
|
|
for (const name of readdirSync(this.directory)) {
|
|
if (!/^[0-9a-f-]{36}\.(part|next)$/.test(name)) continue;
|
|
const id = name.split('.')[0];
|
|
if (
|
|
!this.busy.has(id) &&
|
|
!existsSync(this.path(id, 'json')) &&
|
|
statSync(resolve(this.directory, name)).mtimeMs <
|
|
now - this.service.cfg.orphanHours * 3600000
|
|
)
|
|
this.remove(id);
|
|
}
|
|
}
|
|
}
|