feat(server): split file storage into a separate file service with tickets, receipts and WebDAV backup

- new projects WebDrop.FileApi (file service), WebDrop.FileProtocol (shared tickets and receipts) and WebDrop.Sync (WebDAV)
- file content of any size goes to the file service; chat, avatars and space structure stay on the main backend
- tickets issued by the main backend, receipts by the file service, deletion queue and daily reconciliation
- daily database snapshot and file sync to external storage (OpenList/WebDAV) with a 30 day retention
- reorganize Core, Storage and Api into feature folders (namespaces unchanged)
- web client uploads file frames to the file service and reads them with tickets
This commit is contained in:
justme committed 2026-10-17 15:25:16 +08:00
1 parent d66c93fe98
commit 72a5e5b839
84 files changed
+2156 -39

No files matched your search

+2
View File
@@ -47,6 +47,8 @@ export interface QueuedFrame {
bytes: Uint8Array;
extraHeaders: Record<string, string>;
createdAt: number;
/** 文件内容的帧:开启文件服务时字节传给文件服务,只把头部和回执交给主后端 */
remote?: boolean;
}
interface DraftFile {
+17 -1
View File
@@ -1,5 +1,5 @@
// Generated from backend DTOs; do not edit.
// backend-contract-sha256: 96cd0ffa3588a288af7bc720b2413cb56499dd862ac98d83c73ccc11a1147d25
// backend-contract-sha256: 1b3b8d56bf4262d61d7c2814fcd788b09225ccc94120084673ea46de46c09593
export interface AccountChannelBody {
channel_id: string;
@@ -168,6 +168,17 @@ export interface DomainPublishRequest {
export interface ErrorBody {
error: string;
}
export interface FileGrantBody {
expires_at: number;
mode: string;
ticket?: string | null;
url?: string | null;
}
export interface FileGrantRequest {
from: number;
item: string;
op: string;
}
export interface FinishShareRequest {
manifest_item_id?: string | null;
}
@@ -277,6 +288,7 @@ export interface ServerInfoBody {
choices: SettingChoicesBody;
contract_version: string;
defaults: DefaultsBody;
files: FilesInfoBody;
limits: LimitsBody;
protocol: number;
transports: TransportsBody;
@@ -291,6 +303,10 @@ export interface DefaultsBody {
channel_mode: ChannelMode;
share: ShareSettings;
}
export interface FilesInfoBody {
enabled: boolean;
url?: string | null;
}
export interface LimitsBody {
max_channel_bytes: number;
max_frames_per_item: number;
+69
View File
@@ -0,0 +1,69 @@
import { fromBase64Url, type ChannelSession } from '@webdrop/core';
import type { FileGrantBody } from '../generated/api';
import type { SavedResource } from '../database';
import { apiFetch, apiUpload, externalFetch, getServerInfo } from './http';
/**
* 文件的内容(不分大小)不放主后端,由文件服务(另一台机器)保存。流程:
* 上传:向主后端要上传票据 → 把整帧 PUT 给文件服务,得到回执 → 把 26 字节头部和回执交给主后端登记
* 下载:向主后端要下载票据(按原来的规则计次)→ 带着票据向文件服务取
* 聊天、头像、空间结构、文件目录不受影响,照旧走主后端。服务端没开文件服务时,下面的函数都退回原来的行为。
*/
const toHex = (bytes: Uint8Array): string => Array.from(bytes, (byte) => byte.toString(16).padStart(2, '0')).join('');
export const channelHex = (resourceId: string): string => toHex(fromBase64Url(resourceId, 16) ?? new Uint8Array(16));
export const itemHex = (itemId: string): string => toHex(fromBase64Url(itemId, 16) ?? new Uint8Array(16));
export const frameItemHex = (bytes: Uint8Array): string => toHex(bytes.subarray(2, 18));
export async function filesEnabled(): Promise<boolean> {
try { return (await getServerInfo()).files?.enabled === true; } catch { return false; }
}
const roleOf = (session: ChannelSession): 'owner' | 'write' | 'read' => (session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read');
async function requestGrant(resource: SavedResource, session: ChannelSession, item: string, op: 'put' | 'get', from: number): Promise<FileGrantBody> {
const response = await apiFetch(`/channels/${resource.id}/files/ticket`, {
method: 'POST', headers: { ...session.authHeaders(roleOf(session)), 'Content-Type': 'application/json' },
body: JSON.stringify({ item, op, from }),
});
return response.json() as Promise<FileGrantBody>;
}
const uploadTickets = new Map<string, FileGrantBody>();
/** 上传票据在有效期内复用(一个条目的所有帧共用),快过期时换新的 */
export async function uploadGrant(resource: SavedResource, session: ChannelSession, item: string, fresh = false): Promise<FileGrantBody> {
const key = `${resource.id}:${item}`;
const known = uploadTickets.get(key);
if (!fresh && known && known.expires_at - Date.now() > 60_000) return known;
const grant = await requestGrant(resource, session, item, 'put', 0);
if (grant.mode === 'remote') uploadTickets.set(key, grant); else uploadTickets.delete(key);
return grant;
}
export const forgetUploadGrant = (resource: SavedResource, item: string): void => { uploadTickets.delete(`${resource.id}:${item}`); };
/** 把一帧整个传给文件服务,返回文件服务签的回执 */
export async function putFrame(
grant: FileGrantBody, resource: SavedResource, bytes: Uint8Array, index: number,
onProgress: (loaded: number, total: number) => void, signal?: AbortSignal, stallMs?: number,
): Promise<string> {
const url = `${grant.url!.replace(/\/$/, '')}/f/${channelHex(resource.id)}/${frameItemHex(bytes)}/${index}`;
const text = await apiUpload('', { 'X-WD-Ticket': grant.ticket! }, new Blob([bytes as BlobPart]), onProgress, signal, stallMs, { url, method: 'PUT' });
const receipt = (JSON.parse(text) as { receipt?: string }).receipt;
if (!receipt) throw new Error('文件服务没有返回回执');
return receipt;
}
/**
* 打开一个条目自第 from 帧起的内容(下载、边下边播共用)。条目的帧在文件服务上就带着票据去那里取,
* 否则走主后端原来的接口(票据接口对本地条目回答 local,不计次,计次由原来的接口负责)。
*/
export async function openItemFrames(resource: SavedResource, session: ChannelSession, itemId: string, from: number, signal?: AbortSignal): Promise<Response> {
if (resource.kind === 'channel' && await filesEnabled()) {
const grant = await requestGrant(resource, session, itemId, 'get', from);
if (grant.mode === 'remote')
return externalFetch(`${grant.url!.replace(/\/$/, '')}/f/${channelHex(resource.id)}/${itemHex(itemId)}?from=${from}`, { 'X-WD-Ticket': grant.ticket! }, signal);
}
const path = resource.kind === 'share' ? `/shares/${resource.id}/items/${itemId}/frames?from=${from}` : `/channels/${resource.id}/items/${itemId}/frames?from=${from}`;
const role = resource.kind === 'share' && resource.state === 'draft' ? 'owner' : session.can('read') ? 'read' : 'owner';
return apiFetch(path, { headers: session.authHeaders(role), ...(signal ? { signal } : {}) });
}
+22 -3
View File
@@ -75,7 +75,9 @@ export async function apiUpload(
signal?: AbortSignal,
/** 这么久没有任何上传进展就放弃这次请求(弱网下连接静默卡死),由调用方重试 */
stallTimeoutMs?: number,
): Promise<void> {
/** 传给另一个服务(文件服务)时:完整地址和方法。这时不会带上账号令牌 */
external?: { url: string; method: string },
): Promise<string> {
try {
await getServerInfo();
const response = await new Promise<Response>((resolve, reject) => {
@@ -88,11 +90,11 @@ export async function apiUpload(
clearTimeout(stallTimer);
stallTimer = setTimeout(() => { stalled = true; xhr.abort(); }, stallTimeoutMs);
};
xhr.open('POST', apiPath(path));
xhr.open(external?.method ?? 'POST', external?.url ?? apiPath(path));
xhr.withCredentials = false;
xhr.responseType = 'text';
// 不要设 Cache-Control 请求头:它不在服务端 CORS 允许的头里,跨域部署(API 在另一个域名)时预检会失败;POST 本来也不会被缓存
const token = accountToken();
const token = external ? undefined : accountToken();
if (token) xhr.setRequestHeader('X-WD-Account', token);
for (const [name, value] of Object.entries(headers)) xhr.setRequestHeader(name, value);
xhr.upload.onprogress = (event) => {
@@ -125,6 +127,7 @@ export async function apiUpload(
if (Number.isFinite(wait) && wait > 0) error.retryAfter = wait;
throw error;
}
return await response.text();
} catch (error) {
if (error instanceof Error && 'status' in error) throw error;
if (error instanceof Error && error.message === 'api_contract_mismatch') throw error;
@@ -133,6 +136,22 @@ export async function apiUpload(
}
}
/** 直接向另一个服务(文件服务)取内容:只带它要的票据,不带账号令牌 */
export async function externalFetch(url: string, headers: Record<string, string>, signal?: AbortSignal): Promise<Response> {
try {
const response = await fetch(url, { headers, signal, cache: 'no-store', credentials: 'omit', referrerPolicy: 'no-referrer' });
if (!response.ok) {
const body = await response.json().catch(() => null) as { error?: unknown } | null;
throw fail(response.status, typeof body?.error === 'string' && /^[a-z_]{1,40}$/.test(body.error) ? body.error : 'request_failed');
}
return response;
} catch (error) {
if (signal?.aborted) throw new DOMException('请求已取消', 'AbortError');
if (error instanceof Error && 'status' in error) throw error;
throw new Error('网络连接失败');
}
}
export async function readJson<T>(path: string, headers: Record<string, string>): Promise<T> {
return apiFetch(path, { headers }).then((response) => response.json() as Promise<T>);
}
+2 -3
View File
@@ -1,5 +1,5 @@
import { fromBase64Url, IntegrityError } from '@webdrop/core';
import { apiFetch } from './http';
import { openItemFrames } from './fileService';
import { getSession } from './sessions';
/**
@@ -19,8 +19,7 @@ interface PlaybackRequest { resourceId: string; itemId: string; size: number; mi
async function framesFrom(resourceId: string, itemId: string, from: number): Promise<{ next: () => Promise<Uint8Array | undefined>; close: () => Promise<void> }> {
const { session, resource } = await getSession(resourceId);
if (resource.kind !== 'channel') throw new Error('资源类型不匹配');
const role = session.can('read') ? 'read' : 'owner';
const response = await apiFetch(`/channels/${resourceId}/items/${itemId}/frames?from=${from}`, { headers: session.authHeaders(role) });
const response = await openItemFrames(resource, session, itemId, from);
if (!response.body) throw new Error('服务端没有返回文件内容');
const expectedId = fromBase64Url(itemId, 16);
if (!expectedId) throw new IntegrityError('bad_meta');
+25 -9
View File
@@ -1,8 +1,9 @@
import { fromBase64Url, IntegrityError, MAX_FRAME_DATA, type ChannelSession, type ItemDecodeResult, type ItemPlan } from '@webdrop/core';
import { fromBase64Url, toBase64Url, IntegrityError, MAX_FRAME_DATA, type ChannelSession, type ItemDecodeResult, type ItemPlan } from '@webdrop/core';
import { dequeueFrame, queueFrame, nextQueuedFrame, lastQueuedAt, nextQueuedLocation, dequeueLocation, type QueuedFrame, type SavedResource } from '../database';
import { registerItemLocation } from './workspaceStructure';
import { verifyObjectEvent } from './channelDataKeys';
import { apiFetch, apiUpload, type ApiError } from './http';
import { apiUpload, type ApiError } from './http';
import { filesEnabled, forgetUploadGrant, openItemFrames, putFrame, uploadGrant } from './fileService';
import { signContentEvent } from './accountSigning';
import { accountToken } from './accountSession';
import { serial } from './serial';
@@ -40,6 +41,7 @@ export async function persistPlan(
bytes: frame.bytes,
extraHeaders: { ...(frame.index === 0 ? plan.frame0Headers : {}), ...frame.extraHeaders, ...extraHeaders },
createdAt: createdAt + frame.index,
...(resource.kind === 'channel' && signedEvent?.domain === 'files' && signedEvent.eventKind === 'file.publish' ? { remote: true } : {}),
};
// 提供了 onFrame 的调用方(上传流水线)自己决定何时落盘,见 frameStash
if (!onFrame) await queueFrame(queued);
@@ -96,15 +98,26 @@ export async function uploadFrame(
const role = resource.kind === 'share' ? 'owner' : writeRole(session);
const path = resource.kind === 'share' ? `/shares/${resource.id}/frames` : `/channels/${resource.id}/frames`;
let maxLoadedBytes = 0;
// 受限帧在开启文件服务时,字节传给文件服务,主后端只登记 26 字节头部和回执
let remote = resource.kind === 'channel' && frame.remote === true && await filesEnabled();
let receipt: string | undefined;
let refreshedOnce = false;
for (let attempt = 0; attempt < MAX_UPLOAD_ATTEMPTS; attempt += 1) {
try {
await apiUpload(
const onProgress = (loaded: number, total: number) => { maxLoadedBytes = Math.max(maxLoadedBytes, loaded); hooks?.onLoaded?.(maxLoadedBytes, total); };
const itemId = remote ? toBase64Url(frame.bytes.subarray(2, 18)) : '';
const grant = remote ? await uploadGrant(resource, session, itemId) : undefined;
if (grant && grant.mode === 'remote') {
receipt ??= await putFrame(grant, resource, frame.bytes, frame.index, onProgress, signal, FRAME_STALL_TIMEOUT_MS);
await apiUpload(path, { ...session.authHeaders(role), 'Content-Type': 'application/octet-stream', ...frame.extraHeaders, 'X-WD-Receipt': receipt },
new Blob([frame.bytes.subarray(0, 26) as BlobPart]), () => {}, signal, FRAME_STALL_TIMEOUT_MS);
} else await apiUpload(
path,
{ ...session.authHeaders(role), 'Content-Type': 'application/octet-stream', ...frame.extraHeaders },
// 用 Blob 发而不是字节数组:浏览器对字节数组请求体要多拷贝几次,实测上传只有 20–35 MB/s;Blob 走快速通道,单连接 100+ MB/s。
// 每次重试都从同一份 frame.bytes 重新包装,发出的字节完全相同(protocol §5)
new Blob([frame.bytes as BlobPart]),
(loaded, total) => { maxLoadedBytes = Math.max(maxLoadedBytes, loaded); hooks?.onLoaded?.(maxLoadedBytes, total); },
onProgress,
signal,
FRAME_STALL_TIMEOUT_MS,
);
@@ -114,6 +127,13 @@ export async function uploadFrame(
} catch (error) {
if (signal?.aborted || (error as Error | undefined)?.name === 'AbortError') throw error;
const apiError = error as ApiError;
// 老版本留在离线队列里的文件帧没有 remote 标记:服务端要求走文件服务,就改走文件服务重试
if (!remote && apiError.code === 'use_file_api' && resource.kind === 'channel') { remote = true; continue; }
// 票据或回执过期(401 来自文件服务,403 来自主后端验回执):换新的再试一次
if (remote && !refreshedOnce && (apiError.status === 401 || apiError.status === 403)) {
refreshedOnce = true; receipt = undefined; forgetUploadGrant(resource, toBase64Url(frame.bytes.subarray(2, 18)));
continue;
}
const retryable = apiError.status === undefined || apiError.status === 408 || apiError.status === 425
|| apiError.status === 429 || apiError.status === 499 || apiError.status >= 500;
if (!retryable) throw error;
@@ -177,11 +197,7 @@ export async function downloadDecrypted(
itemId: string,
expectedSize?: number,
): Promise<{ blob: Blob; meta: ItemDecodeResult['meta'] }> {
const role = resource.kind === 'share' && resource.state === 'draft' ? 'owner' : session.can('read') ? 'read' : 'owner';
const path = resource.kind === 'share'
? `/shares/${resource.id}/items/${itemId}/frames?from=0`
: `/channels/${resource.id}/items/${itemId}/frames?from=0`;
const response = await apiFetch(path, { headers: session.authHeaders(role) });
const response = await openItemFrames(resource, session, itemId, 0);
if (!response.body) throw new Error('服务端没有返回文件内容');
const expectedId = fromBase64Url(itemId, 16);
if (!expectedId) throw new IntegrityError('bad_meta');