From a5745839436d961345e89a501b67dc391f6e6d43 Mon Sep 17 00:00:00 2001 From: JustMe Date: Fri, 16 Oct 2026 19:42:33 +0800 Subject: [PATCH] feat(web): space roles, upload queue with pipelined frames, background resume and faster sending --- web/src/core-bridge/api.ts | 7 + web/src/core-bridge/database.ts | 31 ++- web/src/core-bridge/real.ts | 5 + web/src/core-bridge/types.ts | 5 +- web/src/core-bridge/worker.ts | 9 +- web/src/core-bridge/worker/account.ts | 4 +- web/src/core-bridge/worker/channelDataKeys.ts | 4 + web/src/core-bridge/worker/channelSend.ts | 101 +-------- web/src/core-bridge/worker/channelThreads.ts | 12 +- .../core-bridge/worker/channelTransport.ts | 18 +- web/src/core-bridge/worker/frameStash.test.ts | 74 +++++++ web/src/core-bridge/worker/frameStash.ts | 85 +++++++ web/src/core-bridge/worker/http.ts | 34 ++- .../worker/resourceMetadata.test.ts | 2 +- web/src/core-bridge/worker/resumeUploads.ts | 87 ++++++++ web/src/core-bridge/worker/transfer.ts | 128 +++++++---- .../core-bridge/worker/uploadPipeline.test.ts | 139 ++++++++++++ web/src/core-bridge/worker/uploadPipeline.ts | 156 +++++++++++++ web/src/core-bridge/worker/uploadQueue.ts | 192 ++++++++++++++++ .../core-bridge/worker/workspaceStructure.ts | 11 +- .../channel/ChannelSettingsSheet.module.css | 5 + .../features/channel/ChannelSettingsSheet.tsx | 11 +- web/src/features/channel/ChatMessage.tsx | 9 +- web/src/features/channel/MembersPane.tsx | 21 +- .../features/channel/RoleBadges.module.css | 42 ++++ web/src/features/channel/RoleBadges.tsx | 20 ++ .../features/channel/RolesEditor.module.css | 207 ++++++++++++++++++ web/src/features/channel/RolesEditor.tsx | 106 +++++++++ web/src/features/channel/ThreadSidebar.tsx | 7 +- web/src/features/channel/Workspace.tsx | 9 +- .../features/item/TransferProgress.module.css | 45 ++++ web/src/features/item/TransferProgress.tsx | 22 ++ web/src/lib/format.ts | 29 ++- web/src/lib/roles.ts | 35 +++ web/src/store/reduce.test.ts | 2 +- web/src/store/reduce.ts | 5 +- web/src/store/workspace.ts | 23 +- web/src/styles/tokens.css | 3 + web/vite.config.ts | 4 +- 39 files changed, 1515 insertions(+), 194 deletions(-) create mode 100644 web/src/core-bridge/worker/frameStash.test.ts create mode 100644 web/src/core-bridge/worker/frameStash.ts create mode 100644 web/src/core-bridge/worker/resumeUploads.ts create mode 100644 web/src/core-bridge/worker/uploadPipeline.test.ts create mode 100644 web/src/core-bridge/worker/uploadPipeline.ts create mode 100644 web/src/core-bridge/worker/uploadQueue.ts create mode 100644 web/src/features/channel/RoleBadges.module.css create mode 100644 web/src/features/channel/RoleBadges.tsx create mode 100644 web/src/features/channel/RolesEditor.module.css create mode 100644 web/src/features/channel/RolesEditor.tsx create mode 100644 web/src/features/item/TransferProgress.module.css create mode 100644 web/src/features/item/TransferProgress.tsx create mode 100644 web/src/lib/roles.ts diff --git a/web/src/core-bridge/api.ts b/web/src/core-bridge/api.ts index 43292b1..4c8943b 100644 --- a/web/src/core-bridge/api.ts +++ b/web/src/core-bridge/api.ts @@ -1,5 +1,6 @@ import type { ParsedChannelLink } from '@/lib/channelLink'; import type { AccountInvitation } from './account'; +import type { WorkspaceRoleRequest } from './generated/api'; import type { ArchiveEntry, ChannelInfo, ChannelSettingsPatch, CreateOptions, FileUploadOptions, WorkspaceEvent } from './types'; /** @@ -24,10 +25,16 @@ export interface WorkspaceApi { connect(channelId: string, onEvent: (e: WorkspaceEvent) => void): () => void; sendText(text: string, threadId?: string, localId?: string): Promise; sendFiles(files: File[], options?: FileUploadOptions, threadId?: string, folderId?: string, localIds?: string[]): Promise; + /** 取消排队中或上传中的文件;返回是否找到这个任务 */ + cancelUpload(localId: string): Promise; createThread(name: string, groupId?: string): Promise; createGroup(name: string): Promise; renameGroup(id: string, name: string): Promise; deleteGroup(id: string): Promise; + createRole(role: WorkspaceRoleRequest): Promise; + updateRole(id: string, role: WorkspaceRoleRequest): Promise; + deleteRole(id: string): Promise; + setRoleMembers(id: string, members: string[]): Promise; groupThread(id: string, groupId?: string): Promise; renameThread(threadId: string, name: string): Promise; setChannelHistory(threadId: string, chatHistory: boolean, filesHistory: boolean): Promise; diff --git a/web/src/core-bridge/database.ts b/web/src/core-bridge/database.ts index 81c1998..0c547ca 100644 --- a/web/src/core-bridge/database.ts +++ b/web/src/core-bridge/database.ts @@ -201,8 +201,11 @@ export async function clearDraftFiles(resourceId: string): Promise { await tx.done; } +/** 离线队列只是崩溃/断网后的续传保险,不必每帧都等磁盘落盘:relaxed 让浏览器先确认再慢慢写,单帧写入快数倍 */ export async function queueFrame(frame: QueuedFrame): Promise { - await (await db()).put('outbox', frame); + const tx = (await db()).transaction('outbox', 'readwrite', { durability: 'relaxed' }); + void tx.store.put(frame); + await tx.done; } export async function queuedFrames(resourceId: string): Promise { @@ -210,14 +213,29 @@ export async function queuedFrames(resourceId: string): Promise { return frames.sort((a, b) => a.createdAt - b.createdAt || a.itemId.localeCompare(b.itemId) || a.index - b.index); } +/** 只读键不读内容:续传前先看有哪些条目的哪些帧还没发,不把整个离线队列(可能几个 GB)读进内存 */ +export async function queuedFrameKeys(resourceId: string): Promise { + return (await db()).getAllKeysFromIndex('outbox', 'by-resource', resourceId); +} + +export async function queuedFrame(key: string): Promise { + return (await db()).get('outbox', key); +} + export async function dequeueFrame(key: string): Promise { await (await db()).delete('outbox', key); } -export async function nextQueuedFrame(resourceId: string): Promise { +/** 离线队列里最早的一帧;accept 用来跳过正由上传流水线处理的条目,或只取指定条目 */ +export async function nextQueuedFrame(resourceId: string, accept?: (itemId: string) => boolean): Promise { const range = IDBKeyRange.bound([resourceId, 0], [resourceId, Number.MAX_SAFE_INTEGER]); - const values = await (await db()).getAllFromIndex('outbox', 'by-resource-created', range, 1); - return values[0]; + if (!accept) return (await (await db()).getAllFromIndex('outbox', 'by-resource-created', range, 1))[0]; + let cursor = await (await db()).transaction('outbox').store.index('by-resource-created').openCursor(range); + while (cursor) { + if (accept(cursor.value.itemId)) return cursor.value; + cursor = await cursor.continue(); + } + return undefined; } export async function lastQueuedAt(resourceId: string): Promise { @@ -239,7 +257,8 @@ export interface QueuedLocation { key: string; resourceId: string; itemId: strin export async function queueLocation(resourceId: string, itemId: string, threadId: string, folderId?: string): Promise { await (await db()).put('locationOutbox', { key: `${resourceId}:${itemId}`, resourceId, itemId, threadId, folderId }); } -export async function nextQueuedLocation(resourceId: string): Promise { - return (await (await db()).getAllFromIndex('locationOutbox', 'by-resource', resourceId, 1))[0]; +export async function nextQueuedLocation(resourceId: string, accept?: (itemId: string) => boolean): Promise { + if (!accept) return (await (await db()).getAllFromIndex('locationOutbox', 'by-resource', resourceId, 1))[0]; + return (await (await db()).getAllFromIndex('locationOutbox', 'by-resource', resourceId)).find((row) => accept(row.itemId)); } export async function dequeueLocation(key: string): Promise { await (await db()).delete('locationOutbox', key); } diff --git a/web/src/core-bridge/real.ts b/web/src/core-bridge/real.ts index 92ea154..c822022 100644 --- a/web/src/core-bridge/real.ts +++ b/web/src/core-bridge/real.ts @@ -37,8 +37,13 @@ export const realApi: WorkspaceApi = { sendText: (text, threadId, localId) => callWorker('sendText', { id: currentChannelId(), text, threadId, localId, apiBase: API_BASE }), sendFiles: (files, options, threadId, folderId, localIds) => callWorker('sendFiles', { id: currentChannelId(), files, options, threadId, folderId, localIds, apiBase: API_BASE }), + cancelUpload: (localId) => callWorker('cancelUpload', { id: currentChannelId(), localId, apiBase: API_BASE }), createThread: (name, groupId) => callWorker('createThread', { id: currentChannelId(), name, groupId, apiBase: API_BASE }), createGroup: (name) => callWorker('createGroup', { id: currentChannelId(), name, apiBase: API_BASE }), + createRole: (role) => callWorker('createRole', { id: currentChannelId(), role, apiBase: API_BASE }), + updateRole: (roleId, role) => callWorker('updateRole', { id: currentChannelId(), roleId, role, apiBase: API_BASE }), + deleteRole: (roleId) => callWorker('deleteRole', { id: currentChannelId(), roleId, apiBase: API_BASE }), + setRoleMembers: (roleId, members) => callWorker('setRoleMembers', { id: currentChannelId(), roleId, members, apiBase: API_BASE }), deleteGroup: (groupId) => callWorker('deleteGroup', { id: currentChannelId(), groupId, apiBase: API_BASE }), renameGroup: (groupId, name) => callWorker('renameGroup', { id: currentChannelId(), groupId, name, apiBase: API_BASE }), groupThread: (threadId, groupId) => callWorker('groupThread', { id: currentChannelId(), threadId, groupId, apiBase: API_BASE }), diff --git a/web/src/core-bridge/types.ts b/web/src/core-bridge/types.ts index 0ab1844..b77ec93 100644 --- a/web/src/core-bridge/types.ts +++ b/web/src/core-bridge/types.ts @@ -1,4 +1,4 @@ -import type { ChannelMode, WorkspaceChannel, WorkspaceGroup, WorkspaceFolder, WorkspaceState, AccountProfile } from './generated/api'; +import type { ChannelMode, WorkspaceChannel, WorkspaceGroup, WorkspaceRole, WorkspaceFolder, WorkspaceState, AccountProfile } from './generated/api'; export type ConnectionState = 'online' | 'reconnecting' | 'offline'; export type PreviewKind = 'image' | 'video' | 'audio' | 'text' | 'archive' | 'none'; @@ -88,7 +88,7 @@ export interface ShareSettings { /** protocol §7、§14:由帧推导,不单独存储 */ export type ItemState = - | { kind: 'receiving'; progress: number; phase?: 'encrypting' | 'uploading'; speedBytesPerSecond?: number } + | { kind: 'receiving'; progress: number; phase?: 'queued' | 'encrypting' | 'uploading'; speedBytesPerSecond?: number; etaSeconds?: number; sentBytes?: number; totalBytes?: number } | { kind: 'complete' } | { kind: 'incomplete' } | { kind: 'tampered' }; @@ -159,6 +159,7 @@ export interface Snapshot { export type ThreadInfo = Omit & { groupId?: string }; export type ChannelGroupInfo = WorkspaceGroup; +export type RoleInfo = WorkspaceRole; export type FolderInfo = Omit & { parentId?: string }; export type WorkspaceEvent = diff --git a/web/src/core-bridge/worker.ts b/web/src/core-bridge/worker.ts index e40ea88..a75bcb6 100644 --- a/web/src/core-bridge/worker.ts +++ b/web/src/core-bridge/worker.ts @@ -1,8 +1,10 @@ import { setApiBase, getServerInfo } from './worker/http'; +import type { WorkspaceRoleRequest } from './generated/api'; import { listPasskeys, removePasskey, beginPasskeyRegistration, finishPasskeyRegistration, beginPasskeyRecovery, finishPasskeyRecovery, beginPasskeyLogin, finishPasskeyLogin, unlockPasskeyLogin, clearPasskeyState } from './worker/passkeys'; import { connectChannelResource, getChannelBlob, listArchive, markChannelSeen, removeChannelItem, resolveTextConflict } from './worker/channelItems'; import { createChannel, forgetChannel, joinChannel, listChannels, updateChannelSettings, previewChannel, channelMembers } from './worker/channelResources'; import { editChannelText, sendFiles, sendText, updateSnapshot } from './worker/channelSend'; +import { cancelFileUpload } from './worker/uploadQueue'; import { stopChannel } from './worker/channelTransport'; import { addShareFiles, addShareText, editShareText, removeShareItem } from './worker/shareDraft'; import { connectShare, getShareBlob, stopShare } from './worker/shareRead'; @@ -12,7 +14,7 @@ import { setEventSink } from './worker/port'; import { serial } from './worker/serial'; import { errorMessage } from './errors'; import { accountStatus, loginAccount, logoutAccount, registerAccount, syncAccount, listAccountInvitations, acceptInvitation, inviteChannelUser, listChannelInvitations, revokeChannelInvitation } from './worker/account'; -import { createFolder, createThread, createGroup, renameGroup, deleteGroup, groupThread, moveItems, renameFolder, renameThread, setChannelHistory } from './worker/channelThreads'; +import { createFolder, createThread, createGroup, renameGroup, deleteGroup, createRole, updateRole, deleteRole, setRoleMembers, groupThread, moveItems, renameFolder, renameThread, setChannelHistory } from './worker/channelThreads'; import { restoreAccount, updatePersonal, listAccountDevices, renameAccountDevice, revokeAccountDevice, getSavedLogin, clearSavedLogin } from './worker/account'; interface WorkerRequest { requestId: number; operation: string; payload: unknown } @@ -62,6 +64,10 @@ async function dispatch(operation: string, input: unknown): Promise { case 'updateChannelSettings': return updateChannelSettings(id, payload.patch as Parameters[1]); case 'createThread': return createThread(id, String(payload.name ?? ''), typeof payload.groupId === 'string' ? payload.groupId : undefined); case 'createGroup': return createGroup(id, String(payload.name ?? '')); + case 'createRole': return createRole(id, payload.role as WorkspaceRoleRequest); + case 'updateRole': return updateRole(id, String(payload.roleId ?? ''), payload.role as WorkspaceRoleRequest); + case 'deleteRole': return deleteRole(id, String(payload.roleId ?? '')); + case 'setRoleMembers': return setRoleMembers(id, String(payload.roleId ?? ''), (payload.members as string[] | undefined) ?? []); case 'deleteGroup': return deleteGroup(id, String(payload.groupId ?? '')); case 'renameGroup': return renameGroup(id, String(payload.groupId ?? ''), String(payload.name ?? '')); case 'groupThread': return groupThread(id, String(payload.threadId ?? ''), typeof payload.groupId === 'string' ? payload.groupId : undefined); @@ -77,6 +83,7 @@ async function dispatch(operation: string, input: unknown): Promise { case 'connectChannel': void connectChannelResource(id); return true; case 'stopChannel': stopChannel(id); return true; case 'sendText': return sendText(id, String(payload.text ?? ''), typeof payload.threadId === 'string' ? payload.threadId : undefined, undefined, typeof payload.localId === 'string' ? payload.localId : undefined); + case 'cancelUpload': return cancelFileUpload(id, String(payload.localId ?? '')); case 'sendFiles': return sendFiles(id, payload.files as File[], payload.options as Parameters[2] | undefined, typeof payload.threadId === 'string' ? payload.threadId : undefined, typeof payload.folderId === 'string' ? payload.folderId : undefined, Array.isArray(payload.localIds) ? payload.localIds as string[] : undefined); case 'removeChannelItem': return removeChannelItem(id, String(payload.itemId)); case 'getChannelBlob': return getChannelBlob(id, String(payload.itemId), typeof payload.versionId === 'string' ? payload.versionId : undefined); diff --git a/web/src/core-bridge/worker/account.ts b/web/src/core-bridge/worker/account.ts index 9b0b12c..ee0e081 100644 --- a/web/src/core-bridge/worker/account.ts +++ b/web/src/core-bridge/worker/account.ts @@ -278,8 +278,8 @@ export async function inviteChannelUser(channelId: string, input: string): Promi await apiFetch(`/channels/${channelId}/invitations`, { method: 'POST', headers: { ...session.authHeaders('owner'), 'Content-Type': 'application/json' }, body: JSON.stringify({ username, payload, domain_envelopes: domainEnvelopes }) }); } export async function listChannelInvitations(channelId: string): Promise { - const { session } = await getSession(channelId); return apiFetch(`/channels/${channelId}/invitations`, { headers: session.authHeaders('owner') }).then((r) => r.json() as Promise); + const { session } = await getSession(channelId); return apiFetch(`/channels/${channelId}/invitations`, { headers: session.authHeaders(session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read') }).then((r) => r.json() as Promise); } export async function revokeChannelInvitation(channelId: string, input: string): Promise { - const { session } = await getSession(channelId); await apiFetch(`/channels/${channelId}/invitations/${usernameOf(input)}`, { method: 'DELETE', headers: session.authHeaders('owner') }); + const { session } = await getSession(channelId); await apiFetch(`/channels/${channelId}/invitations/${usernameOf(input)}`, { method: 'DELETE', headers: session.authHeaders(session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read') }); } diff --git a/web/src/core-bridge/worker/channelDataKeys.ts b/web/src/core-bridge/worker/channelDataKeys.ts index b694cce..01fcf46 100644 --- a/web/src/core-bridge/worker/channelDataKeys.ts +++ b/web/src/core-bridge/worker/channelDataKeys.ts @@ -214,6 +214,8 @@ export async function signObjectDeletion(resource: SavedResource, session: Chann const { state } = await readDataKeys(resource, session); const object = state.objects.find(row => row.itemId === itemId); const signed = object?.event; + // 末帧从未上传的对象没有签名清单:没有可签的内容,服务端允许作者和空间主直接删除 + if (object && !signed?.signature) return {}; if (!signed?.manifestHash || !signed.frameCount || (object?.domain !== 'chat' && object?.domain !== 'files')) throw new Error('对象缺少已验签的密文清单，不能生成删除事件'); const manifestHash = fromBase64Url(signed.manifestHash, 32); @@ -231,6 +233,8 @@ export async function verifyObjectDeletion(resource: SavedResource, session: Cha } const deleted = object?.deleteEvent; if (resource.mode === 'raw' && !deleted?.signature) return true; + // 没传完的上传从未发布过有效内容,删掉它不需要事件(见 signObjectDeletion) + if (object && !object.event?.signature && !deleted?.signature) return true; if (!object || !deleted?.signature || !deleted.eventId || !deleted.kind || !deleted.signerDeviceId || !deleted.signerRootPublicKey || !deleted.signerDevicePublicKey || !deleted.deviceCertificateSignature || !object.authorAccount || !object.event?.manifestHash || !object.event.frameCount || (object.domain !== 'chat' && object.domain !== 'files')) return false; diff --git a/web/src/core-bridge/worker/channelSend.ts b/web/src/core-bridge/worker/channelSend.ts index d0d3b24..0041f28 100644 --- a/web/src/core-bridge/worker/channelSend.ts +++ b/web/src/core-bridge/worker/channelSend.ts @@ -1,21 +1,12 @@ import { removeChannelItem } from './channelItems'; -import { queueFileCatalog } from './channelThreads'; +import { enqueueFileUploads } from './uploadQueue'; import { senderSnapshot } from './senderSnapshot'; import { prepareObject } from './channelDataKeys'; -import { errorMessage } from '../errors'; - -import { saveResource, queueLocation } from '../database'; - -import { previewKindOf } from '@/lib/fileKind'; -import type { FileItem, FileUploadOptions, Item } from '../types'; - +import { queueLocation } from '../database'; +import type { FileUploadOptions, Item } from '../types'; import { blobs, getSession } from './sessions'; -import { deviceIdentity } from './presentation'; import { accountDisplay } from './accountSession'; - - - -import { flushOutbox, MAX_FRAME_DATA, MOBILE_UPLOAD_FRAME_DATA_BYTES, persistPlan } from './transfer'; +import { flushOutbox, persistPlan } from './transfer'; import { publishWorkspace } from './presentation'; export async function sendText(resourceId: string, text: string, threadId?: string, @@ -33,93 +24,17 @@ export async function sendText(resourceId: string, text: string, threadId?: stri await persistPlan(resource, plan, undefined, { channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: event?.kind ?? 'chat.create', ...(event?.targetItemId ? { targetItemId: event.targetItemId } : {}) }); await queueLocation(resourceId, item.id, threadId); - const uploaded = await flushOutbox(resource, session); + // 只发这条消息自己的帧,不被队列里已有的大文件堵住 + const uploaded = await flushOutbox(resource, session, undefined, { only: new Set([plan.itemId]) }); publishWorkspace(resourceId, { t: 'item', item: { ...item, state: uploaded ? { kind: 'complete' } : { kind: 'incomplete' } } }); if (!uploaded) publishWorkspace(resourceId, { t: 'connection', state: 'offline' }); return { itemId: plan.itemId, uploaded }; } +/** 文件进入该空间的上传队列:一次传一个,其余排队。文件内部的帧并发上传,见 uploadQueue */ export async function sendFiles(resourceId: string, files: File[], suppliedOptions?: FileUploadOptions, threadId?: string, folderId?: string, localIds?: string[]): Promise { if (!threadId) throw new Error('请先选择频道'); - const { session, resource } = await getSession(resourceId); - const options = suppliedOptions ?? { directPreview: true, maxDownloads: resource.maxDownloads ?? 0 }; - if (resource.kind !== 'channel' || (!session.can('owner') && !session.can('write'))) throw new Error('此链接没有上传权限'); - const device = await deviceIdentity(session.endpointSecret); - const maxFrameData = device.kind === 'phone' ? MOBILE_UPLOAD_FRAME_DATA_BYTES : MAX_FRAME_DATA; - const author = await senderSnapshot(); - for (const [index, file] of files.entries()) { - const localId = localIds?.[index]; - const ts = Date.now(); - let item: FileItem | undefined; - try { - const prepared = await prepareObject(resource, session, threadId, 'files'); - const plan = session.encryptFile(file, {}, { - itemId: prepared.itemIdBytes, - ...(options.directPreview ? { gated: 'auto' as const } : { gated: true, maxDownloads: options.maxDownloads }), - ts, maxFrameData, threadId, folderId, - }); - item = { - id: plan.itemId, kind: 'file', ts, state: { kind: 'receiving', progress: 0, phase: 'encrypting' }, - ...(author ? { author } : {}), - ...(threadId ? { threadId } : {}), ...(folderId ? { folderId } : {}), ...(localId ? { localId } : {}), - name: file.name, mime: file.type || 'application/octet-stream', size: file.size, - remaining: plan.gated - ? (options.directPreview ? resource.maxDownloads ?? 0 : options.maxDownloads) || null - : null, - preview: previewKindOf(file.type, file.name), gated: plan.gated, - }; - resource.localItems = [...(resource.localItems ?? []).filter((entry) => entry.id !== item?.id), item]; - await saveResource(resource); - blobs.set(`${resourceId}:${item.id}`, file); - publishWorkspace(resourceId, { t: 'item', item }); - const updateProgress = (progress: number, phase: 'encrypting' | 'uploading', speedBytesPerSecond?: number) => { - if (!item) return; - item = { - ...item, - state: { - kind: 'receiving', - progress: Math.max(0, Math.min(0.99, progress)), - phase, - ...(speedBytesPerSecond === undefined ? {} : { speedBytesPerSecond }), - }, - }; - publishWorkspace(resourceId, { t: 'item', item }); - }; - const planned = await persistPlan(resource, plan, (_frames) => { - const processed = Math.min(file.size, _frames * MAX_FRAME_DATA); - updateProgress(file.size === 0 ? 0.5 : Math.min(0.5, processed / file.size * 0.5), 'encrypting'); - }, { channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: 'file.publish' }); - const uploadedFrames = new Set(); - await queueFileCatalog(resource, session, item); - await queueLocation(resourceId, item.id, threadId, folderId); - let uploadedBytes = 0; - const uploadStartedAt = performance.now(); - const uploaded = await flushOutbox(resource, session, (frame, loadedBytes, totalBytes, complete) => { - if (frame.itemId !== item?.id) return; - if (complete && !uploadedFrames.has(frame.index)) { - uploadedFrames.add(frame.index); - uploadedBytes += totalBytes; - } - const currentBytes = uploadedBytes + (complete ? 0 : loadedBytes); - const elapsed = Math.max(1, performance.now() - uploadStartedAt); - updateProgress(0.5 + 0.5 * currentBytes / planned.bytes, 'uploading', currentBytes * 1000 / elapsed); - }); - item = { ...item, state: uploaded ? { kind: 'complete' } : { kind: 'incomplete' } }; - resource.localItems = resource.localItems.map((entry) => entry.id === item?.id ? item! : entry); - await saveResource(resource); - publishWorkspace(resourceId, { t: 'item', item }); - if (!uploaded) publishWorkspace(resourceId, { t: 'error', message: '上传暂停了。重新打开空间并连接网络后,可以继续发送已加密内容。' }); - } catch (error) { - if (item) { - item = { ...item, state: { kind: 'incomplete' } }; - resource.localItems = (resource.localItems ?? []).map((entry) => entry.id === item?.id ? item! : entry); - await saveResource(resource); - publishWorkspace(resourceId, { t: 'item', item }); - } - publishWorkspace(resourceId, { t: 'error', message: `文件上传失败：${errorMessage(error)}` }); - throw error; - } - } + return enqueueFileUploads(resourceId, files, suppliedOptions, threadId, folderId, localIds); } export async function updateSnapshot(resourceId: string, text: string): Promise { diff --git a/web/src/core-bridge/worker/channelThreads.ts b/web/src/core-bridge/worker/channelThreads.ts index 1d4f4d7..3c87696 100644 --- a/web/src/core-bridge/worker/channelThreads.ts +++ b/web/src/core-bridge/worker/channelThreads.ts @@ -2,13 +2,13 @@ import type { ChannelSession } from '@webdrop/core'; import type { SavedResource } from '../database'; import type { FileItem } from '../types'; import { persistPlan } from './transfer'; -import { changeWorkspace, createWorkspaceChannel, createWorkspaceGroup } from './workspaceStructure'; +import { changeWorkspace, createWorkspaceChannel, createWorkspaceGroup, createWorkspaceRole } from './workspaceStructure'; import { getSession } from './sessions'; import { initializeChannelKeys, prepareObject } from './channelDataKeys'; import { toPublishedFile } from './fileCatalog'; import { apiFetch } from './http'; import { WORKSPACE_APP, type ResourceControl } from '../protocol/workspace'; -import type { WorkspaceChannelRequest, WorkspaceGroupRequest, WorkspaceNameRequest, WorkspaceFolderRequest, WorkspaceMoveRequest } from '../generated/api'; +import type { WorkspaceChannelRequest, WorkspaceGroupRequest, WorkspaceNameRequest, WorkspaceFolderRequest, WorkspaceMoveRequest, WorkspaceRoleRequest, WorkspaceRoleMembersRequest } from '../generated/api'; const encoder = new TextEncoder(); @@ -27,6 +27,14 @@ export async function renameGroup(resourceId: string, id: string, name: string): export async function deleteGroup(resourceId: string, id: string): Promise { await changeWorkspace(resourceId, 'groups/' + id, 'DELETE'); } +export async function createRole(resourceId: string, role: WorkspaceRoleRequest): Promise { return createWorkspaceRole(resourceId, role); } +export async function updateRole(resourceId: string, id: string, role: WorkspaceRoleRequest): Promise { + await changeWorkspace(resourceId, 'roles/' + id, 'PATCH', role); +} +export async function deleteRole(resourceId: string, id: string): Promise { await changeWorkspace(resourceId, 'roles/' + id, 'DELETE'); } +export async function setRoleMembers(resourceId: string, id: string, members: string[]): Promise { + await changeWorkspace(resourceId, 'roles/' + id + '/members', 'PUT', { members } satisfies WorkspaceRoleMembersRequest); +} export async function groupThread(resourceId: string, id: string, groupId?: string): Promise { await changeWorkspace(resourceId, 'channels/' + id + '/group', 'PUT', { groupId: groupId ?? null } satisfies WorkspaceGroupRequest); } diff --git a/web/src/core-bridge/worker/channelTransport.ts b/web/src/core-bridge/worker/channelTransport.ts index 85e2a4d..8da25a9 100644 --- a/web/src/core-bridge/worker/channelTransport.ts +++ b/web/src/core-bridge/worker/channelTransport.ts @@ -5,6 +5,7 @@ import { publishWorkspace } from './presentation'; import { channelMetadata, processChannelBytes, syncChannel } from './channelEvents'; import { loadWorkspace } from './workspaceStructure'; import { flushOutbox } from './transfer'; +import { resumeOutbox } from './resumeUploads'; import { socketStream, sseStream, delay, type StreamConsumer } from './transports'; import { errorMessage } from '../errors'; import { saveResource } from '../database'; @@ -43,19 +44,24 @@ export async function connectChannel(id: string): Promise { let initialized = false; let transport = 0; let backoff = 1000; - flushTimer = setInterval(() => { - if (!signal.aborted) void flushOutbox(resource, session).catch((error: unknown) => { - publishWorkspace(id, { t: 'error', message: errorMessage(error) }); - }); - }, 2500); + // 离线队列的续传在后台跑:几个 GB 的遗留不能挡住频道加载,也不能叠出一堆并行的续传 + let resuming = false; + const resume = () => { + if (signal.aborted || resuming) return; + resuming = true; + void resumeOutbox(resource, session, signal).then(() => flushOutbox(resource, session)) + .catch((error: unknown) => { if (!signal.aborted) publishWorkspace(id, { t: 'error', message: errorMessage(error) }); }) + .finally(() => { resuming = false; }); + }; + flushTimer = setInterval(resume, 2500); while (!signal.aborted) { try { if (!initialized) { [stream.cursor] = await Promise.all([syncChannel(resource, session, signal), channelMetadata(resource, session), joining]); if (signal.aborted) break; - await flushOutbox(resource, session); initialized = true; publishWorkspace(id, { t: 'ready' }); + resume(); } stream.online(); if (transport === 0) await socketStream(id, session, signal, stream); diff --git a/web/src/core-bridge/worker/frameStash.test.ts b/web/src/core-bridge/worker/frameStash.test.ts new file mode 100644 index 0000000..ec5bf2b --- /dev/null +++ b/web/src/core-bridge/worker/frameStash.test.ts @@ -0,0 +1,74 @@ +import { describe, expect, it } from 'vitest'; +import type { QueuedFrame } from '../database'; +import { FrameStash } from './frameStash'; + +const frame = (index: number): QueuedFrame => ({ key: `k${index}`, resourceId: 'r', itemId: 'i', index, bytes: new Uint8Array(10), extraHeaders: {}, createdAt: index }); +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +function fakeDisk(writeMs = 5) { + const stored = new Set(); const writes: string[] = []; const removes: string[] = []; + return { + stored, writes, removes, + put: async (f: QueuedFrame) => { await sleep(writeMs); stored.add(f.key); writes.push(f.key); }, + remove: async (key: string) => { stored.delete(key); removes.push(key); }, + }; +} + +describe('FrameStash', () => { + it('宽限期内就被确认的帧:根本不写盘', async () => { + const disk = fakeDisk(5); + const stash = new FrameStash(disk.put, disk.remove, 8, 100, 4); + const frames = [0, 1, 2, 3].map(frame); + for (const f of frames) await stash.add(f); + await sleep(20); + for (const f of frames) stash.accepted(f); + await stash.drain(); + expect(disk.writes).toEqual([]); + expect(disk.stored.size).toBe(0); + }); + + it('超过宽限期还没确认的帧会写入,之后确认时再删除', async () => { + const disk = fakeDisk(1); + const stash = new FrameStash(disk.put, disk.remove, 8, 30, 4); + const first = frame(0); await stash.add(first); + await sleep(80); + expect(disk.stored.has('k0')).toBe(true); + stash.accepted(first); + await sleep(10); + expect(disk.stored.has('k0')).toBe(false); + }); + + it('断网时(一直不确认)drain 之后全部帧都已落盘,续传才有东西可发', async () => { + const disk = fakeDisk(1); + const stash = new FrameStash(disk.put, disk.remove, 3, 5000, 4); + for (let index = 0; index < 10; index += 1) await stash.add(frame(index)); + await stash.drain(); + expect(disk.stored.size).toBe(10); + }); + + it('积压超过上限时 add 会等待写盘,不让内存里越堆越多', async () => { + const disk = fakeDisk(30); + const stash = new FrameStash(disk.put, disk.remove, 2, 5000, 1); + let added = 0; + const producer = (async () => { for (let index = 0; index < 6; index += 1) { await stash.add(frame(index)); added += 1; } })(); + await sleep(40); + expect(added).toBeLessThan(6); + await producer; await stash.drain(); + expect(disk.stored.size).toBe(6); + }); + + it('取消后不会再写入新的帧', async () => { + const disk = fakeDisk(5); + const stash = new FrameStash(disk.put, disk.remove, 8, 20, 4); + await stash.add(frame(0)); await stash.add(frame(1)); + await stash.discard(); + const before = disk.writes.length; + await sleep(60); + expect(disk.writes.length).toBe(before); + }); + + it('写盘失败会让加密端停下来并抛出', async () => { + const stash = new FrameStash(async () => { throw new Error('quota'); }, async () => {}, 1, 0, 0); + await expect((async () => { for (let index = 0; index < 5; index += 1) await stash.add(frame(index)); await stash.drain(); })()).rejects.toThrow('quota'); + }); +}); diff --git a/web/src/core-bridge/worker/frameStash.ts b/web/src/core-bridge/worker/frameStash.ts new file mode 100644 index 0000000..a1fe51f --- /dev/null +++ b/web/src/core-bridge/worker/frameStash.ts @@ -0,0 +1,85 @@ +import type { QueuedFrame } from '../database'; + +interface Pending { frame: QueuedFrame; at: number } + +/** + * 离线队列的「后写」:加密好的帧先在内存里直接上传,只有没能及时上传的才写进 IndexedDB 留作续传保险。 + * 每帧写盘要几十毫秒,同步写会把上传速度压到约 25 MB/s;网络够快时帧在宽限期内就已被确认,根本不用写。 + * 网络慢或断了,帧超过宽限期(或积压太多)就在后台写入,离线续传不受影响(protocol §5:重试发送的仍是同一份字节)。 + */ +export class FrameStash { + private pending: Pending[] = []; + private readonly written = new Set(); + private writing: Promise | undefined; + private error: unknown; + private forced = false; + private waiters: (() => void)[] = []; + + constructor( + private readonly put: (frame: QueuedFrame) => Promise, + private readonly remove: (key: string) => Promise, + /** 还没写入的帧超过这个数就让加密端等一等,避免断网后内存里越堆越多 */ + private readonly maxPending = 16, + /** 帧在内存里等服务端确认的宽限时间,超过才写盘 */ + private readonly graceMs = 400, + /** 积压超过这个数就不再等宽限期,马上写最早的。要大于上传流水线本身在途+缓冲的帧数(约 10),否则正常上传也会被逼着写盘 */ + private readonly eagerAbove = 12, + ) {} + + /** 新加密出一帧。积压太多时等待写盘追上来 */ + async add(frame: QueuedFrame): Promise { + if (this.error) throw this.error; + this.pending.push({ frame, at: performance.now() }); + this.kick(); + this.notify(); + while (this.pending.length > this.maxPending && !this.error) await this.wait(); + if (this.error) throw this.error; + } + + /** 服务端确认了这一帧:还没写盘的不用写了,已经写了的从队列里删掉 */ + accepted(frame: QueuedFrame): void { + const at = this.pending.findIndex((entry) => entry.frame.key === frame.key); + if (at >= 0) { this.pending.splice(at, 1); this.notify(); return; } + if (this.written.delete(frame.key)) void this.remove(frame.key).catch(() => undefined); + } + + /** 等所有该写的帧都落盘(不再等宽限期)。全部上传完成时这里几乎是空的 */ + async drain(): Promise { + this.forced = true; this.kick(); this.notify(); + while ((this.pending.length || this.writing) && !this.error) await this.wait(); + if (this.error) throw this.error; + } + + /** 取消上传:丢掉还没写的帧,等正在写的那一帧结束。已写入的由调用方用 dequeueItem 清掉 */ + async discard(): Promise { + this.pending = []; + this.notify(); + await this.writing; + } + + private kick(): void { + if (this.writing || !this.pending.length) return; + this.writing = (async () => { + try { + while (this.pending.length) { + const head = this.pending[0]!; + const age = performance.now() - head.at; + if (!this.forced && age < this.graceMs && this.pending.length <= this.eagerAbove && this.pending.length < this.maxPending) { + // 还在宽限期里:等它被确认(会被移出 pending)、或有新帧到来、或宽限期到 + await Promise.race([this.wait(), new Promise((resolve) => setTimeout(resolve, this.graceMs - age))]); + continue; + } + await this.put(head.frame); + // 写的过程中如果已经被确认,accepted() 已把它移出 pending:这次写入白做了,补一个删除 + if (this.pending[0] === head) { this.pending.shift(); this.written.add(head.frame.key); } + else await this.remove(head.frame.key).catch(() => undefined); + this.notify(); + } + } catch (error) { this.error = error; } + finally { this.writing = undefined; this.notify(); } + })(); + } + + private wait(): Promise { return new Promise((resolve) => { this.waiters.push(resolve); }); } + private notify(): void { const waiting = this.waiters; this.waiters = []; for (const resume of waiting) resume(); } +} diff --git a/web/src/core-bridge/worker/http.ts b/web/src/core-bridge/worker/http.ts index cbaf0d4..7c160cd 100644 --- a/web/src/core-bridge/worker/http.ts +++ b/web/src/core-bridge/worker/http.ts @@ -2,7 +2,7 @@ import { supportsApiContract } from '../contract'; import { accountToken } from './accountSession'; import type { ServerInfoBody } from '../generated/api'; -export type ApiError = Error & { status?: number; code?: string }; +export type ApiError = Error & { status?: number; code?: string; /** 服务端要求的等待秒数(429 的 Retry-After) */ retryAfter?: number }; let apiBase = '/api/v1'; let contractCheck: Promise | undefined; @@ -69,24 +69,37 @@ export async function apiFetch(path: string, init: RequestInit = {}): Promise, - body: ArrayBuffer, + body: ArrayBuffer | Uint8Array | Blob, onProgress: (loadedBytes: number, totalBytes: number) => void, + signal?: AbortSignal, + /** 这么久没有任何上传进展就放弃这次请求(弱网下连接静默卡死),由调用方重试 */ + stallTimeoutMs?: number, ): Promise { try { await getServerInfo(); const response = await new Promise((resolve, reject) => { + if (signal?.aborted) { reject(new DOMException('上传已取消', 'AbortError')); return; } const xhr = new XMLHttpRequest(); + signal?.addEventListener('abort', () => xhr.abort(), { once: true }); + let stalled = false; let stallTimer: ReturnType | undefined; + const watch = () => { + if (!stallTimeoutMs) return; + clearTimeout(stallTimer); + stallTimer = setTimeout(() => { stalled = true; xhr.abort(); }, stallTimeoutMs); + }; xhr.open('POST', apiPath(path)); xhr.withCredentials = false; xhr.responseType = 'text'; - xhr.setRequestHeader('Cache-Control', 'no-store'); + // 不要设 Cache-Control 请求头:它不在服务端 CORS 允许的头里,跨域部署(API 在另一个域名)时预检会失败;POST 本来也不会被缓存 const token = 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) => { + watch(); if (event.lengthComputable) onProgress(event.loaded, event.total); }; xhr.onload = () => { + clearTimeout(stallTimer); if (xhr.status === 0) { reject(new Error('network_error')); return; @@ -94,20 +107,27 @@ export async function apiUpload( const responseHeaders = new Headers(); const contentType = xhr.getResponseHeader('Content-Type'); if (contentType) responseHeaders.set('Content-Type', contentType); + const retryAfter = xhr.getResponseHeader('Retry-After'); + if (retryAfter) responseHeaders.set('Retry-After', retryAfter); resolve(new Response(xhr.responseText, { status: xhr.status, statusText: xhr.statusText, headers: responseHeaders })); }; - xhr.onerror = () => reject(new Error('network_error')); - xhr.onabort = () => reject(new DOMException('上传已取消', 'AbortError')); - xhr.send(body); + xhr.onerror = () => { clearTimeout(stallTimer); reject(new Error('network_error')); }; + xhr.onabort = () => { clearTimeout(stallTimer); reject(stalled ? new Error('network_error') : new DOMException('上传已取消', 'AbortError')); }; + watch(); + xhr.send(body as XMLHttpRequestBodyInit); }); if (!response.ok) { const result = await response.json().catch(() => null) as { error?: unknown } | null; const code = typeof result?.error === 'string' && /^[a-z_]{1,40}$/.test(result.error) ? result.error : 'request_failed'; - throw fail(response.status, code); + const error = fail(response.status, code); + const wait = Number(response.headers.get('Retry-After')); + if (Number.isFinite(wait) && wait > 0) error.retryAfter = wait; + throw error; } } catch (error) { if (error instanceof Error && 'status' in error) throw error; if (error instanceof Error && error.message === 'api_contract_mismatch') throw error; + if (signal?.aborted) throw error; throw new Error('网络连接失败'); } } diff --git a/web/src/core-bridge/worker/resourceMetadata.test.ts b/web/src/core-bridge/worker/resourceMetadata.test.ts index 3334930..5661dd3 100644 --- a/web/src/core-bridge/worker/resourceMetadata.test.ts +++ b/web/src/core-bridge/worker/resourceMetadata.test.ts @@ -22,7 +22,7 @@ describe('resource and rendering boundaries', () => { it('never synchronizes draft content or server structure snapshots inside the key vault', () => { const resource: SavedResource = { id: 'a', kind: 'share', mode: 'enc', role: 'owner', link: 'key', createdAt: 1, expiresAt: 2, endpointSecret: new Uint8Array(32), maxMembers: 0, draftItems: [], manifestEntries: [], - workspaceMetadata: { structure: { revision: 1, channels: [], groups: [], folders: [], locations: [] }, people: [] } }; + workspaceMetadata: { structure: { revision: 1, channels: [], groups: [], roles: [], rolesSeeded: true, folders: [], locations: [] }, people: [] } }; const exported = exportResource(resource); for (const field of ['draftItems', 'manifestEntries', 'workspaceMetadata', 'endpointSecret']) expect(exported).not.toHaveProperty(field); }); diff --git a/web/src/core-bridge/worker/resumeUploads.ts b/web/src/core-bridge/worker/resumeUploads.ts new file mode 100644 index 0000000..c71bac1 --- /dev/null +++ b/web/src/core-bridge/worker/resumeUploads.ts @@ -0,0 +1,87 @@ +import { dequeueItem, queuedFrame, queuedFrameKeys, saveResource, type QueuedFrame, type SavedResource } from '../database'; +import type { ChannelSession } from '@webdrop/core'; +import { publishWorkspace } from './presentation'; +import { pipelinedItems, uploadFrame } from './transfer'; +import { FrameUploader, TransferMeter } from './uploadPipeline'; + +const FRAME_CONCURRENCY = 4; +const keyOf = (key: string) => { const [, itemId = '', index = '0'] = key.split(':'); return { itemId, index: Number(index) }; }; +/** 内容事件的签名头只在末帧上;离线队列里没有带签名头的末帧,说明上次加密被打断,这个条目无法补完 */ +const isFinalFrame = (frame: QueuedFrame): boolean => 'X-WD-Event-Signature' in frame.extraHeaders; + +/** + * 续传离线队列里上次没发完的内容:按条目分组,同一个条目内的帧按 FrameUploader 的规则并发上传, + * 进度会回到聊天里的文件卡片。在后台运行,不挡住频道加载。 + * 返回 true 表示没有遗留(或都发完了)。 + */ +export async function resumeOutbox(resource: SavedResource, session: ChannelSession, signal?: AbortSignal): Promise { + const items = new Map(); + for (const key of await queuedFrameKeys(resource.id)) { + const { itemId, index } = keyOf(key); + if (pipelinedItems.has(`${resource.id}:${itemId}`)) continue; + (items.get(itemId) ?? items.set(itemId, []).get(itemId)!).push(index); + } + // 上次会话里还显示「传输中」的文件,如果既没有在续传、也没有可续传的帧,就是被打断的:标成未完成,用户可以删除 + const stale = (resource.localItems ?? []).filter((item) => item.kind === 'file' && item.state.kind === 'receiving' + && !pipelinedItems.has(`${resource.id}:${item.id}`) && !items.has(item.id)); + for (const item of stale) await markIncomplete(resource, item.id); + let complete = true; + for (const [itemId, indices] of items) { + if (signal?.aborted) return false; + if (!await resumeItem(resource, session, itemId, indices.sort((a, b) => a - b), signal)) complete = false; + } + return complete; +} + +async function resumeItem(resource: SavedResource, session: ChannelSession, itemId: string, indices: number[], signal?: AbortSignal): Promise { + const keyFor = (index: number) => `${resource.id}:${itemId}:${String(index).padStart(10, '0')}`; + const last = indices[indices.length - 1]!; + const lastFrame = await queuedFrame(keyFor(last)); + // 没有末帧的条目补不完整:留给全局队列处理,或等用户重新发送 + if (!lastFrame || !isFinalFrame(lastFrame)) { await markIncomplete(resource, itemId); return true; } + const queued = new Set(indices); + const accepted: number[] = []; + for (let index = 0; index <= last; index += 1) if (!queued.has(index)) accepted.push(index); + const local = resource.localItems?.find((item) => item.id === itemId); + const controller = new AbortController(); + signal?.addEventListener('abort', () => controller.abort(), { once: true }); + const meter = new TransferMeter(); + const total = last + 1; + let lastPublished = 0; + const publish = (sent: number, force = false) => { + if (!local || local.kind !== 'file') return; + const now = performance.now(); + if (!force && now - lastPublished < 150) return; + lastPublished = now; + const speed = meter.speed(); + publishWorkspace(resource.id, { t: 'item', item: { ...local, state: { kind: 'receiving', progress: Math.min(0.99, (accepted.length + sent) / total), phase: 'uploading', ...(speed ? { speedBytesPerSecond: speed } : {}) } } }); + }; + const uploader = new FrameUploader({ + gated: local?.kind === 'file' ? Boolean(local.gated) : false, concurrency: FRAME_CONCURRENCY, signal: controller.signal, alreadyAccepted: accepted, + send: (frame, hooks, s) => uploadFrame(resource, session, frame, hooks, s), + onBytes: (bytes) => { meter.add(bytes); publish(bytes / Math.max(1, lastFrame.bytes.byteLength)); }, + }); + try { + for (const index of indices) { + const frame = index === last ? lastFrame : await queuedFrame(keyFor(index)); + if (frame) await uploader.push(frame, index === last); + } + const done = await uploader.finish(); + if (local && local.kind === 'file') publishWorkspace(resource.id, { t: 'item', item: { ...local, state: done ? { kind: 'complete' } : { kind: 'incomplete' } } }); + return done; + } catch (error) { + if (controller.signal.aborted) return false; + // 服务端明确拒绝(例如已被删除):这些帧再发也没用,清掉 + await dequeueItem(resource.id, itemId); + throw error; + } +} + +async function markIncomplete(resource: SavedResource, itemId: string): Promise { + const item = resource.localItems?.find((entry) => entry.id === itemId); + if (!item || item.state.kind === 'incomplete') return; + const next = { ...item, state: { kind: 'incomplete' as const } }; + resource.localItems = (resource.localItems ?? []).map((entry) => entry.id === itemId ? next : entry); + await saveResource(resource); + publishWorkspace(resource.id, { t: 'item', item: next }); +} diff --git a/web/src/core-bridge/worker/transfer.ts b/web/src/core-bridge/worker/transfer.ts index 85d910f..b29968e 100644 --- a/web/src/core-bridge/worker/transfer.ts +++ b/web/src/core-bridge/worker/transfer.ts @@ -1,5 +1,5 @@ import { fromBase64Url, IntegrityError, MAX_FRAME_DATA, type ChannelSession, type ItemDecodeResult, type ItemPlan } from '@webdrop/core'; -import { dequeueFrame, queueFrame, nextQueuedFrame, lastQueuedAt, nextQueuedLocation, dequeueLocation, type SavedResource } from '../database'; +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'; @@ -11,21 +11,28 @@ import { storeVerifiedDownload } from './downloadStorage'; // Frame plaintext includes a 4-byte length prefix before padding to the 256 KiB tier. export const MOBILE_UPLOAD_FRAME_DATA_BYTES = 256 * 1024 - 4; export { MAX_FRAME_DATA }; -const UPLOAD_RETRY_DELAYS_MS = [500, 1500, 4000] as const; + +/** 正由上传流水线逐帧发送的条目(空间 id:条目 id)。全局续传 flushOutbox 要绕开它们,否则同一帧会发两遍、进度互相抢 */ +const pipelined = new Set(); +export { pipelined as pipelinedItems }; +export function markPipelined(resourceId: string, itemId: string): void { pipelined.add(`${resourceId}:${itemId}`); } +export function unmarkPipelined(resourceId: string, itemId: string): void { pipelined.delete(`${resourceId}:${itemId}`); } export async function persistPlan( resource: SavedResource, plan: ItemPlan, onProgress?: (frames: number, bytes: number) => void, signedEvent?: { channelId: string; domain: 'chat' | 'files'; epoch: number; eventKind: string; targetItemId?: string }, + /** 每一帧写入离线队列之后调用;上传流水线借此边加密边上传,返回的 Promise 用来给加密端施加背压 */ + onFrame?: (frame: QueuedFrame, final: boolean) => Promise, ): Promise<{ frames: number; bytes: number }> { const createdAt = Math.max(Date.now(), (await lastQueuedAt(resource.id)) + 1); let frames = 0; let bytes = 0; const frameDigests: Uint8Array[] = []; let pending: { index: number; bytes: Uint8Array; extraHeaders?: Record } | undefined; - const persistFrame = async (frame: NonNullable, extraHeaders: Record = {}) => { - await queueFrame({ + const persistFrame = async (frame: NonNullable, extraHeaders: Record = {}, final = false) => { + const queued: QueuedFrame = { key: `${resource.id}:${plan.itemId}:${String(frame.index).padStart(10, '0')}`, resourceId: resource.id, itemId: plan.itemId, @@ -33,15 +40,17 @@ export async function persistPlan( bytes: frame.bytes, extraHeaders: { ...(frame.index === 0 ? plan.frame0Headers : {}), ...frame.extraHeaders, ...extraHeaders }, createdAt: createdAt + frame.index, - }); + }; + // 提供了 onFrame 的调用方(上传流水线)自己决定何时落盘,见 frameStash + if (!onFrame) await queueFrame(queued); frames += 1; bytes += frame.bytes.byteLength; onProgress?.(frames, bytes); + await onFrame?.(queued, final); }; for await (const frame of plan.frames) { if (pending) await persistFrame(pending); - const frameBuffer = frame.bytes.slice().buffer as ArrayBuffer; - frameDigests.push(new Uint8Array(await crypto.subtle.digest('SHA-256', frameBuffer))); + frameDigests.push(new Uint8Array(await crypto.subtle.digest('SHA-256', frame.bytes))); pending = frame; } if (pending) { @@ -50,7 +59,7 @@ export async function persistPlan( epoch: signedEvent.epoch, itemId: plan.itemId, eventKind: signedEvent.eventKind, ...(signedEvent.targetItemId ? { targetItemId: signedEvent.targetItemId } : {}), frameDigests }) : {}; - await persistFrame(pending, headers); + await persistFrame(pending, headers, true); } return { frames, bytes }; } @@ -61,25 +70,88 @@ function writeRole(session: ChannelSession): 'owner' | 'write' { throw new Error('此链接没有上传权限'); } -/** The outbox contains ciphertext only. Every retry sends these exact bytes. */ +/** 一次请求这么久没有任何上传进展就放弃重试:弱网下连接可能静默卡死,不能一直等 */ +const FRAME_STALL_TIMEOUT_MS = 30_000; +const MAX_UPLOAD_ATTEMPTS = 8; +const MAX_RETRY_WAIT_MS = 60_000; + +/** 第 n 次失败后的等待:指数退避加抖动;服务端给了 Retry-After(限流)就按它来 */ +export function retryDelayMs(attempt: number, retryAfterSeconds?: number): number { + if (retryAfterSeconds) return Math.min(MAX_RETRY_WAIT_MS, retryAfterSeconds * 1000 + Math.random() * 250); + return Math.min(15_000, 500 * 2 ** attempt) * (0.75 + Math.random() * 0.5); +} + +/** + * 上传一帧。可重试的错误(断网、卡住、5xx、429 等)按退避重试,重试发送的永远是同一份字节(protocol §5)。 + * 返回 false 表示重试用尽(多半断网),帧仍留在离线队列;不可重试的错误会抛出;取消会抛 AbortError。 + * manageQueue 为 false 时调用方自己管理离线队列(frameStash),这里不动它。 + */ +export async function uploadFrame( + resource: SavedResource, + session: ChannelSession, + frame: QueuedFrame, + hooks?: { onLoaded?: (loadedBytes: number, totalBytes: number) => void; onAccepted?: () => void; onRetry?: () => void; manageQueue?: boolean }, + signal?: AbortSignal, +): Promise { + const role = resource.kind === 'share' ? 'owner' : writeRole(session); + const path = resource.kind === 'share' ? `/shares/${resource.id}/frames` : `/channels/${resource.id}/frames`; + let maxLoadedBytes = 0; + for (let attempt = 0; attempt < MAX_UPLOAD_ATTEMPTS; attempt += 1) { + try { + 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); }, + signal, + FRAME_STALL_TIMEOUT_MS, + ); + if (hooks?.manageQueue !== false) await dequeueFrame(frame.key); + hooks?.onAccepted?.(); + return true; + } catch (error) { + if (signal?.aborted || (error as Error | undefined)?.name === 'AbortError') throw error; + const apiError = error as ApiError; + const retryable = apiError.status === undefined || apiError.status === 408 || apiError.status === 425 + || apiError.status === 429 || apiError.status === 499 || apiError.status >= 500; + if (!retryable) throw error; + if (attempt === MAX_UPLOAD_ATTEMPTS - 1) return false; + hooks?.onRetry?.(); + maxLoadedBytes = 0; + await new Promise((resolve) => setTimeout(resolve, retryDelayMs(attempt, apiError.retryAfter))); + } + } + return false; +} + +/** + * 把离线队列里的帧(和待登记的位置)发完。默认处理整个空间的队列,但跳过正由上传流水线处理的条目; + * only 指定时只处理这些条目,发文字消息就用它,不被已入队的大文件堵住。 + */ export async function flushOutbox( resource: SavedResource, session: ChannelSession, onFrameProgress?: (frame: { itemId: string; index: number }, loadedBytes: number, totalBytes: number, complete: boolean) => void, + scope?: { only?: ReadonlySet }, ): Promise { - return serial(`outbox:${resource.id}`, () => flush(resource, session, onFrameProgress)); + const lane = scope?.only ? [...scope.only].sort().join(',') : 'all'; + return serial(`outbox:${resource.id}:${lane}`, () => flush(resource, session, onFrameProgress, scope)); } async function flush( resource: SavedResource, session: ChannelSession, onFrameProgress?: (frame: { itemId: string; index: number }, loadedBytes: number, totalBytes: number, complete: boolean) => void, + scope?: { only?: ReadonlySet }, ): Promise { + const accept = (itemId: string) => !pipelined.has(`${resource.id}:${itemId}`) && (!scope?.only || scope.only.has(itemId)); for (;;) { - const frame = await nextQueuedFrame(resource.id); + const frame = await nextQueuedFrame(resource.id, accept); if (!frame) { if (resource.kind !== 'channel') return true; - const location = await nextQueuedLocation(resource.id); + const location = await nextQueuedLocation(resource.id, accept); if (!location) return true; try { await registerItemLocation(resource.id, { id: location.itemId, threadId: location.threadId, folderId: location.folderId }); @@ -91,34 +163,10 @@ async function flush( } continue; } - const role = resource.kind === 'share' ? 'owner' : writeRole(session); - const path = resource.kind === 'share' ? `/shares/${resource.id}/frames` : `/channels/${resource.id}/frames`; - let maxLoadedBytes = 0; - let accepted = false; - for (let attempt = 0; attempt <= UPLOAD_RETRY_DELAYS_MS.length; attempt += 1) { - try { - await apiUpload( - path, - { ...session.authHeaders(role), 'Content-Type': 'application/octet-stream', ...frame.extraHeaders }, - frame.bytes.slice().buffer as ArrayBuffer, - (loaded, total) => { - maxLoadedBytes = Math.max(maxLoadedBytes, loaded); - onFrameProgress?.(frame, maxLoadedBytes, total, false); - }, - ); - await dequeueFrame(frame.key); - onFrameProgress?.(frame, frame.bytes.byteLength, frame.bytes.byteLength, true); - accepted = true; - break; - } catch (error) { - const apiError = error as ApiError; - const retryable = apiError.status === undefined || apiError.status === 408 || apiError.status === 425 - || apiError.status === 429 || apiError.status === 499 || apiError.status >= 500; - if (!retryable) throw error; - if (attempt === UPLOAD_RETRY_DELAYS_MS.length) return false; - await new Promise((resolve) => setTimeout(resolve, UPLOAD_RETRY_DELAYS_MS[attempt])); - } - } + const accepted = await uploadFrame(resource, session, frame, { + onLoaded: (loaded, total) => onFrameProgress?.(frame, loaded, total, false), + onAccepted: () => onFrameProgress?.(frame, frame.bytes.byteLength, frame.bytes.byteLength, true), + }); if (!accepted) return false; } } diff --git a/web/src/core-bridge/worker/uploadPipeline.test.ts b/web/src/core-bridge/worker/uploadPipeline.test.ts new file mode 100644 index 0000000..9110a67 --- /dev/null +++ b/web/src/core-bridge/worker/uploadPipeline.test.ts @@ -0,0 +1,139 @@ +import { describe, expect, it } from 'vitest'; +import type { QueuedFrame } from '../database'; +import { estimateCipherBytes, FrameUploader, TransferMeter, type SendFrame } from './uploadPipeline'; + +const frame = (index: number, size = 100): QueuedFrame => ({ key: `k${index}`, resourceId: 'r', itemId: 'i', index, bytes: new Uint8Array(size), extraHeaders: {}, createdAt: index }); +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +/** 记录每一帧开始、结束的顺序和同时在途的最大数量 */ +function fakeServer(options: { delay?: (index: number) => number; reject?: (index: number) => 'pause' | 'error' | undefined } = {}) { + const log: string[] = []; let inflight = 0; let peak = 0; + const send: SendFrame = async (f, { onLoaded }) => { + inflight += 1; peak = Math.max(peak, inflight); log.push(`start:${f.index}`); + await new Promise((resolve) => setTimeout(resolve, options.delay?.(f.index) ?? 1)); + onLoaded(f.bytes.byteLength); + inflight -= 1; + const verdict = options.reject?.(f.index); + if (verdict === 'error') throw new Error('boom'); + if (verdict === 'pause') return false; + log.push(`done:${f.index}`); + return true; + }; + return { send, log, peak: () => peak }; +} + +async function run(uploader: FrameUploader, count: number) { + for (let index = 0; index < count; index += 1) await uploader.push(frame(index), index === count - 1); + return uploader.finish(); +} + +describe('FrameUploader', () => { + it('受限文件:第 0 帧先传,中间帧并发,末帧等其余帧都确认后才发', async () => { + const server = fakeServer({ delay: (index) => (index === 2 ? 20 : 2) }); + const uploader = new FrameUploader({ gated: true, concurrency: 3, send: server.send, signal: new AbortController().signal, onBytes: () => {} }); + expect(await run(uploader, 6)).toBe(true); + expect(server.log.indexOf('done:0')).toBeLessThan(server.log.indexOf('start:1')); + expect(server.peak()).toBeGreaterThan(1); + expect(server.peak()).toBeLessThanOrEqual(3); + const start5 = server.log.indexOf('start:5'); + for (const index of [0, 1, 2, 3, 4]) expect(server.log.indexOf(`done:${index}`)).toBeLessThan(start5); + }); + + it('非受限文件:逐帧按顺序上传,不并发', async () => { + const server = fakeServer(); + const uploader = new FrameUploader({ gated: false, concurrency: 4, send: server.send, signal: new AbortController().signal, onBytes: () => {} }); + expect(await run(uploader, 4)).toBe(true); + expect(server.peak()).toBe(1); + expect(server.log).toEqual(['start:0', 'done:0', 'start:1', 'done:1', 'start:2', 'done:2', 'start:3', 'done:3']); + }); + + it('只有一帧时直接上传', async () => { + const server = fakeServer(); + const uploader = new FrameUploader({ gated: true, concurrency: 4, send: server.send, signal: new AbortController().signal, onBytes: () => {} }); + expect(await run(uploader, 1)).toBe(true); + expect(server.log).toEqual(['start:0', 'done:0']); + }); + + it('某帧重试用尽暂停:不再发新帧,finish 返回 false', async () => { + const server = fakeServer({ reject: (index) => (index === 1 ? 'pause' : undefined) }); + const uploader = new FrameUploader({ gated: true, concurrency: 2, send: server.send, signal: new AbortController().signal, onBytes: () => {} }); + expect(await run(uploader, 8)).toBe(false); + expect(server.log).not.toContain('start:7'); + }); + + it('不可重试的错误:push 或 finish 抛出,并且不再发新帧', async () => { + const server = fakeServer({ reject: (index) => (index === 1 ? 'error' : undefined) }); + const uploader = new FrameUploader({ gated: true, concurrency: 2, send: server.send, signal: new AbortController().signal, onBytes: () => {} }); + await expect(run(uploader, 8)).rejects.toThrow('boom'); + expect(server.log).not.toContain('start:7'); + }); + + it('积压超过上限时 push 会等待,不让加密一口气跑完整个文件', async () => { + const server = fakeServer({ delay: () => 15 }); + const uploader = new FrameUploader({ gated: true, concurrency: 1, send: server.send, signal: new AbortController().signal, onBytes: () => {}, buffer: 2 }); + let pushed = 0; + const producer = (async () => { for (let index = 0; index < 6; index += 1) { await uploader.push(frame(index), index === 5); pushed += 1; } })(); + await tick(); await tick(); + expect(pushed).toBeLessThan(6); + await producer; expect(await uploader.finish()).toBe(true); + }); + + it('已取消的信号:push 直接中止', async () => { + const controller = new AbortController(); controller.abort(); + const uploader = new FrameUploader({ gated: true, concurrency: 2, send: fakeServer().send, signal: controller.signal, onBytes: () => {} }); + await expect(uploader.push(frame(0), true)).rejects.toMatchObject({ name: 'AbortError' }); + }); + + it('上报的是已确认加在途的字节数', async () => { + const seen: number[] = []; + const uploader = new FrameUploader({ gated: false, concurrency: 1, send: fakeServer().send, signal: new AbortController().signal, onBytes: (sent) => seen.push(sent) }); + await run(uploader, 3); + expect(Math.max(...seen)).toBe(300); + expect(seen[seen.length - 1]).toBe(300); + }); +}); + +describe('FrameUploader 自适应并发', () => { + it('遇到重试就把并发减半,不再占满连接', async () => { + let retried = false; let droppedTo = 0; + const send: SendFrame = async (f, hooks) => { + await new Promise((resolve) => setTimeout(resolve, 5)); + if (f.index === 1 && !retried) { retried = true; hooks.onRetry(); droppedTo = uploader.currentConcurrency; } + return true; + }; + const uploader: FrameUploader = new FrameUploader({ gated: true, concurrency: 4, send, signal: new AbortController().signal, onBytes: () => {} }); + expect(await run(uploader, 20)).toBe(true); + expect(droppedTo).toBe(2); + }); + + it('连续顺利后并发度慢慢恢复', async () => { + let first = true; + const send: SendFrame = async (f, hooks) => { await new Promise((resolve) => setTimeout(resolve, 1)); if (first && f.index === 1) { first = false; hooks.onRetry(); } return true; }; + const uploader = new FrameUploader({ gated: true, concurrency: 4, send, signal: new AbortController().signal, onBytes: () => {} }); + expect(await run(uploader, 60)).toBe(true); + expect(uploader.currentConcurrency).toBeGreaterThan(2); + }); +}); + +describe('TransferMeter', () => { + it('用最近一段时间估算速度和剩余时间', () => { + const meter = new TransferMeter(3000); + expect(meter.speed()).toBeUndefined(); + meter.add(0, 0); meter.add(1_000_000, 1000); meter.add(2_000_000, 2000); + expect(meter.speed()).toBeCloseTo(1_000_000, -3); + expect(meter.eta(2_000_000, 5_000_000)).toBe(3); + }); + + it('速度只看窗口内的样本,网络变慢后很快反映出来', () => { + const meter = new TransferMeter(3000); + meter.add(0, 0); meter.add(10_000_000, 1000); meter.add(10_100_000, 5000); meter.add(10_200_000, 7000); + expect(meter.speed()!).toBeLessThan(100_000); + }); +}); + +describe('estimateCipherBytes', () => { + it('按帧数乘第一帧密文长度估算', () => { + expect(estimateCipherBytes(0, 1000, 1016)).toBe(1016); + expect(estimateCipherBytes(2500, 1000, 1016)).toBe(3048); + }); +}); diff --git a/web/src/core-bridge/worker/uploadPipeline.ts b/web/src/core-bridge/worker/uploadPipeline.ts new file mode 100644 index 0000000..9966aa1 --- /dev/null +++ b/web/src/core-bridge/worker/uploadPipeline.ts @@ -0,0 +1,156 @@ +import type { QueuedFrame } from '../database'; + +/** + * 单个文件的帧上传调度(protocol §5、服务端 AppendFrame 的约束): + * - 第 0 帧先单独上传(它在服务端创建条目); + * - 受限(gated)文件的中间帧彼此独立,可以并发;非受限文件的帧会按到达顺序进入日志,必须逐帧顺序上传; + * - 末帧带整份内容的签名,服务端要求前面的帧都已入库,所以必须等其余帧全部确认后再发。 + */ +export type SendFrame = (frame: QueuedFrame, hooks: { onLoaded: (loadedBytes: number) => void; onRetry: () => void }, signal: AbortSignal) => Promise; + +interface Options { + /** 受限文件的中间帧可以并发 */ + gated: boolean; + concurrency: number; + send: SendFrame; + signal: AbortSignal; + /** 已确认 + 在途的字节数变化时回调 */ + onBytes: (sentBytes: number) => void; + /** 生产者(加密)领先上传的最大帧数,防止整个文件堆在内存里 */ + buffer?: number; + /** 续传:这些帧上次已经被服务端确认,不必再发 */ + alreadyAccepted?: Iterable; +} + +export class FrameUploader { + private ready: { frame: QueuedFrame; final: boolean }[] = []; + private inflight = 0; + private readonly accepted = new Set(); + private readonly loaded = new Map(); + private acceptedBytes = 0; + private finalIndex: number | undefined; + private paused = false; + private error: unknown; + private failed = false; + private waiters: (() => void)[] = []; + + constructor(private readonly options: Options) { + this.dynamicLimit = options.concurrency; + for (const index of options.alreadyAccepted ?? []) this.accepted.add(index); + } + + /** 已确认的帧数(含续传前已确认的) */ + get acceptedCount(): number { return this.accepted.size; } + + /** 当前并发度:遇到重试就减半,连续顺利再慢慢加回去(弱网下少占连接,好网下跑满) */ + private dynamicLimit: number; + private streak = 0; + get currentConcurrency(): number { return this.limit; } + private get limit(): number { return this.options.gated ? Math.max(1, Math.min(this.options.concurrency, this.dynamicLimit)) : 1; } + private get bufferLimit(): number { return this.options.buffer ?? this.limit * 2 + 2; } + + /** 一帧已加密并写入离线队列。积压太多时等待,让加密不要跑得比上传快太多 */ + async push(frame: QueuedFrame, final: boolean): Promise { + if (this.failed) throw this.error; + if (this.options.signal.aborted) throw new DOMException('上传已取消', 'AbortError'); + if (final) this.finalIndex = frame.index; + // 已暂停:帧留在离线队列里,等网络恢复后由 flushOutbox 续传 + if (this.paused) return; + this.ready.push({ frame, final }); + this.pump(); + while (!final && !this.paused && !this.failed && this.ready.length + this.inflight >= this.bufferLimit) await this.wait(); + if (this.failed) throw this.error; + } + + /** 等所有帧确认。返回 false 表示中途暂停(重试用尽,多半断网);失败会抛出 */ + async finish(): Promise { + while (!this.settled()) await this.wait(); + if (this.failed) throw this.error; + return !this.paused; + } + + private settled(): boolean { + if (this.failed || this.paused) return this.inflight === 0; + return this.finalIndex !== undefined && this.accepted.size === this.finalIndex + 1; + } + + private canStart(frame: QueuedFrame, final: boolean): boolean { + if (final) return this.accepted.size === frame.index && this.inflight === 0; + if (frame.index === 0) return this.inflight === 0; + if (!this.options.gated) return this.accepted.size === frame.index && this.inflight === 0; + return this.accepted.has(0); + } + + private pump(): void { + while (!this.failed && !this.paused && this.inflight < this.limit) { + const at = this.ready.findIndex((entry) => this.canStart(entry.frame, entry.final)); + if (at < 0) break; + const [entry] = this.ready.splice(at, 1); + this.start(entry!.frame); + } + this.notify(); + } + + private start(frame: QueuedFrame): void { + this.inflight += 1; + let retried = false; + this.options.send(frame, { + onLoaded: (loaded) => { this.loaded.set(frame.index, loaded); this.report(); }, + onRetry: () => { retried = true; this.dynamicLimit = Math.max(1, Math.floor(this.dynamicLimit / 2)); this.streak = 0; }, + }, this.options.signal) + .then((accepted) => { + this.loaded.delete(frame.index); + if (accepted) { + this.accepted.add(frame.index); this.acceptedBytes += frame.bytes.byteLength; + if (!retried && ++this.streak >= 8 && this.dynamicLimit < this.options.concurrency) { this.dynamicLimit += 1; this.streak = 0; } + } + else { this.paused = true; this.ready = []; } + }) + .catch((error: unknown) => { + this.loaded.delete(frame.index); + if (!this.failed) { this.failed = true; this.error = error; } + this.ready = []; + }) + .finally(() => { this.inflight -= 1; this.report(); this.pump(); }); + } + + private report(): void { + let sent = this.acceptedBytes; + for (const value of this.loaded.values()) sent += value; + this.options.onBytes(sent); + } + + private wait(): Promise { return new Promise((resolve) => { this.waiters.push(resolve); }); } + private notify(): void { const waiting = this.waiters; this.waiters = []; for (const resume of waiting) resume(); } +} + +/** 最近几秒的滑动平均速度,网络变化后很快跟上,不会被开头的数据拖累 */ +export class TransferMeter { + private samples: { at: number; bytes: number }[] = []; + constructor(private readonly windowMs = 3000) {} + + add(bytes: number, at = performance.now()): void { + this.samples.push({ at, bytes }); + const from = at - this.windowMs; + while (this.samples.length > 2 && this.samples[1]!.at <= from) this.samples.shift(); + } + + /** 字节每秒;样本不足或时间太短时为 undefined */ + speed(): number | undefined { + const first = this.samples[0]; const last = this.samples[this.samples.length - 1]; + if (!first || !last || last.at - first.at < 400) return undefined; + const value = (last.bytes - first.bytes) * 1000 / (last.at - first.at); + return value > 0 ? value : undefined; + } + + /** 剩余秒数 */ + eta(sentBytes: number, totalBytes: number): number | undefined { + const speed = this.speed(); + return speed ? Math.max(0, Math.ceil((totalBytes - sentBytes) / speed)) : undefined; + } +} + +/** 按第一帧的密文长度估算整个文件上传的总字节数;全部加密完成后会用精确值替换 */ +export function estimateCipherBytes(fileSize: number, maxFrameData: number, firstFrameBytes: number): number { + return Math.max(1, Math.ceil(fileSize / maxFrameData)) * firstFrameBytes; +} diff --git a/web/src/core-bridge/worker/uploadQueue.ts b/web/src/core-bridge/worker/uploadQueue.ts new file mode 100644 index 0000000..3a370ab --- /dev/null +++ b/web/src/core-bridge/worker/uploadQueue.ts @@ -0,0 +1,192 @@ +import { previewKindOf } from '@/lib/fileKind'; +import { dequeueFrame, dequeueItem, queueFrame, queueLocation, saveResource, type QueuedFrame, type SavedResource } from '../database'; +import { errorMessage } from '../errors'; +import type { FileItem, FileUploadOptions } from '../types'; +import { prepareObject } from './channelDataKeys'; +import { queueFileCatalog } from './channelThreads'; +import { deviceIdentity, publishWorkspace } from './presentation'; +import { senderSnapshot } from './senderSnapshot'; +import { blobs, getSession } from './sessions'; +import { flushOutbox, markPipelined, MAX_FRAME_DATA, MOBILE_UPLOAD_FRAME_DATA_BYTES, persistPlan, unmarkPipelined, uploadFrame } from './transfer'; +import { FrameStash } from './frameStash'; +import { estimateCipherBytes, FrameUploader, TransferMeter } from './uploadPipeline'; + +/** 受限文件内部同时在途的帧数。再多对单条连接的带宽提升有限,还会占满浏览器的并发连接 */ +const FRAME_CONCURRENCY = 4; +const PROGRESS_INTERVAL_MS = 150; + +interface Job { + resourceId: string; + file: File; + options: FileUploadOptions | undefined; + threadId: string; + folderId: string | undefined; + localId: string | undefined; + controller: AbortController; + /** 到运行时才建立的条目 id,取消时据此清理 */ + itemId?: string; + run: () => Promise; + settle: { resolve: () => void; reject: (error: unknown) => void }; +} +interface Lane { active?: Job; waiting: Job[] } + +/** 每个空间一条上传队列:一次只传一个文件,其余排队;文件内部的帧可以并发(见 FrameUploader) */ +const lanes = new Map(); + +const isAbort = (error: unknown): boolean => (error as Error | undefined)?.name === 'AbortError'; + +export function enqueueFileUploads(resourceId: string, files: File[], options: FileUploadOptions | undefined, + threadId: string, folderId: string | undefined, localIds: string[] | undefined): Promise { + const lane = lanes.get(resourceId) ?? { waiting: [] }; + lanes.set(resourceId, lane); + const done = files.map((file, index) => new Promise((resolve, reject) => { + const job: Job = { resourceId, file, options, threadId, folderId, localId: localIds?.[index], controller: new AbortController(), + run: () => runJob(job), settle: { resolve, reject } }; + lane.waiting.push(job); + })); + drain(resourceId, lane); + return Promise.allSettled(done).then((results) => { + const failed = results.find((result): result is PromiseRejectedResult => result.status === 'rejected'); + if (failed) throw failed.reason; + }); +} + +/** 取消排队中或上传中的文件。返回是否找到了这个任务 */ +export function cancelFileUpload(resourceId: string, localId: string): boolean { + const lane = lanes.get(resourceId); + if (!lane) return false; + const waiting = lane.waiting.findIndex((job) => job.localId === localId); + if (waiting >= 0) { const [job] = lane.waiting.splice(waiting, 1); job!.settle.resolve(); return true; } + if (lane.active?.localId === localId) { lane.active.controller.abort(); return true; } + return false; +} + +function drain(resourceId: string, lane: Lane): void { + if (lane.active) return; + const job = lane.waiting.shift(); + if (!job) { lanes.delete(resourceId); return; } + lane.active = job; + job.run().then(job.settle.resolve, job.settle.reject).finally(() => { lane.active = undefined; drain(resourceId, lane); }); +} + +async function runJob(job: Job): Promise { + const { resourceId, file, threadId, folderId, localId } = job; + const signal = job.controller.signal; + const { session, resource } = await getSession(resourceId); + const options = job.options ?? { directPreview: true, maxDownloads: resource.maxDownloads ?? 0 }; + if (resource.kind !== 'channel' || (!session.can('owner') && !session.can('write'))) throw new Error('此链接没有上传权限'); + const device = await deviceIdentity(session.endpointSecret); + const maxFrameData = device.kind === 'phone' ? MOBILE_UPLOAD_FRAME_DATA_BYTES : MAX_FRAME_DATA; + const author = await senderSnapshot(); + const ts = Date.now(); + let item: FileItem | undefined; + let stash: FrameStash | undefined; + try { + if (signal.aborted) return; + const prepared = await prepareObject(resource, session, threadId, 'files'); + const plan = session.encryptFile(file, {}, { + itemId: prepared.itemIdBytes, + ...(options.directPreview ? { gated: 'auto' as const } : { gated: true, maxDownloads: options.maxDownloads }), + ts, maxFrameData, threadId, folderId, + }); + item = { + id: plan.itemId, kind: 'file', ts, state: { kind: 'receiving', progress: 0, phase: 'encrypting' }, + ...(author ? { author } : {}), + ...(threadId ? { threadId } : {}), ...(folderId ? { folderId } : {}), ...(localId ? { localId } : {}), + name: file.name, mime: file.type || 'application/octet-stream', size: file.size, + remaining: plan.gated + ? (options.directPreview ? resource.maxDownloads ?? 0 : options.maxDownloads) || null + : null, + preview: previewKindOf(file.type, file.name), gated: plan.gated, + }; + job.itemId = item.id; + resource.localItems = [...(resource.localItems ?? []).filter((entry) => entry.id !== item?.id), item]; + await saveResource(resource); + blobs.set(`${resourceId}:${item.id}`, file); + publishWorkspace(resourceId, { t: 'item', item }); + markPipelined(resourceId, item.id); + + const progress = createProgress(resourceId, () => item, (next) => { item = next; }); + // 帧先直接上传,来不及上传的才在后台写进离线队列(见 frameStash);离线队列由 stash 管理,上传成功后不用再去删 + const queued = new FrameStash(queueFrame, dequeueFrame); + stash = queued; + const uploader = new FrameUploader({ + gated: plan.gated, concurrency: FRAME_CONCURRENCY, signal, + send: (frame, hooks, s) => uploadFrame(resource, session, frame, { ...hooks, manageQueue: false, onAccepted: () => queued.accepted(frame) }, s), + onBytes: progress.sent, + }); + const planned = await persistPlan(resource, plan, undefined, + { channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: 'file.publish' }, + async (frame: QueuedFrame, final) => { + if (frame.index === 0) progress.estimate(estimateCipherBytes(file.size, maxFrameData, frame.bytes.byteLength)); + await queued.add(frame); + await uploader.push(frame, final); + }); + progress.exact(planned.bytes); + await queueFileCatalog(resource, session, item); + await queueLocation(resourceId, item.id, threadId, folderId); + const framesUploaded = await uploader.finish(); + // 没传完的(断网暂停)帧要先写进离线队列,flushOutbox 才续传得了 + await queued.drain(); + unmarkPipelined(resourceId, item.id); + // 帧都确认后,再发目录条目和位置登记(以及上次没发完的遗留) + const uploaded = framesUploaded && await flushOutbox(resource, session); + await finish(resource, item, uploaded ? { kind: 'complete' } : { kind: 'incomplete' }); + if (!uploaded) publishWorkspace(resourceId, { t: 'error', message: '上传暂停了。重新打开空间并连接网络后,可以继续发送已加密内容。' }); + } catch (error) { + if (item) unmarkPipelined(resourceId, item.id); + if (signal.aborted || isAbort(error)) { await stash?.discard(); await discard(resource, item); return; } + job.controller.abort(); + // 失败时已加密还没确认的帧要保住,之后才能续传 + await stash?.drain().catch(() => undefined); + if (item) await finish(resource, item, { kind: 'incomplete' }); + publishWorkspace(resourceId, { t: 'error', message: `文件上传失败：${errorMessage(error)}` }); + throw error; + } +} + +async function finish(resource: SavedResource, item: FileItem, state: FileItem['state']): Promise { + const next = { ...item, state }; + resource.localItems = (resource.localItems ?? []).map((entry) => entry.id === next.id ? next : entry); + await saveResource(resource); + publishWorkspace(resource.id, { t: 'item', item: next }); +} + +/** 取消:清掉离线队列里这个文件的帧和本地条目。服务端已收下的帧会随空间保留期自然过期 */ +async function discard(resource: SavedResource, item: FileItem | undefined): Promise { + if (!item) return; + await dequeueItem(resource.id, item.id); + resource.localItems = (resource.localItems ?? []).filter((entry) => entry.id !== item.id); + await saveResource(resource); + blobs.delete(`${resource.id}:${item.id}`); + publishWorkspace(resource.id, { t: 'removed', id: item.id }); +} + +/** + * 把「已发送字节数」换算成一条 0–100% 的进度,并带上速度和剩余时间。 + * 总量先按第一帧估算,全部加密完成后换成精确值;最后一帧确认前封顶 99%。 + */ +function createProgress(resourceId: string, current: () => FileItem | undefined, update: (item: FileItem) => void) { + const meter = new TransferMeter(); + let total = 0; let lastSent = 0; let lastPublished = 0; + const publish = (force: boolean) => { + const item = current(); + if (!item || total <= 0) return; + const now = performance.now(); + if (!force && now - lastPublished < PROGRESS_INTERVAL_MS) return; + lastPublished = now; + const speed = meter.speed(); const eta = meter.eta(lastSent, total); + const next: FileItem = { ...item, state: { + kind: 'receiving', progress: Math.max(0, Math.min(0.99, lastSent / total)), phase: lastSent > 0 ? 'uploading' : 'encrypting', + sentBytes: Math.min(lastSent, total), totalBytes: total, + ...(speed === undefined ? {} : { speedBytesPerSecond: speed }), ...(eta === undefined ? {} : { etaSeconds: eta }), + } }; + update(next); + publishWorkspace(resourceId, { t: 'item', item: next }); + }; + return { + estimate: (bytes: number) => { total = bytes; publish(true); }, + exact: (bytes: number) => { total = bytes; publish(true); }, + sent: (bytes: number) => { lastSent = bytes; meter.add(bytes); publish(false); }, + }; +} diff --git a/web/src/core-bridge/worker/workspaceStructure.ts b/web/src/core-bridge/worker/workspaceStructure.ts index 94992ab..09f5bf3 100644 --- a/web/src/core-bridge/worker/workspaceStructure.ts +++ b/web/src/core-bridge/worker/workspaceStructure.ts @@ -1,4 +1,4 @@ -import type { WorkspaceBody, WorkspaceChannelCreatedBody, WorkspaceChannelRequest, WorkspaceGroupCreatedBody, WorkspaceMoveRequest, WorkspaceNameRequest } from '../generated/api'; +import type { WorkspaceBody, WorkspaceChannelCreatedBody, WorkspaceChannelRequest, WorkspaceGroupCreatedBody, WorkspaceMoveRequest, WorkspaceNameRequest, WorkspaceRoleCreatedBody, WorkspaceRoleRequest } from '../generated/api'; import type { FileItem } from '../types'; import { getResource, cacheResource } from '../database'; import { getSession } from './sessions'; @@ -40,7 +40,7 @@ export async function loadWorkspace(id: string): Promise { export async function changeWorkspace(id: string, path: string, method: 'POST' | 'PATCH' | 'PUT' | 'DELETE', value?: unknown): Promise { const { session } = await getSession(id); const body = await apiFetch(`/spaces/${id}/${path}`, { - method, headers: { ...session.authHeaders(session.can('owner') ? 'owner' : 'write'), 'Content-Type': 'application/json' }, body: JSON.stringify(value), + method, headers: { ...session.authHeaders(session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read'), 'Content-Type': 'application/json' }, body: JSON.stringify(value), }).then(r => r.json() as Promise); await publish(id, body); return body; @@ -59,6 +59,13 @@ export async function createWorkspaceGroup(id: string, value: WorkspaceNameReque await publish(id, body.workspace); return body.group.id; } +export async function createWorkspaceRole(id: string, value: WorkspaceRoleRequest): Promise { + const { session } = await getSession(id); + const body = await apiFetch(`/spaces/${id}/roles`, { method: 'POST', headers: { ...session.authHeaders('owner'), 'Content-Type': 'application/json' }, body: JSON.stringify(value) }) + .then(r => r.json() as Promise); + await publish(id, body.workspace); + return body.role.id; +} export async function registerItemLocation(id: string, item: Pick): Promise { if (!item.threadId) return; await changeWorkspace(id, 'locations', 'PUT', { itemIds: [item.id], threadId: item.threadId, folderId: item.folderId } satisfies WorkspaceMoveRequest); diff --git a/web/src/features/channel/ChannelSettingsSheet.module.css b/web/src/features/channel/ChannelSettingsSheet.module.css index ee763c2..bea0c21 100644 --- a/web/src/features/channel/ChannelSettingsSheet.module.css +++ b/web/src/features/channel/ChannelSettingsSheet.module.css @@ -22,6 +22,11 @@ height: min(360px, 70dvh); } +/* 身份组页内容多:加高,列表在滚动时固定在左边 */ +.tall { + height: min(560px, 80dvh); +} + .tabs { display: flex; flex-direction: column; diff --git a/web/src/features/channel/ChannelSettingsSheet.tsx b/web/src/features/channel/ChannelSettingsSheet.tsx index 54c831b..ac3e62f 100644 --- a/web/src/features/channel/ChannelSettingsSheet.tsx +++ b/web/src/features/channel/ChannelSettingsSheet.tsx @@ -3,17 +3,19 @@ import type { ChannelInfo, ChannelSettingsPatch } from '@/core-bridge/types'; import { Button } from '@/ui/Button'; import { Icon, type IconName } from '@/ui/Icon'; import { Sheet } from '@/ui/Sheet'; +import { RolesEditor } from './RolesEditor'; import styles from './ChannelSettingsSheet.module.css'; import { useServerSettings } from '@/store/server'; import type { ServerInfoBody } from '@/core-bridge/generated/api'; import { ttlLabel, withCurrent } from '@/lib/settingOptions'; -type Tab = 'basic' | 'content' | 'join'; +type Tab = 'basic' | 'content' | 'join' | 'roles'; const TABS: { id: Tab; label: string; icon: IconName }[] = [ { id: 'basic', label: '基本', icon: 'edit' }, { id: 'content', label: '内容', icon: 'history' }, { id: 'join', label: '加入', icon: 'users' }, + { id: 'roles', label: '身份组', icon: 'users' }, ]; interface Props { @@ -66,9 +68,9 @@ function SettingsForm({ channel, onClose, onSave, info }: Props & { info: Server return ( -
+
- {TABS.map((item) => )}
@@ -105,8 +107,9 @@ function SettingsForm({ channel, onClose, onSave, info }: Props & { info: Server setClearPassword(event.target.checked)} />移除加入密码 : null}

空间主及受邀用户无需加入密码。修改规则不会移除已加入成员。

: null} + {tab === 'roles' ? : null} - {owner ? ( + {tab === 'roles' ? null : owner ? (
{failed ?

保存失败,请检查名称和设置后重试。

: null} diff --git a/web/src/features/channel/ChatMessage.tsx b/web/src/features/channel/ChatMessage.tsx index b8526ee..db073e4 100644 --- a/web/src/features/channel/ChatMessage.tsx +++ b/web/src/features/channel/ChatMessage.tsx @@ -4,10 +4,12 @@ import { useAccount } from '@/store/account'; import { useDownload } from '@/features/item/download'; import { iconOf } from '@/features/item/summary'; import { senderLabel } from '@/lib/sender'; -import { formatSize, formatTime, formatTransferProgress } from '@/lib/format'; +import { formatSize, formatTime } from '@/lib/format'; +import { TransferProgress } from '@/features/item/TransferProgress'; import { Icon } from '@/ui/Icon'; import { Button } from '@/ui/Button'; import { ContextMenu, type MenuAction } from '@/ui/ContextMenu'; +import { NameWithBadges } from './RoleBadges'; import styles from './ChatPane.module.css'; interface Props { id: string; compact?: boolean; onOpenFile: (id: string) => void; onDelete: (id: string) => void } @@ -16,6 +18,7 @@ export function ChatMessage({ id, compact = false, onOpenFile, onDelete }: Props const item = useWorkspace((s) => s.byId[id]); const owner = useWorkspace((s) => s.channel?.role === 'owner'); const retry = useWorkspace((s) => s.retrySend); + const cancelUpload = useWorkspace((s) => s.cancelUpload); const username = useAccount((s) => s.username); const download = useDownload(); const [downloading, setDownloading] = useState(false); @@ -49,7 +52,7 @@ export function ChatMessage({ id, compact = false, onOpenFile, onDelete }: Props : sender?.avatar ? {`${name} : {Array.from(name)[0] ?? '?'}}
- {!compact ?
{name}
: null} + {!compact ?
{name}
: null} {item.state.kind === 'tampered' ?

内容校验失败，已拒绝显示。

: item.kind === 'text' ?

{item.text}

:
} - {item.state.kind === 'receiving' && item.kind === 'file' ?
{formatTransferProgress(item.state)}
+ {item.state.kind === 'receiving' && item.kind === 'file' ? { void cancelUpload(item.localId!); } } : {})} /> : failed ? 发送失败。 : item.state.kind === 'incomplete' ? 发送未完成，等待续传。 : null}
diff --git a/web/src/features/channel/MembersPane.tsx b/web/src/features/channel/MembersPane.tsx index e49a894..2e3cb46 100644 --- a/web/src/features/channel/MembersPane.tsx +++ b/web/src/features/channel/MembersPane.tsx @@ -5,10 +5,13 @@ import { errorMessage } from '@/core-bridge/errors'; import type { SenderInfo } from '@/core-bridge/types'; import { useToast } from '@/store/toast'; import { useWorkspace } from '@/store/workspace'; +import { permissionsOf } from '@/lib/roles'; +import { useAccount } from '@/store/account'; import { Button } from '@/ui/Button'; +import { NameWithBadges } from './RoleBadges'; import styles from './MembersPane.module.css'; -interface Props { owner: boolean; refreshKey: number; onInvite: () => void } +interface Props { owner: boolean; /** 空间主,或所在身份组被授予「移除成员」 */ canRemove: boolean; refreshKey: number; onInvite: () => void } function Avatar({ person, name }: { person?: SenderInfo; name: string }) { return person?.avatar ? @@ -22,10 +25,12 @@ function Section({ title, count, children }: { title: string; count: number; chi /** 右栏「成员」页:以用户为单位,按在线 / 离线 / 待接受分组。设备只是用户的属性(在线设备数由后端按账号统计) */ const PRESENCE_POLL_MS = 8000; -export function MembersPane({ owner, refreshKey, onInvite }: Props) { +export function MembersPane({ owner, canRemove, refreshKey, onInvite }: Props) { const channelId = useWorkspace((s) => s.channel?.id); const maxMembers = useWorkspace((s) => s.channel?.maxMembers); const people = useWorkspace((s) => s.people); + const roles = useWorkspace((s) => s.roles); + const self = useAccount((s) => s.username); const listInvitations = useWorkspace((s) => s.listInvitations); const revoke = useWorkspace((s) => s.revokeInvitation); const [members, setMembers] = useState([]); @@ -37,10 +42,10 @@ export function MembersPane({ owner, refreshKey, onInvite }: Props) { if (!channelId) return; try { setMembers(await channelAccessApi.members(channelId)); - setPending(owner ? await listInvitations() : []); + setPending(canRemove ? await listInvitations() : []); setError(''); } catch { setError('成员清单读取失败'); } - }, [channelId, owner, listInvitations]); + }, [channelId, canRemove, listInvitations]); useEffect(() => { void refresh(); // 在线状态不走订阅流,成员页可见期间定时读取 @@ -66,15 +71,17 @@ export function MembersPane({ owner, refreshKey, onInvite }: Props) { const name = person?.displayName || username; return
  • - {name}{isOwner ? 空间主 : null} + {name} {detail || name !== username ? {[name !== username ? username : '', detail].filter(Boolean).join(' · ')} : null} {isOnline ? : null} {removable ? : null}
    • ; }; + // 被授权的成员不能移除自己、空间主,也不能移除同样有移除权限的人 + const removable = (item: ChannelMember) => item.role !== 'owner' && (owner || (canRemove && item.username !== self && !permissionsOf(roles, item.username, false).has('members.remove'))); const member = (item: ChannelMember) => row(item.username, item.devices > 1 || item.online_devices > 1 ? `${item.online_devices}/${item.devices} 台设备在线` : undefined, - false, owner && item.role !== 'owner', item.role === 'owner', item.online); + false, removable(item), item.role === 'owner', item.online); return
    @@ -86,7 +93,7 @@ export function MembersPane({ owner, refreshKey, onInvite }: Props) {
    {offline.map(member)}
    {pending.map((invitation) => row(invitation.username, undefined, true, owner))}
    {error ?

    {error}

    : null} - {!owner ?

    只有空间主可以邀请或移除成员。

    : null} + {!owner ?

    {canRemove ? '你可以移除成员；邀请成员只有空间主可以。' : '只有空间主可以邀请或移除成员。'}

    : null}
    ; } diff --git a/web/src/features/channel/RoleBadges.module.css b/web/src/features/channel/RoleBadges.module.css new file mode 100644 index 0000000..44bf379 --- /dev/null +++ b/web/src/features/channel/RoleBadges.module.css @@ -0,0 +1,42 @@ +.badge { + --c: var(--text-dim); + + display: inline-block; + margin-inline: var(--space-1); + padding: 0 var(--space-1); + border-radius: var(--radius-sm); + background: var(--accent-soft); + color: var(--c); + font-size: 11px; + font-weight: 500; + line-height: 16px; + vertical-align: 1px; +} + +.owner { + --c: var(--text); +} + +.gray { + --c: var(--text-dim); +} + +.blue { + --c: var(--busy); +} + +.green { + --c: var(--ok); +} + +.amber { + --c: var(--warn); +} + +.red { + --c: var(--danger); +} + +.purple { + --c: var(--role-purple); +} diff --git a/web/src/features/channel/RoleBadges.tsx b/web/src/features/channel/RoleBadges.tsx new file mode 100644 index 0000000..35a2cb2 --- /dev/null +++ b/web/src/features/channel/RoleBadges.tsx @@ -0,0 +1,20 @@ +import { useMemo } from 'react'; +import { badgesOf, type Badge } from '@/lib/roles'; +import { useWorkspace } from '@/store/workspace'; +import styles from './RoleBadges.module.css'; + +function List({ items }: { items: Badge[] }) { + return <>{items.map((badge) => {badge.text})}; +} + +/** 名字前后的身份标签:空间主固定一个,其余是身份组里自定义的前缀、后缀 */ +export function useBadges(username: string | undefined): { prefix: Badge[]; suffix: Badge[] } { + const roles = useWorkspace((s) => s.roles); + const creator = useWorkspace((s) => s.channel?.creatorAccount); + return useMemo(() => badgesOf(roles, username, creator), [roles, username, creator]); +} + +export function NameWithBadges({ username, children }: { username: string | undefined; children: React.ReactNode }) { + const { prefix, suffix } = useBadges(username); + return <>{children}; +} diff --git a/web/src/features/channel/RolesEditor.module.css b/web/src/features/channel/RolesEditor.module.css new file mode 100644 index 0000000..943b999 --- /dev/null +++ b/web/src/features/channel/RolesEditor.module.css @@ -0,0 +1,207 @@ +.layout { + display: grid; + grid-template-columns: 148px minmax(0, 1fr); + gap: var(--space-4); +} + +.list { + display: flex; + flex-direction: column; + position: sticky; + top: 0; + align-self: start; + gap: var(--space-1); + margin: 0; + padding: 0; + list-style: none; +} + +.list button { + display: flex; + align-items: center; + gap: var(--space-2); + width: 100%; + min-height: 36px; + padding: 0 var(--space-2); + border: 0; + border-radius: var(--radius-md); + background: transparent; + color: var(--text-dim); + font: inherit; + font-size: 13px; + text-align: left; + cursor: pointer; +} + +.list button:hover, +.list button[aria-current='true'] { + background: var(--accent-soft); + color: var(--text); +} + +.list span { + flex: 1; + min-width: 0; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.list small { + color: var(--text-dim); + font-size: 11px; +} + +.form { + display: flex; + flex-direction: column; + gap: var(--space-3); + min-width: 0; +} + +.field { + display: flex; + flex-direction: column; + gap: var(--space-1); + color: var(--text-dim); + font-size: 12px; +} + +.field input { + min-width: 0; + min-height: 36px; + padding: 0 var(--space-2); + border: 1px solid var(--line); + border-radius: var(--radius-md); + background: var(--surface-2); + color: var(--text); + font: inherit; + font-size: 14px; +} + +.field b { + color: var(--text); + font-weight: 600; +} + +.pair { + display: grid; + grid-template-columns: 1fr 1fr; + gap: var(--space-2); +} + +.colors { + display: flex; + flex-wrap: wrap; + gap: var(--space-1); +} + +.swatch { + --c: var(--text-dim); + + min-width: 36px; + min-height: 28px; + border: 1px solid var(--line-strong); + border-radius: var(--radius-sm); + background: var(--accent-soft); + color: var(--c); + font: inherit; + font-size: 12px; + cursor: pointer; +} + +.swatch[aria-checked='true'] { + border-color: var(--c); + font-weight: 600; +} + +.gray { + --c: var(--text-dim); +} + +.blue { + --c: var(--busy); +} + +.green { + --c: var(--ok); +} + +.amber { + --c: var(--warn); +} + +.red { + --c: var(--danger); +} + +.purple { + --c: var(--role-purple); +} + +.group { + display: flex; + flex-direction: column; + gap: var(--space-1); + margin: 0; + padding: var(--space-2) var(--space-3); + border: 1px solid var(--line); + border-radius: var(--radius-md); +} + +.group legend { + padding: 0 var(--space-1); + color: var(--text-dim); + font-size: 12px; +} + +.check { + display: flex; + align-items: flex-start; + gap: var(--space-2); + min-height: 32px; + padding: var(--space-1) 0; + font-size: 14px; +} + +.check input { + margin-top: 3px; + accent-color: var(--accent); +} + +.check small { + display: block; + color: var(--text-dim); + font-size: 11px; + line-height: 1.5; +} + +.empty { + margin: 0; + color: var(--text-dim); + font-size: 12px; +} + +.actions { + display: flex; + justify-content: flex-end; + gap: var(--space-2); + padding-top: var(--space-2); + border-top: 1px solid var(--line); +} + +@media (width <= 720px) { + .layout { + grid-template-columns: minmax(0, 1fr); + } + + .list { + flex-direction: row; + overflow-x: auto; + } + + .list button { + width: auto; + white-space: nowrap; + } +} diff --git a/web/src/features/channel/RolesEditor.tsx b/web/src/features/channel/RolesEditor.tsx new file mode 100644 index 0000000..cb6e804 --- /dev/null +++ b/web/src/features/channel/RolesEditor.tsx @@ -0,0 +1,106 @@ +import { useEffect, useState } from 'react'; +import { channelAccessApi, type ChannelMember } from '@/core-bridge/channelAccess'; +import { errorMessage } from '@/core-bridge/errors'; +import type { RoleInfo } from '@/core-bridge/types'; +import { PERMISSIONS, ROLE_COLORS } from '@/lib/roles'; +import { useToast } from '@/store/toast'; +import { useWorkspace } from '@/store/workspace'; +import { Button } from '@/ui/Button'; +import { Icon } from '@/ui/Icon'; +import styles from './RolesEditor.module.css'; + +const COLOR_LABEL = { gray: '灰', blue: '蓝', green: '绿', amber: '黄', red: '红', purple: '紫' } as const; + +function RoleForm({ role, members, onDone }: { role?: RoleInfo; members: ChannelMember[]; onDone: (id?: string) => void }) { + const people = useWorkspace((s) => s.people); + const createRole = useWorkspace((s) => s.createRole); + const updateRole = useWorkspace((s) => s.updateRole); + const deleteRole = useWorkspace((s) => s.deleteRole); + const setRoleMembers = useWorkspace((s) => s.setRoleMembers); + const [name, setName] = useState(role?.name ?? ''); + const [prefix, setPrefix] = useState(role?.prefix ?? ''); + const [suffix, setSuffix] = useState(role?.suffix ?? ''); + const [color, setColor] = useState(role?.color ?? 'gray'); + const [permissions, setPermissions] = useState>(new Set(role?.permissions)); + const [chosen, setChosen] = useState>(new Set(role?.members)); + const [busy, setBusy] = useState(false); + const [confirmDelete, setConfirmDelete] = useState(false); + const toggle = (set: ReadonlySet, id: string) => { const next = new Set(set); if (next.has(id)) next.delete(id); else next.add(id); return next; }; + + async function run(task: () => Promise) { + setBusy(true); + try { onDone(await task()); } + catch (cause) { useToast.getState().push({ kind: 'error', text: `操作失败：${errorMessage(cause)}` }); } + finally { setBusy(false); } + } + const save = () => run(async () => { + const request = { name: name.trim(), prefix: prefix.trim() || null, suffix: suffix.trim() || null, color, permissions: [...permissions] }; + let id = role?.id; + if (id) await updateRole(id, request); else id = await createRole(request); + await setRoleMembers(id, [...chosen]); + useToast.getState().push({ kind: 'ok', text: '身份组已保存' }); + return id; + }); + const remove = () => run(async () => { if (role) await deleteRole(role.id); return undefined; }); + const sample = `${prefix.trim()}${prefix.trim() ? ' ' : ''}你的名字${suffix.trim() ? ' ' : ''}${suffix.trim()}`; + + return
    { event.preventDefault(); if (name.trim()) void save(); }}> + +
    + + +
    +
    颜色 · 聊天里显示为 {sample} +
    + {ROLE_COLORS.map((item) => )} +
    +
    +
    权限 + {PERMISSIONS.map((item) => )} +
    +
    成员 · {chosen.size} + {members.length ? members.map((member) => ) + :

    还没有其他成员。成员接受邀请后可以加入身份组。

    } +
    +
    + {role ? (confirmDelete + ? <> + : ) : null} + +
    +
    ; +} + +/** 空间设置里的「身份组」页:左边是身份组列表,右边是编辑表单 */ +export function RolesEditor() { + const channelId = useWorkspace((s) => s.channel?.id); + const roles = useWorkspace((s) => s.roles); + const [members, setMembers] = useState([]); + const [selected, setSelected] = useState(''); + useEffect(() => { + if (!channelId) return; + let active = true; + channelAccessApi.members(channelId).then((rows) => { if (active) setMembers(rows.filter((row) => row.role !== 'owner')); }).catch(() => {}); + return () => { active = false; }; + }, [channelId]); + const role = roles.find((item) => item.id === selected); + const first = role ?? (selected === '' ? roles[0] : undefined); + const creating = !first; + + return
    +
      + {roles.map((item) =>
      • )} +
      • +
    + setSelected(id ?? '')} /> +
    ; +} diff --git a/web/src/features/channel/ThreadSidebar.tsx b/web/src/features/channel/ThreadSidebar.tsx index 1b3186e..6512503 100644 --- a/web/src/features/channel/ThreadSidebar.tsx +++ b/web/src/features/channel/ThreadSidebar.tsx @@ -9,7 +9,10 @@ interface Props { threads: ThreadInfo[]; groups: ChannelGroupInfo[]; selectedThreadId?: string; + /** 能重命名频道、管理分组(空间主,或被授予「管理频道和分组」) */ enabled: boolean; + /** 新建频道涉及密钥,只有空间主 */ + canCreate: boolean; owner: boolean; onHistory: (id: string, chatHistory: boolean, filesHistory: boolean) => Promise; onSelectThread: (id: string) => void; @@ -72,7 +75,7 @@ export function ThreadSidebar(props: Props) { - {props.enabled ? + {props.canCreate ?
    @@ -85,7 +88,7 @@ export function ThreadSidebar(props: Props) {
    {props.enabled ? -
    {!collapsed.has(group.id) ? void }) const activeFolderId = activeFolders.some((folder) => folder.id === folderId) ? folderId : undefined; const enabled = Boolean(activeThreadId && channel?.canUpload); const owner = channel?.role === 'owner'; + const roles = useWorkspace((s) => s.roles); + const self = useAccount((s) => s.username); + const perms = permissionsOf(roles, self ?? undefined, owner); const files = useWorkspace(useShallow((s) => s.order.filter((id) => { const item = s.byId[id]; return item?.kind === 'file' && !item.metadataHidden && item.threadId === activeThreadId && (item.folderId ?? undefined) === activeFolderId && matchesFilter(item, filter); @@ -126,7 +131,7 @@ export function Workspace({ onCreateChannel }: { onCreateChannel: () => void }) const channelId = channel?.id; useEffect(() => { setDrawerOpen(false); }, [channelId]); const rail = setScanOpen(true)} />; - const threadSidebar = { selectThread(id); setDrawerOpen(false); setMobileView('chat'); }} onCreateThread={createThread} onRenameThread={renameThread} onCreateGroup={createGroup} onRenameGroup={renameGroup} onDeleteGroup={deleteGroup} onGroupThread={groupThread} />; const navigation = <>{threadSidebar}{desktop ? : null}; @@ -143,7 +148,7 @@ export function Workspace({ onCreateChannel }: { onCreateChannel: () => void }) onDownloadItems={downloadItems} onDeleteItems={deleteFiles} onMoveItems={async (ids, target) => { if (activeThreadId) { await moveItems(ids, activeThreadId, target); setChecked(new Set()); setSelectedId(undefined); } }} /> {restricted.length ?
    等待同步文件目录{restricted.map((id) => )}
    : null} ; - const members = setInviteOpen(true)} />; + const members = setInviteOpen(true)} />; function goMobile(view: 'chat' | 'files' | 'members') { setMobileView(view); setDrawerOpen(false); } const moreRows: { icon: IconName; label: string; run: () => void }[] = [ { icon: 'settings', label: '空间设置', run: () => setSettingsOpen(true) }, diff --git a/web/src/features/item/TransferProgress.module.css b/web/src/features/item/TransferProgress.module.css new file mode 100644 index 0000000..7cbad43 --- /dev/null +++ b/web/src/features/item/TransferProgress.module.css @@ -0,0 +1,45 @@ +.transfer { + display: flex; + flex-direction: column; + gap: var(--space-1); + width: 360px; + max-width: 100%; + margin-top: var(--space-1); + color: var(--text-dim); + font-size: 11px; +} + +.line { + display: flex; + align-items: baseline; + gap: var(--space-2); +} + +.line small { + margin-inline-start: auto; + font-size: 11px; +} + +.cancel { + padding: 0; + border: 0; + background: none; + color: var(--text-dim); + font: inherit; + text-decoration: underline; + cursor: pointer; +} + +.line small + .cancel { + margin-inline-start: 0; +} + +.line > span:first-child + .cancel { + margin-inline-start: auto; +} + +.transfer progress { + width: 100%; + height: var(--space-1); + accent-color: var(--busy); +} diff --git a/web/src/features/item/TransferProgress.tsx b/web/src/features/item/TransferProgress.tsx new file mode 100644 index 0000000..c7e6a52 --- /dev/null +++ b/web/src/features/item/TransferProgress.tsx @@ -0,0 +1,22 @@ +import { formatSize, formatTransferProgress, type TransferState } from '@/lib/format'; +import styles from './TransferProgress.module.css'; + +interface Props { + state: TransferState & { sentBytes?: number; totalBytes?: number }; + /** 排队中或上传中的文件可以取消 */ + onCancel?: () => void; +} + +/** 文件上传的进度:一行状态文字,一条进度条,可取消 */ +export function TransferProgress({ state, onCancel }: Props) { + const queued = state.phase === 'queued'; + const detail = state.sentBytes !== undefined && state.totalBytes ? `${formatSize(state.sentBytes)} / ${formatSize(state.totalBytes)}` : ''; + return
    +
    + {formatTransferProgress(state)} + {detail && !queued ? {detail} : null} + {onCancel ? : null} +
    + {queued ? null : } +
    ; +} diff --git a/web/src/lib/format.ts b/web/src/lib/format.ts index bc41da8..991a1b4 100644 --- a/web/src/lib/format.ts +++ b/web/src/lib/format.ts @@ -10,16 +10,29 @@ export function formatSize(bytes: number): string { return `${v >= 100 ? v.toFixed(0) : v.toFixed(1)} ${units[i]}`; } -export function formatTransferProgress(state: { +export interface TransferState { progress: number; - phase?: 'encrypting' | 'uploading'; + phase?: 'queued' | 'encrypting' | 'uploading'; speedBytesPerSecond?: number; -}): string { - const phase = state.phase === 'encrypting' ? '加密中' : state.phase === 'uploading' ? '上传中' : '传输中'; - const speed = state.speedBytesPerSecond && state.speedBytesPerSecond > 0 - ? ` · ${formatSize(state.speedBytesPerSecond)}/s` - : ''; - return `${phase} ${Math.round(state.progress * 100)}%${speed}`; + etaSeconds?: number; +} + +export function formatEta(seconds: number): string { + if (seconds < 60) return `剩 ${Math.max(1, Math.round(seconds))} 秒`; + if (seconds < 3600) return `剩 ${Math.ceil(seconds / 60)} 分钟`; + const hours = Math.floor(seconds / 3600); + return `剩 ${hours} 小时 ${Math.ceil((seconds % 3600) / 60)} 分钟`; +} + +/** 传输状态的一行文字:排队中 / 加密中 / 上传中 42% · 3.2 MB/s · 剩 12 秒 */ +export function formatTransferProgress(state: TransferState): string { + if (state.phase === 'queued') return '排队中'; + const percent = Math.round(state.progress * 100); + if (state.phase === 'encrypting' && percent === 0) return '加密中…'; + const parts = [`${state.phase === 'encrypting' ? '加密中' : '上传中'} ${percent}%`]; + if (state.speedBytesPerSecond && state.speedBytesPerSecond > 0) parts.push(`${formatSize(state.speedBytesPerSecond)}/s`); + if (state.etaSeconds !== undefined) parts.push(formatEta(state.etaSeconds)); + return parts.join(' · '); } const timeFmt = new Intl.DateTimeFormat('zh-CN', { hour: '2-digit', minute: '2-digit', hour12: false }); diff --git a/web/src/lib/roles.ts b/web/src/lib/roles.ts new file mode 100644 index 0000000..79b1d48 --- /dev/null +++ b/web/src/lib/roles.ts @@ -0,0 +1,35 @@ +import type { RoleInfo } from '@/core-bridge/types'; + +export const ROLE_COLORS = ['gray', 'blue', 'green', 'amber', 'red', 'purple'] as const; +export type RoleColor = (typeof ROLE_COLORS)[number]; + +/** 与后端 WorkspacePermission 一致 */ +export const PERMISSIONS = [ + { id: 'members.remove', label: '移除成员', hint: '移除成员、撤销邀请;不能移除空间主和同样有此权限的人' }, + { id: 'structure.manage', label: '管理频道和分组', hint: '重命名频道、调整频道所属分组、新建和删除分组。新建频道涉及密钥,仍只有空间主' }, +] as const; + +export interface Badge { key: string; text: string; color: RoleColor | 'owner' } + +const hasText = (value: string | null | undefined): value is string => Boolean(value && value.trim()); + +/** 聊天里名字前后要显示的文字:空间主固定一个,其余按身份组的创建顺序 */ +export function badgesOf(roles: readonly RoleInfo[], username: string | undefined, creator: string | undefined): { prefix: Badge[]; suffix: Badge[] } { + const prefix: Badge[] = []; const suffix: Badge[] = []; + if (!username) return { prefix, suffix }; + if (username === creator) suffix.push({ key: 'owner', text: '空间主', color: 'owner' }); + for (const role of roles) { + if (!role.members.includes(username)) continue; + const color = (ROLE_COLORS as readonly string[]).includes(role.color ?? '') ? (role.color as RoleColor) : 'gray'; + if (hasText(role.prefix)) prefix.push({ key: role.id + ':p', text: role.prefix, color }); + if (hasText(role.suffix)) suffix.push({ key: role.id + ':s', text: role.suffix, color }); + } + return { prefix, suffix }; +} + +export function permissionsOf(roles: readonly RoleInfo[], username: string | undefined, owner: boolean): ReadonlySet { + if (owner) return new Set(PERMISSIONS.map((item) => item.id)); + const granted = new Set(); + if (username) for (const role of roles) if (role.members.includes(username)) for (const id of role.permissions) granted.add(id); + return granted; +} diff --git a/web/src/store/reduce.test.ts b/web/src/store/reduce.test.ts index 0a6364a..fbc506e 100644 --- a/web/src/store/reduce.test.ts +++ b/web/src/store/reduce.test.ts @@ -8,7 +8,7 @@ const text = (id: string, ts: number, body = id): TextItem => ({ describe('reduce', () => { it('renders server structure and preserves server author and location across late decrypted content', () => { - const stateEvent = { t: 'structure' as const, state: { revision: 1, + const stateEvent = { t: 'structure' as const, state: { revision: 1, roles: [], rolesSeeded: true, channels: [{ id: 'channel-a', name: '常规', groupId: 'group-a', createdAt: 1, createdBy: 'alice', chatEpoch: 1, filesEpoch: 1, chatHistory: true, filesHistory: true, fileBytes: 0 }], groups: [{ id: 'group-a', name: '项目', createdAt: 1 }], folders: [{ id: 'folder-a', name: '目录', threadId: 'channel-a' }], diff --git a/web/src/store/reduce.ts b/web/src/store/reduce.ts index fe0a064..59ab35d 100644 --- a/web/src/store/reduce.ts +++ b/web/src/store/reduce.ts @@ -1,6 +1,7 @@ import type { ChannelInfo, ChannelGroupInfo, + RoleInfo, ConnectionState, FolderInfo, Item, @@ -16,6 +17,7 @@ export interface WorkspaceData { channel: ChannelInfo | null; threads: ThreadInfo[]; groups: ChannelGroupInfo[]; + roles: RoleInfo[]; folders: FolderInfo[]; locations: Record; byId: Record; @@ -34,6 +36,7 @@ export const initialData: WorkspaceData = { channel: null, threads: [], groups: [], + roles: [], folders: [], locations: {}, byId: {}, @@ -71,7 +74,7 @@ export function reduce(s: WorkspaceData, e: WorkspaceEvent): WorkspaceData { ...item, ...(locations[id] ?? {}), ...(authors[id] ? { author: authors[id] } : {}), }])); const profiles = Object.fromEntries(e.people.map(person => [person.username, { username: person.username, displayName: person.displayName || person.username, ...(person.avatar ? { avatar: person.avatar } : {}) }])); - return { ...s, people: profiles, threads: e.state.channels.map(c => ({ ...c, groupId: c.groupId ?? undefined })), groups: e.state.groups, + return { ...s, people: profiles, threads: e.state.channels.map(c => ({ ...c, groupId: c.groupId ?? undefined })), groups: e.state.groups, roles: e.state.roles ?? [], folders: e.state.folders.map(f => ({ ...f, parentId: f.parentId ?? undefined })), locations, authors, byId }; } case 'reset': diff --git a/web/src/store/workspace.ts b/web/src/store/workspace.ts index 0af4d87..4e250b0 100644 --- a/web/src/store/workspace.ts +++ b/web/src/store/workspace.ts @@ -44,7 +44,8 @@ function settleLocal(id: string): void { async function deliver(id: string): Promise { const job = pendingJobs.get(id); if (!job) return; - useWorkspace.setState((s) => s.byId[id] ? { byId: { ...s.byId, [id]: { ...s.byId[id], state: { kind: 'receiving', progress: 0 } } } } : s); + const phase = job.kind === 'file' ? 'queued' as const : undefined; + useWorkspace.setState((s) => s.byId[id] ? { byId: { ...s.byId, [id]: { ...s.byId[id], state: { kind: 'receiving', progress: 0, ...(phase ? { phase } : {}) } } } } : s); try { if (job.kind === 'text') await api.sendText(job.text, job.threadId, id); else await api.sendFiles([job.file], job.options, job.threadId, job.folderId, [id]); @@ -61,12 +62,18 @@ interface Actions { sendText: (text: string, threadId?: string) => Promise; /** 重试发送失败的本地占位项 */ retrySend: (localId: string) => Promise; + /** 取消排队中或上传中的文件 */ + cancelUpload: (localId: string) => Promise; updateChannelSettings: (patch: ChannelSettingsPatch) => Promise; sendFiles: (files: File[], options?: FileUploadOptions, threadId?: string, folderId?: string) => Promise; createThread: (name: string, groupId?: string) => Promise; createGroup: typeof api.createGroup; renameGroup: typeof api.renameGroup; deleteGroup: typeof api.deleteGroup; + createRole: typeof api.createRole; + updateRole: typeof api.updateRole; + deleteRole: typeof api.deleteRole; + setRoleMembers: typeof api.setRoleMembers; groupThread: typeof api.groupThread; renameThread: (threadId: string, name: string) => Promise; setChannelHistory: typeof api.setChannelHistory; @@ -98,22 +105,32 @@ export const useWorkspace = create()(() => ({ return deliver(id); }, retrySend: (id) => deliver(id), + cancelUpload: async (localId) => { + pendingJobs.delete(localId); + useWorkspace.setState((s) => withoutItem(s, localId)); + await api.cancelUpload(localId).catch(() => false); + }, updateChannelSettings: (patch) => api.updateChannelSettings(patch), sendFiles: async (files, options, threadId, folderId) => { const ids = files.map((file) => { const id = newLocalId(); pendingJobs.set(id, { kind: 'file', file, ...(options ? { options } : {}), ...(threadId ? { threadId } : {}), ...(folderId ? { folderId } : {}) }); const item: FileItem = { id, kind: 'file', localId: id, name: file.name, mime: file.type || 'application/octet-stream', size: file.size, remaining: null, - preview: previewKindOf(file.type, file.name), ...localBase(threadId, folderId), state: { kind: 'receiving', progress: 0, phase: 'encrypting' } }; + preview: previewKindOf(file.type, file.name), ...localBase(threadId, folderId), state: { kind: 'receiving', progress: 0, phase: 'queued' } }; putLocal(item); return id; }); - for (const id of ids) await deliver(id); + // 全部交给 worker 的上传队列:一次传一个,其余显示「排队中」 + await Promise.all(ids.map(deliver)); }, createThread: (name, groupId) => api.createThread(name, groupId), createGroup: (name) => api.createGroup(name), renameGroup: (id, name) => api.renameGroup(id, name), deleteGroup: (id) => api.deleteGroup(id), + createRole: (role) => api.createRole(role), + updateRole: (id, role) => api.updateRole(id, role), + deleteRole: (id) => api.deleteRole(id), + setRoleMembers: (id, members) => api.setRoleMembers(id, members), groupThread: (id, groupId) => api.groupThread(id, groupId), renameThread: (threadId, name) => api.renameThread(threadId, name), setChannelHistory: (threadId, chatHistory, filesHistory) => api.setChannelHistory(threadId, chatHistory, filesHistory), diff --git a/web/src/styles/tokens.css b/web/src/styles/tokens.css index 69d82e6..50c8f15 100644 --- a/web/src/styles/tokens.css +++ b/web/src/styles/tokens.css @@ -14,6 +14,7 @@ --warn: #b25e00; --danger: #b3261e; --busy: #3b6ea8; + --role-purple: #7b4fb3; --surface-2: #efefed; --line-strong: rgb(0 0 0 / 14%); @@ -76,6 +77,7 @@ --warn: #e0a050; --danger: #f2877f; --busy: #7fb0e6; + --role-purple: #b79af0; --surface-2: #262626; --line-strong: rgb(255 255 255 / 16%); @@ -104,6 +106,7 @@ --warn: #e0a050; --danger: #f2877f; --busy: #7fb0e6; + --role-purple: #b79af0; --surface-2: #262626; --line-strong: rgb(255 255 255 / 16%); diff --git a/web/vite.config.ts b/web/vite.config.ts index 65f3abe..ee6fff2 100644 --- a/web/vite.config.ts +++ b/web/vite.config.ts @@ -1,4 +1,5 @@ import { execFileSync } from 'node:child_process'; +import http from 'node:http'; import { readFileSync } from 'node:fs'; import { fileURLToPath } from 'node:url'; import react from '@vitejs/plugin-react'; @@ -34,6 +35,7 @@ export default defineConfig({ }, }, build: { target: 'es2022', sourcemap: false }, - server: { proxy: { '/api/v1': { target: process.env.WEBDROP_DEV_API_TARGET ?? 'http://127.0.0.1:5000', ws: true } } }, + // 开发代理复用到后端的连接:默认每个请求新开一条 TCP,上传 1 MiB 的帧会被连接建立和慢启动拖得很慢 + server: { proxy: { '/api/v1': { target: process.env.WEBDROP_DEV_API_TARGET ?? 'http://127.0.0.1:5000', ws: true, agent: new http.Agent({ keepAlive: true, maxSockets: 64 }) } } }, test: { environment: 'node' }, });