Files
gongxue-base/apps/worker/src/storage.ts
2026-06-29 17:07:24 +08:00

247 lines
9.4 KiB
TypeScript

import fs from 'node:fs/promises';
import path from 'node:path';
import crypto from 'node:crypto';
import { StorageClient } from '@supabase/storage-js';
import { config } from './config.js';
export type StorageProviderName = 'external_url' | 'supabase_storage' | 'aliyun_oss' | 'tencent_cos' | 'qiniu_kodo' | 'local_dev';
interface PutObjectInput {
tenantId: string;
provider: StorageProviderName;
bucket: string;
objectKey: string;
body: Buffer;
mimeType: string;
}
interface PutObjectResult {
provider: StorageProviderName;
bucket: string;
objectKey: string;
checksumSha256: string;
sizeBytes: number;
mimeType: string;
localPath?: string;
etag?: string | null;
}
const SAFE_OBJECT_KEY_RE = /^[A-Za-z0-9][A-Za-z0-9._~!$&'()+,;=@/-]{0,1023}$/;
const SUPPORTED_UPLOAD_PROVIDERS = new Set<StorageProviderName>(['local_dev', 'supabase_storage', 'aliyun_oss', 'tencent_cos']);
export function normalizeStorageProvider(value: string | null | undefined): StorageProviderName {
const provider = (value?.trim() || config.storageDefaultProvider || 'local_dev') as StorageProviderName;
if (!SUPPORTED_UPLOAD_PROVIDERS.has(provider)) {
throw new Error(`Unsupported export storage provider: ${provider}`);
}
return provider;
}
export function storageBucket(value: string | null | undefined) {
return value?.trim() || config.storageDefaultBucket || 'tenant-assets';
}
export function canonicalObjectKey(objectKey: string) {
return objectKey.replace(/^\/+/, '').replace(/\/{2,}/g, '/');
}
export function validateObjectKey(tenantId: string, objectKey: string) {
const clean = canonicalObjectKey(objectKey);
if (!clean || clean.includes('..') || clean.includes('\\') || clean.includes('%2f') || clean.includes('%2F')) {
throw new Error('Invalid objectKey');
}
if (!SAFE_OBJECT_KEY_RE.test(clean)) {
throw new Error('objectKey contains unsafe characters');
}
if (config.storageRequireTenantPrefix && !clean.startsWith(`${tenantId}/`)) {
throw new Error('objectKey must be scoped by tenantId prefix');
}
return clean;
}
function sha256(body: Buffer) {
return crypto.createHash('sha256').update(body).digest('hex');
}
function requireConfigured(condition: unknown, provider: StorageProviderName, missing: string) {
if (!condition) throw new Error(`${provider} is not configured: ${missing}`);
}
async function putLocalDev(input: PutObjectInput, checksumSha256: string): Promise<PutObjectResult> {
const root = path.resolve(process.cwd(), config.exportLocalStorageRoot || '.local-storage');
const bucketRoot = path.resolve(root, input.bucket);
const target = path.resolve(bucketRoot, input.objectKey);
if (!target.startsWith(bucketRoot + path.sep)) {
throw new Error('Resolved local object path escapes storage root');
}
await fs.mkdir(path.dirname(target), { recursive: true });
await fs.writeFile(target, input.body);
return {
provider: input.provider,
bucket: input.bucket,
objectKey: input.objectKey,
checksumSha256,
sizeBytes: input.body.byteLength,
mimeType: input.mimeType,
localPath: target,
};
}
async function putAliyunOss(input: PutObjectInput, checksumSha256: string): Promise<PutObjectResult> {
requireConfigured(config.aliyunOssAccessKeyId, 'aliyun_oss', 'ALIYUN_OSS_ACCESS_KEY_ID');
requireConfigured(config.aliyunOssAccessKeySecret, 'aliyun_oss', 'ALIYUN_OSS_ACCESS_KEY_SECRET');
requireConfigured(config.aliyunOssRegion || config.aliyunOssEndpoint, 'aliyun_oss', 'ALIYUN_OSS_REGION or ALIYUN_OSS_ENDPOINT');
const { default: OSS } = await import('ali-oss');
const client = new OSS({
region: config.aliyunOssRegion || undefined,
endpoint: config.aliyunOssEndpoint || undefined,
accessKeyId: config.aliyunOssAccessKeyId,
accessKeySecret: config.aliyunOssAccessKeySecret,
stsToken: config.aliyunOssStsToken || undefined,
bucket: input.bucket,
internal: config.aliyunOssInternal,
secure: true,
});
const response = await (client as unknown as {
put: (objectKey: string, body: Buffer, options: Record<string, unknown>) => Promise<{ res?: { headers?: Record<string, string> } }>;
}).put(input.objectKey, input.body, {
headers: {
'Content-Type': input.mimeType,
'x-oss-meta-sha256': checksumSha256,
},
});
return {
provider: input.provider,
bucket: input.bucket,
objectKey: input.objectKey,
checksumSha256,
sizeBytes: input.body.byteLength,
mimeType: input.mimeType,
etag: response.res?.headers?.etag || null,
};
}
async function putSupabaseStorage(input: PutObjectInput, checksumSha256: string): Promise<PutObjectResult> {
requireConfigured(config.supabaseStorageUrl, 'supabase_storage', 'SUPABASE_STORAGE_URL');
requireConfigured(config.supabaseStorageServiceKey, 'supabase_storage', 'SUPABASE_STORAGE_SERVICE_KEY');
const client = new StorageClient(config.supabaseStorageUrl.replace(/\/+$/, ''), {
apikey: config.supabaseStorageServiceKey,
authorization: `Bearer ${config.supabaseStorageServiceKey}`,
});
const response = await client.from(input.bucket).upload(input.objectKey, input.body, {
contentType: input.mimeType,
upsert: true,
duplex: 'half',
metadata: { sha256: checksumSha256 },
} as never);
if (response.error) {
throw new Error(response.error.message || 'Supabase Storage upload failed');
}
return {
provider: input.provider,
bucket: input.bucket,
objectKey: input.objectKey,
checksumSha256,
sizeBytes: input.body.byteLength,
mimeType: input.mimeType,
};
}
function hmacSha1Hex(key: string | Buffer, value: string) {
return crypto.createHmac('sha1', key).update(value).digest('hex');
}
function sha1Hex(value: string) {
return crypto.createHash('sha1').update(value).digest('hex');
}
function cosEncodePath(objectKey: string) {
return objectKey
.split('/')
.map(part => encodeURIComponent(part).replace(/[!'()*]/g, char => `%${char.charCodeAt(0).toString(16).toUpperCase()}`))
.join('/');
}
function cosHost(bucket: string) {
requireConfigured(config.tencentCosRegion, 'tencent_cos', 'TENCENT_COS_REGION');
const bucketWithAppId = config.tencentCosAppId && !bucket.endsWith(`-${config.tencentCosAppId}`)
? `${bucket}-${config.tencentCosAppId}`
: bucket;
return `${bucketWithAppId}.cos.${config.tencentCosRegion}.myqcloud.com`;
}
function nowSeconds() {
return Math.floor(Date.now() / 1000);
}
async function putTencentCos(input: PutObjectInput, checksumSha256: string): Promise<PutObjectResult> {
requireConfigured(config.tencentCosSecretId, 'tencent_cos', 'TENCENT_COS_SECRET_ID');
requireConfigured(config.tencentCosSecretKey, 'tencent_cos', 'TENCENT_COS_SECRET_KEY');
const host = cosHost(input.bucket);
const start = nowSeconds();
const end = start + 300;
const keyTime = `${start};${end}`;
const pathname = `/${cosEncodePath(input.objectKey)}`;
const signedHeaders: Record<string, string> = {
host,
'content-type': input.mimeType,
'x-cos-meta-sha256': checksumSha256,
};
const headerKeys = Object.keys(signedHeaders).sort();
const headerList = headerKeys.join(';');
const httpHeaders = headerKeys
.map(key => `${encodeURIComponent(key)}=${encodeURIComponent(signedHeaders[key]).toLowerCase()}`)
.join('&');
const signedQuery: Record<string, string> = {};
if (config.tencentCosSecurityToken) signedQuery['x-cos-security-token'] = config.tencentCosSecurityToken;
const queryKeys = Object.keys(signedQuery).sort();
const urlParamList = queryKeys.join(';');
const httpParameters = queryKeys
.map(key => `${encodeURIComponent(key)}=${encodeURIComponent(signedQuery[key])}`)
.join('&');
const httpString = `put\n${pathname}\n${httpParameters}\n${httpHeaders}\n`;
const stringToSign = `sha1\n${keyTime}\n${sha1Hex(httpString)}\n`;
const signKey = hmacSha1Hex(config.tencentCosSecretKey, keyTime);
const signature = hmacSha1Hex(signKey, stringToSign);
const query = new URLSearchParams();
query.set('q-sign-algorithm', 'sha1');
query.set('q-ak', config.tencentCosSecretId);
query.set('q-sign-time', keyTime);
query.set('q-key-time', keyTime);
query.set('q-header-list', headerList);
query.set('q-url-param-list', urlParamList);
query.set('q-signature', signature);
for (const key of queryKeys) query.set(key, signedQuery[key]);
const response = await fetch(`https://${host}${pathname}?${query.toString()}`, {
method: 'PUT',
headers: signedHeaders,
body: input.body as unknown as BodyInit,
});
if (!response.ok) {
throw new Error(`Tencent COS upload failed: ${response.status}`);
}
return {
provider: input.provider,
bucket: input.bucket,
objectKey: input.objectKey,
checksumSha256,
sizeBytes: input.body.byteLength,
mimeType: input.mimeType,
etag: response.headers.get('etag'),
};
}
export async function putStorageObject(input: PutObjectInput): Promise<PutObjectResult> {
const provider = normalizeStorageProvider(input.provider);
const objectKey = validateObjectKey(input.tenantId, input.objectKey);
const normalized = { ...input, provider, objectKey };
const checksumSha256 = sha256(input.body);
if (provider === 'local_dev') return putLocalDev(normalized, checksumSha256);
if (provider === 'aliyun_oss') return putAliyunOss(normalized, checksumSha256);
if (provider === 'tencent_cos') return putTencentCos(normalized, checksumSha256);
if (provider === 'supabase_storage') return putSupabaseStorage(normalized, checksumSha256);
throw new Error(`${provider} does not support worker object upload`);
}