File size: 5,832 Bytes
68d7816 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | import { randomUUID } from 'node:crypto';
import { Readable } from 'node:stream';
import { LifecycleScope } from '#/app/scopes';
import { ScopeActivation, registerScopedService } from '#/_base/di/scope';
import { IBlobStore } from '#/persistence/interface/blobStore';
import {
IFileService,
fileNotFoundError,
isFileId,
type FileMeta,
type FileReadRange,
type GetResult,
type SaveOptions,
} from './fileService';
const BLOB_SCOPE = 'files';
const INDEX_SCOPE = 'file';
const INDEX_KEY = 'index.json';
const textEncoder = new TextEncoder();
const textDecoder = new TextDecoder();
interface IndexFile {
readonly version: 1;
readonly files: FileMeta[];
}
function isFileMeta(value: unknown): value is FileMeta {
if (typeof value !== 'object' || value === null || Array.isArray(value)) return false;
const meta = value as Record<string, unknown>;
return (
typeof meta['id'] === 'string' &&
isFileId(meta['id']) &&
typeof meta['name'] === 'string' &&
typeof meta['media_type'] === 'string' &&
typeof meta['size'] === 'number' &&
Number.isSafeInteger(meta['size']) &&
meta['size'] >= 0 &&
typeof meta['created_at'] === 'string' &&
(meta['expires_at'] === undefined || typeof meta['expires_at'] === 'string')
);
}
export class FileServiceImpl implements IFileService {
declare readonly _serviceBrand: undefined;
private indexCache: Map<string, FileMeta> | undefined;
private indexLoadPromise: Promise<void> | undefined;
private indexWritePromise: Promise<void> = Promise.resolve();
constructor(@IBlobStore private readonly blobs: IBlobStore) {}
async save(source: Readable, filename: string, options: SaveOptions = {}): Promise<FileMeta> {
await this.ensureIndex();
await this.pruneExpired();
const id = `f_${randomUUID()}`;
let size = 0;
const counting = async function* (): AsyncIterable<Uint8Array> {
for await (const chunk of source) {
const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as string);
size += buf.length;
yield buf;
}
};
try {
await this.blobs.putStream(BLOB_SCOPE, id, counting());
} catch (error) {
await this.blobs.delete(BLOB_SCOPE, id).catch(() => undefined);
throw error;
}
const now = Date.now();
const meta: FileMeta = {
id,
name: options.name ?? filename,
media_type: options.mimeType ?? 'application/octet-stream',
size,
created_at: new Date(now).toISOString(),
expires_at:
options.expiresInSec === undefined
? undefined
: new Date(now + options.expiresInSec * 1000).toISOString(),
};
this.indexCache!.set(id, meta);
await this.writeIndex();
return meta;
}
async get(fileId: string): Promise<GetResult> {
if (!isFileId(fileId)) {
throw fileNotFoundError(fileId);
}
await this.ensureIndex();
await this.pruneExpired();
const meta = this.indexCache!.get(fileId);
if (meta === undefined) {
throw fileNotFoundError(fileId);
}
const present = await this.blobs.has(BLOB_SCOPE, fileId);
if (!present) {
this.indexCache!.delete(fileId);
await this.writeIndex();
throw fileNotFoundError(fileId);
}
return {
meta,
stream: (range?: FileReadRange) =>
Readable.from(this.blobs.getStream(BLOB_SCOPE, fileId, range)),
};
}
async delete(fileId: string): Promise<void> {
if (!isFileId(fileId)) {
throw fileNotFoundError(fileId);
}
await this.ensureIndex();
if (!this.indexCache!.has(fileId)) {
throw fileNotFoundError(fileId);
}
this.indexCache!.delete(fileId);
await this.blobs.delete(BLOB_SCOPE, fileId);
await this.writeIndex();
}
private ensureIndex(): Promise<void> {
if (this.indexCache !== undefined) return Promise.resolve();
if (this.indexLoadPromise !== undefined) return this.indexLoadPromise;
this.indexLoadPromise = this.loadIndex().finally(() => {
this.indexLoadPromise = undefined;
});
return this.indexLoadPromise;
}
private async loadIndex(): Promise<void> {
const raw = await this.blobs.get(INDEX_SCOPE, INDEX_KEY);
if (raw === undefined) {
this.indexCache = new Map();
return;
}
try {
const parsed = JSON.parse(textDecoder.decode(raw)) as IndexFile;
const map = new Map<string, FileMeta>();
if (parsed && Array.isArray(parsed.files)) {
for (const f of parsed.files) {
if (isFileMeta(f)) {
map.set(f.id, f);
}
}
}
this.indexCache = map;
await this.pruneExpired();
} catch {
this.indexCache = new Map();
}
}
private async pruneExpired(): Promise<void> {
const cache = this.indexCache;
if (cache === undefined) return;
const now = Date.now();
const expired = [...cache.values()].filter(
(meta) => meta.expires_at !== undefined && Date.parse(meta.expires_at) <= now,
);
if (expired.length === 0) return;
for (const meta of expired) {
cache.delete(meta.id);
await this.blobs.delete(BLOB_SCOPE, meta.id).catch(() => undefined);
}
await this.writeIndex().catch(() => undefined);
}
private async writeIndex(): Promise<void> {
const write = this.indexWritePromise.catch(() => undefined).then(async () => {
const cache = this.indexCache;
if (cache === undefined) return;
const payload: IndexFile = { version: 1, files: Array.from(cache.values()) };
await this.blobs.put(INDEX_SCOPE, INDEX_KEY, textEncoder.encode(JSON.stringify(payload)));
});
this.indexWritePromise = write;
await write;
}
}
registerScopedService(
LifecycleScope.App,
IFileService,
FileServiceImpl,
ScopeActivation.OnScopeCreated,
'file',
);
|