feat: integrate client and server sharing workflows

This commit is contained in:
justme committed 2026-10-02 10:49:45 +08:00
1 parent 42b93ae2d8
commit 7406e2fe37
101 files changed
+3153 -253

No files matched your search

+131
View File
@@ -0,0 +1,131 @@
import { parseLogRecords, toBase64Url, type ChannelSession, type LogRecord, type ReceivedItem } from '@webdrop/core';
import { getResource, saveResource, type SavedResource } from '../database';
import type { FileItem, Item, Snapshot } from '../types';
import { previewKindOf } from '@/lib/fileKind';
import { blobs, readers } from './sessions';
import { showRejected, deviceIdentity, deviceInfo, publishWorkspace } from './presentation';
import { type ChannelWire, toChannelInfo } from './wire';
import { apiFetch, readJson } from './http';
export function mapReceived(item: ReceivedItem): Item {
const timestamp = item.meta?.ts ?? Date.now();
const deviceId = item.meta?.dev ?? '未知设备';
const state = item.complete ? { kind: 'complete' as const } : { kind: 'incomplete' as const };
if (item.kind === 'file' || item.kind === 'raw') {
const name = item.meta?.name ?? '加密文件';
return {
id: item.itemId, kind: 'file', ts: timestamp, deviceId, state, name,
mime: item.meta?.mime ?? 'application/octet-stream', size: item.meta?.size ?? item.content.byteLength,
remaining: null, preview: previewKindOf(item.meta?.mime ?? '', name),
gated: item.gated,
};
}
return { id: item.itemId, kind: 'text', ts: timestamp, deviceId, state, text: new TextDecoder().decode(item.content) };
}
export async function applyChannelRecord(resource: SavedResource, session: ChannelSession, input: Uint8Array | LogRecord): Promise<void> {
let reader = readers.get(resource.id);
if (!reader) {
reader = session.createItemReader();
readers.set(resource.id, reader);
}
const events = await reader.push(input);
for (const event of events) {
switch (event.type) {
case 'item': {
const item = mapReceived(event.item);
if (event.item.snapshot) {
let snapshot: Snapshot = { text: '', updatedBy: item.deviceId, updatedAt: item.ts };
if (item.kind === 'text') {
try {
const value = JSON.parse(item.text) as Partial<Snapshot>;
if (typeof value.text === 'string') snapshot = { text: value.text, updatedBy: value.updatedBy ?? item.deviceId, updatedAt: value.updatedAt ?? item.ts };
else snapshot.text = item.text;
} catch { snapshot.text = item.text; }
}
publishWorkspace(resource.id, { t: 'snapshot', snapshot });
} else {
const blob = item.kind === 'file' && event.item.complete ? new Blob([event.item.content], { type: item.mime }) : undefined;
if (blob) blobs.set(`${resource.id}:${item.id}`, blob);
publishWorkspace(resource.id, { t: 'item', item });
}
break;
}
case 'announce': {
const local = (await getResource(resource.id))?.localItems?.find((item) => item.id === event.itemId);
const placeholder: FileItem = local?.kind === 'file' ? local : {
id: event.itemId, kind: 'file', ts: event.createdAt, deviceId: '未知设备', state: { kind: 'incomplete' },
name: '受限下载内容', mime: 'application/octet-stream', size: 0, remaining: null, preview: 'none',
gated: true,
};
publishWorkspace(resource.id, { t: 'item', item: placeholder });
break;
}
case 'removed':
blobs.delete(`${resource.id}:${event.itemId}`);
publishWorkspace(resource.id, { t: 'removed', id: event.itemId });
break;
case 'rejected':
showRejected(resource.id, event);
break;
default: {
const exhaustive: never = event;
throw new Error(`未知的加密事件 ${String(exhaustive)}`);
}
}
}
}
export async function channelMetadata(resource: SavedResource, session: ChannelSession): Promise<void> {
const info = await readJson<ChannelWire>(`/channels/${resource.id}`, session.authHeaders('read'));
session.verifyChannelInfo(info);
resource.membersCanUpload = info.settings.members_can_upload;
resource.expiresAt = info.expires_at;
const latest = await getResource(resource.id);
await saveResource({ ...latest, ...resource, localItems: latest?.localItems });
publishWorkspace(resource.id, { t: 'channel', info: toChannelInfo(info, resource) });
const ownId = toBase64Url(session.endpointHash);
let devices = [deviceInfo(ownId, Date.now(), deviceIdentity().name, ownId)];
if (session.can('owner')) {
const seats = await readJson<Array<{ endpoint_hash: string; joined_at: number; label?: string }>>(
`/channels/${resource.id}/seats`, session.authHeaders('owner'));
devices = await Promise.all(seats.map(async (seat) => {
let name: string | undefined;
if (seat.label) {
try { name = await session.openLabel(seat.label, seat.endpoint_hash); }
catch { name = '设备名校验失败'; }
}
return deviceInfo(seat.endpoint_hash, seat.joined_at, name, ownId);
}));
const own = devices.find((device) => device.self);
if (own) own.name = deviceIdentity().name;
}
publishWorkspace(resource.id, { t: 'devices', devices });
}
export async function syncChannel(resource: SavedResource, session: ChannelSession, signal: AbortSignal): Promise<number> {
publishWorkspace(resource.id, { t: 'reset' });
readers.set(resource.id, session.createItemReader());
let from = 0;
for (;;) {
const response = await apiFetch(`/channels/${resource.id}/log?from=${from}&limit=200`, {
headers: session.authHeaders('read'), signal,
});
const records = parseLogRecords(new Uint8Array(await response.arrayBuffer()));
for (const record of records) {
await applyChannelRecord(resource, session, record);
from = Math.max(from, record.seq + 1);
}
if (records.length < 200) return from;
}
}
export async function processChannelBytes(resource: SavedResource, session: ChannelSession, bytes: Uint8Array): Promise<number> {
const records = parseLogRecords(bytes);
let next = 0;
for (const record of records) {
await applyChannelRecord(resource, session, record);
next = Math.max(next, record.seq + 1);
}
return next;
}
@@ -0,0 +1,62 @@
import { previewKindOf } from '@/lib/fileKind';
import type { FileItem } from '../types';
import { apiFetch } from './http';
import { blobs, getSession } from './sessions';
import { stopChannel, connectChannel } from './channelTransport';
import { downloadDecrypted } from './transfer';
import { publishWorkspace } from './presentation';
import { IntegrityError } from '@webdrop/core';
export function connectChannelResource(id: string): () => void {
void connectChannel(id);
return () => stopChannel(id);
}
export async function removeChannelItem(resourceId: string, itemId: string): Promise<void> {
const { session } = await getSession(resourceId);
const role = session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read';
await apiFetch(`/channels/${resourceId}/items/${itemId}`, {
method: 'DELETE', headers: session.authHeaders(role),
});
}
export async function getChannelBlob(resourceId: string, itemId: string, versionId?: string): Promise<Blob> {
if (versionId) throw new Error('此版本暂不可下载');
const cached = blobs.get(`${resourceId}:${itemId}`);
if (cached) return cached;
const { session, resource } = await getSession(resourceId);
if (resource.kind !== 'channel') throw new Error('资源类型不匹配');
let result: Awaited<ReturnType<typeof downloadDecrypted>>;
try { result = await downloadDecrypted(session, resource, itemId); }
catch (error) {
if (error instanceof IntegrityError) publishWorkspace(resourceId, { t: 'item', item: { id: itemId, kind: 'text', ts: Date.now(), deviceId: '', text: '', state: { kind: 'tampered' } } });
throw error;
}
blobs.set(`${resourceId}:${itemId}`, result.blob);
if (result.meta?.kind === 'file') {
const item: FileItem = {
id: itemId, kind: 'file', ts: result.meta.ts, deviceId: result.meta.dev ?? '未知设备', state: { kind: 'complete' },
name: result.meta.name ?? '加密文件', mime: result.meta.mime ?? result.blob.type, size: result.blob.size,
remaining: null, preview: 'none', gated: true,
};
item.preview = previewKindOf(item.mime, item.name);
publishWorkspace(resourceId, { t: 'item', item });
}
return result.blob;
}
export async function resolveTextConflict(): Promise<void> {
throw new Error('此条目没有可合并的冲突版本');
}
export async function markChannelSeen(): Promise<void> {
return undefined;
}
export async function listArchive(_id: string): Promise<never[]> {
throw new Error('服务端没有提供压缩包目录接口,可以直接下载。');
}
@@ -0,0 +1,95 @@
import { ChannelSession } from '@webdrop/core';
import { getResource, listResources, removeResource, saveResource, type SavedResource } from '../database';
import type { ParsedChannelLink } from '@/lib/channelLink';
import type { ChannelInfo, CreateOptions } from '../types';
import { apiFetch, readJson } from './http';
import { sessions } from './sessions';
import { deviceIdentity } from './presentation';
import { makeLink } from './links';
import { toChannelInfo, type ChannelWire } from './wire';
import { stopChannel } from './channelTransport';
export async function createChannel(options: CreateOptions, captchaToken?: string): Promise<ChannelInfo> {
const endpointSecret = crypto.getRandomValues(new Uint8Array(32));
const { session, request } = await ChannelSession.create({
mode: options.mode,
endpointSecret,
settings: {
seats: options.seats,
ttl_seconds: options.ttlSeconds,
members_can_upload: options.membersCanUpload,
delete_policy: 'own',
max_items: 1000,
file_max_downloads: 10,
ui_restrict_copy_download: false,
},
});
try {
const info = await apiFetch('/channels', {
method: 'POST', headers: { ...request.headers, 'Content-Type': 'application/json', ...(captchaToken ? { 'X-WD-Cap': captchaToken } : {}) }, body: JSON.stringify(request.body),
}).then((response) => response.json() as Promise<ChannelWire>);
session.verifyChannelInfo(info);
const resource: SavedResource = {
id: session.id,
kind: 'channel', mode: session.mode, role: 'owner',
link: session.link(self.location.origin), ownerLink: session.link(self.location.origin, 'owner'),
endpointSecret: Uint8Array.from(session.endpointSecret), createdAt: info.created_at, expiresAt: info.expires_at,
seats: info.settings.seats, membersCanUpload: info.settings.members_can_upload,
maxDownloads: info.settings.file_max_downloads, localItems: [],
};
await saveResource(resource);
sessions.set(session.id, session);
return toChannelInfo(info, resource);
} catch (error) {
session.dispose();
throw error;
}
}
export async function joinChannel(input: { link: ParsedChannelLink }): Promise<ChannelInfo> {
const { id, keys } = input.link;
const existing = await getResource(id);
const endpointSecret = existing?.endpointSecret ?? crypto.getRandomValues(new Uint8Array(32));
const session = await ChannelSession.open(makeLink(self.location.origin, id, keys), { endpointSecret });
try {
const info = await readJson<ChannelWire>(`/channels/${id}`, session.authHeaders('read'));
session.verifyChannelInfo(info);
await apiFetch(`/channels/${id}/seats`, { method: 'POST', headers: session.authHeaders(session.can('owner') ? 'owner' : 'read') });
const resource: SavedResource = {
id, kind: 'channel', mode: session.mode, role: session.role,
link: session.link(self.location.origin), ownerLink: session.can('owner') ? session.link(self.location.origin, 'owner') : undefined,
endpointSecret: Uint8Array.from(session.endpointSecret), createdAt: info.created_at, expiresAt: info.expires_at,
seats: info.settings.seats, membersCanUpload: info.settings.members_can_upload,
maxDownloads: info.settings.file_max_downloads, localItems: existing?.localItems ?? [],
};
const label = await session.sealLabel(deviceIdentity().name);
await apiFetch(`/channels/${id}/seats/me`, {
method: 'PUT', headers: { ...session.authHeaders('read'), 'Content-Type': 'application/octet-stream' }, body: label,
});
await saveResource(resource);
sessions.set(id, session);
return toChannelInfo(info, resource);
} catch (error) {
session.dispose();
throw error;
}
}
export async function listChannels(): Promise<ChannelInfo[]> {
const saved = await listResources('channel');
return saved.map((resource) => ({
id: resource.id, mode: resource.mode, role: resource.role, seats: resource.seats,
memberLink: resource.link, ownerLink: resource.ownerLink, createdAt: resource.createdAt, expiresAt: resource.expiresAt,
}));
}
export async function forgetChannel(id: string): Promise<void> {
stopChannel(id);
sessions.get(id)?.dispose();
sessions.delete(id);
await removeResource(id);
}
+71
View File
@@ -0,0 +1,71 @@
import { removeChannelItem } from './channelItems';
import { saveResource } from '../database';
import { previewKindOf } from '@/lib/fileKind';
import type { FileItem, Item } from '../types';
import { blobs, getSession } from './sessions';
import { deviceIdentity } from './presentation';
import { flushOutbox, persistPlan } from './transfer';
import { publishWorkspace } from './presentation';
export async function sendText(resourceId: string, text: string): Promise<{ itemId: string; uploaded: boolean }> {
const { session, resource } = await getSession(resourceId);
if (resource.kind !== 'channel' || (!session.can('owner') && (!session.can('write') || !resource.membersCanUpload))) throw new Error('此链接没有上传权限');
const device = deviceIdentity();
session.deviceName = device.name;
const ts = Date.now();
const plan = session.encryptText(text, { dev: device.name, ts });
const item: Item = { id: plan.itemId, kind: 'text', ts, deviceId: device.name, state: { kind: 'receiving', progress: 0 }, text };
blobs.set(`${resourceId}:${item.id}`, new Blob([text], { type: 'text/plain' }));
publishWorkspace(resourceId, { t: 'item', item });
await persistPlan(resource, plan);
const uploaded = await flushOutbox(resource, session);
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 };
}
export async function sendFiles(resourceId: string, files: File[]): Promise<void> {
const { session, resource } = await getSession(resourceId);
if (resource.kind !== 'channel' || (!session.can('owner') && (!session.can('write') || !resource.membersCanUpload))) throw new Error('此链接没有上传权限');
const device = deviceIdentity();
session.deviceName = device.name;
for (const file of files) {
const ts = Date.now();
const plan = session.encryptFile(file, {}, { maxDownloads: resource.maxDownloads ?? 10, dev: device.name, ts });
let item: FileItem = {
id: plan.itemId, kind: 'file', ts, deviceId: device.name, state: { kind: 'receiving', progress: 0 },
name: file.name, mime: file.type || 'application/octet-stream', size: file.size,
remaining: resource.maxDownloads || null, preview: previewKindOf(file.type, file.name), gated: true,
};
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 });
await persistPlan(resource, plan);
const uploaded = await flushOutbox(resource, session);
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 });
}
}
export async function updateSnapshot(resourceId: string, text: string): Promise<void> {
const { session, resource } = await getSession(resourceId);
const plan = session.encryptJson({ text, updatedBy: deviceIdentity().name, updatedAt: Date.now() }, { snapshot: true, dev: deviceIdentity().name });
await persistPlan(resource, plan);
await flushOutbox(resource, session);
}
export async function editChannelText(resourceId: string, itemId: string, text: string): Promise<void> {
const created = await sendText(resourceId, text);
if (!created.uploaded) throw new Error('新版本已排队,网络恢复前保留旧内容。');
await removeChannelItem(resourceId, itemId);
}
@@ -0,0 +1,80 @@
import { IntegrityError } from '@webdrop/core';
import { active, getSession } from './sessions';
import { apiFetch, type ApiError } from './http';
import { publishWorkspace } from './presentation';
import { channelMetadata, processChannelBytes, syncChannel } from './channelEvents';
import { flushOutbox } from './transfer';
import { socketStream, sseStream, delay, type StreamConsumer } from './transports';
import { errorMessage } from '../errors';
export async function connectChannel(id: string): Promise<void> {
active.get(id)?.abort();
const controller = new AbortController();
active.set(id, controller);
const signal = controller.signal;
let flushTimer: ReturnType<typeof setInterval> | undefined;
try {
const { session, resource } = await getSession(id);
if (signal.aborted) return;
if (resource.kind !== 'channel') throw new Error('资源类型不匹配');
const stream: StreamConsumer = {
cursor: 0,
async record(bytes) { stream.cursor = Math.max(stream.cursor, await processChannelBytes(resource, session, bytes)); },
async control(value) { if (value.t === 'settings' || value.t === 'seats') await channelMetadata(resource, session); },
online() { if (!signal.aborted) publishWorkspace(id, { t: 'connection', state: 'online' }); },
};
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);
while (!signal.aborted) {
try {
if (!initialized) {
stream.cursor = await syncChannel(resource, session, signal);
if (signal.aborted) break;
await channelMetadata(resource, session);
await flushOutbox(resource, session);
initialized = true;
publishWorkspace(id, { t: 'ready' });
}
stream.online();
if (transport === 0) await socketStream(id, session, signal, stream);
else if (transport === 1) await sseStream(id, session, signal, stream);
else {
const response = await apiFetch(`/channels/${id}/log?from=${stream.cursor}&limit=200&wait=25`, { headers: session.authHeaders('read'), signal });
await stream.record(new Uint8Array(await response.arrayBuffer()));
}
backoff = 1000;
if (transport < 2) transport++;
} catch (error) {
if (signal.aborted) break;
const code = (error as ApiError).status;
if (code === 410) { initialized = false; continue; }
if (code === 401 || code === 403 || code === 404 || error instanceof IntegrityError) throw error;
if (transport < 2 && initialized) { transport++; continue; }
publishWorkspace(id, { t: 'connection', state: 'offline' });
publishWorkspace(id, { t: 'ready' });
await delay(backoff + Math.random() * backoff / 3, signal);
backoff = Math.min(30_000, backoff * 2);
}
}
} catch (error) {
if (!signal.aborted) {
publishWorkspace(id, { t: 'error', message: error instanceof IntegrityError ? '内容校验失败,已停止接收。' : errorMessage(error) });
publishWorkspace(id, { t: 'connection', state: 'offline' });
publishWorkspace(id, { t: 'ready' });
}
} finally {
clearInterval(flushTimer);
if (active.get(id) === controller) active.delete(id);
}
}
export function stopChannel(id: string): void {
active.get(id)?.abort();
active.delete(id);
}
+32
View File
@@ -0,0 +1,32 @@
import { afterEach, describe, expect, it, vi } from 'vitest';
import { apiFetch, setApiBase } from './http';
describe('API contract preflight', () => {
afterEach(() => {
setApiBase('/api/v1');
vi.unstubAllGlobals();
});
it('blocks API requests when the server has a newer breaking major', async () => {
const fetchMock = vi.fn(async () => new Response(JSON.stringify({ contract_version: '2.0.0' }), { status: 200 }));
vi.stubGlobal('fetch', fetchMock);
setApiBase('http://api.test/api/v1');
await expect(apiFetch('/channels')).rejects.toThrow('api_contract_mismatch');
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(fetchMock).toHaveBeenCalledWith('http://api.test/api/v1/info', expect.any(Object));
});
it('checks the contract once and then sends the API request', async () => {
const fetchMock = vi.fn()
.mockResolvedValueOnce(new Response(JSON.stringify({ contract_version: '1.2.0' }), { status: 200 }))
.mockResolvedValueOnce(new Response('{}', { status: 200 }))
.mockResolvedValueOnce(new Response('{}', { status: 200 }));
vi.stubGlobal('fetch', fetchMock);
setApiBase('http://api.test/api/v1');
await expect(apiFetch('/channels')).resolves.toBeInstanceOf(Response);
await expect(apiFetch('/shares')).resolves.toBeInstanceOf(Response);
expect(fetchMock).toHaveBeenCalledTimes(3);
});
});
+58
View File
@@ -0,0 +1,58 @@
import { supportsApiContract } from '../contract';
export type ApiError = Error & { status?: number; code?: string };
let apiBase = '/api/v1';
let contractCheck: Promise<void> | undefined;
export function setApiBase(value: string): void {
if (apiBase === value) return;
apiBase = value;
contractCheck = undefined;
}
export const fail = (status: number, code: string): ApiError => Object.assign(new Error(code), { status, code });
export const apiPath = (path: string): string => `${apiBase.replace(/\/$/, '')}${path}`;
function ensureApiContract(): Promise<void> {
contractCheck ??= fetch(apiPath('/info'), {
cache: 'no-store',
credentials: 'omit',
referrerPolicy: 'no-referrer',
}).then(async (response) => {
if (!response.ok) throw new Error('api_info_unavailable');
const info = await response.json() as { contract_version?: unknown };
if (!supportsApiContract(info.contract_version)) throw new Error('api_contract_mismatch');
}).catch((error: unknown) => {
contractCheck = undefined;
throw error;
});
return contractCheck;
}
export async function apiFetch(path: string, init: RequestInit = {}): Promise<Response> {
try {
if (path !== '/info') await ensureApiContract();
const response = await fetch(apiPath(path), {
...init,
cache: 'no-store',
credentials: 'omit',
referrerPolicy: 'no-referrer',
});
if (!response.ok) {
const body = await response.json().catch(() => null) as { error?: unknown } | null;
const code = typeof body?.error === 'string' && /^[a-z_]{1,40}$/.test(body.error) ? body.error : 'request_failed';
throw fail(response.status, code);
}
return response;
} catch (error) {
if (init.signal?.aborted) throw new DOMException('请求已取消', 'AbortError');
if (error instanceof Error && 'status' in error) throw error;
if (error instanceof Error && error.message === 'api_contract_mismatch') throw error;
throw new Error('网络连接失败');
}
}
export async function readJson<T>(path: string, headers: Record<string, string>): Promise<T> {
return apiFetch(path, { headers }).then((response) => response.json() as Promise<T>);
}
+17
View File
@@ -0,0 +1,17 @@
export function toCoreLink(link: string): string {
const parsed = new URL(link);
if (parsed.pathname.startsWith('/s/')) parsed.pathname = parsed.pathname.replace(/^\/s\//, '/c/');
return parsed.href;
}
export function toShareLink(link: string): string {
const parsed = new URL(link);
parsed.pathname = parsed.pathname.replace(/^\/c\//, '/s/');
return parsed.href;
}
export function makeLink(origin: string, id: string, keys: Record<string, string | undefined>): string {
const fragment = Object.entries(keys).filter((entry): entry is [string, string] => Boolean(entry[1]))
.map(([key, value]) => `${key}=${value}`).join('&');
return `${origin.replace(/\/$/, '')}/c/${id}${fragment ? `#${fragment}` : ''}`;
}
+14
View File
@@ -0,0 +1,14 @@
import type { ShareEvent, WorkspaceEvent } from '../types';
type BridgeEvent = WorkspaceEvent | ShareEvent;
type EventSink = (resourceId: string, event: BridgeEvent, blob?: Blob) => void;
let sink: EventSink = () => {};
export function setEventSink(next: EventSink): void {
sink = next;
}
export function publish(resourceId: string, event: BridgeEvent, blob?: Blob): void {
sink(resourceId, event, blob);
}
@@ -0,0 +1,30 @@
import type { ItemEvent } from '@webdrop/core';
import type { Device, FileItem, ShareEvent, WorkspaceEvent } from '../types';
import { previewKindOf } from '@/lib/fileKind';
import { publish } from './port';
export function deviceIdentity(): { name: string; kind: Device['kind'] } {
const mobile = /Android|iPhone|iPad|Mobile/i.test(navigator.userAgent);
return { name: mobile ? '本机 · 移动设备' : '本机 · 浏览器', kind: mobile ? 'phone' : 'desktop' };
}
export function deviceInfo(id: string, joinedAt: number, label: string | undefined, selfId: string): Device {
return { id, name: label ?? '未知设备', kind: 'desktop', online: id === selfId, self: id === selfId, lastSeen: joinedAt };
}
export function showRejected(resourceId: string, event: Extract<ItemEvent, { type: 'rejected' }>, share = false): void {
if (share) publish(resourceId, { t: 'integrity', message: '分享清单校验失败,内容已拒绝显示。' });
else if (event.itemId) publish(resourceId, {
t: 'item', item: { id: event.itemId, kind: 'text', ts: Date.now(), deviceId: '未知设备', state: { kind: 'tampered' }, text: '' },
});
else publish(resourceId, { t: 'error', message: '内容校验失败,已拒绝该记录。' });
}
export function makeFileItem(id: string, name: string, mime: string, size: number, ts: number, state: FileItem['state'], remaining: number | null = null): FileItem {
return { id, kind: 'file', ts, deviceId: '创建者', state, name, mime, size, remaining, preview: previewKindOf(mime, name) };
}
export function publishWorkspace(resourceId: string, event: WorkspaceEvent): void {
publish(resourceId, event);
}
export function publishShare(resourceId: string, event: ShareEvent): void {
publish(resourceId, event);
}
+9
View File
@@ -0,0 +1,9 @@
const running = new Map<string, Promise<unknown>>();
export function serial<T>(key: string, action: () => Promise<T>): Promise<T> {
const previous = running.get(key) ?? Promise.resolve();
const next = previous.catch(() => undefined).then(action);
running.set(key, next);
void next.finally(() => { if (running.get(key) === next) running.delete(key); }).catch(() => undefined);
return next;
}
+25
View File
@@ -0,0 +1,25 @@
import { ChannelSession } from '@webdrop/core';
import { getResource, type SavedResource } from '../database';
import { fail } from './http';
import { toCoreLink } from './links';
export const sessions = new Map<string, ChannelSession>();
export const readers = new Map<string, ReturnType<ChannelSession['createItemReader']>>();
export const active = new Map<string, AbortController>();
export const blobs = new Map<string, Blob>();
const opening = new Map<string, Promise<ChannelSession>>();
export async function getSession(id: string): Promise<{ session: ChannelSession; resource: SavedResource }> {
const resource = await getResource(id);
if (!resource) throw fail(404, 'not_found');
let session = sessions.get(id);
if (!session) {
const sourceLink = resource.ownerLink ?? resource.link;
let task = opening.get(id);
if (!task) {
task = ChannelSession.open(toCoreLink(sourceLink), { endpointSecret: resource.endpointSecret });
opening.set(id, task);
}
try { session = await task; } finally { opening.delete(id); }
sessions.set(id, session);
}
return { session, resource };
}
+99
View File
@@ -0,0 +1,99 @@
import { dequeueItem, removeDraftFile, saveDraftFile, saveResource, type ManifestEntry } from '../database';
import type { FileItem, TextItem } from '../types';
import { makeFileItem } from './presentation';
import { getSession } from './sessions';
import { apiFetch, fail } from './http';
import { flushOutbox, persistPlan } from './transfer';
import { publishShare } from './presentation';
import { invalidateManifest } from './shareState';
import { blobs } from './sessions';
function newTextEntry(text: string): { item: TextItem; entry: Extract<ManifestEntry, { t: 'text' }> } {
const ts = Date.now();
const localId = `draft-text-${crypto.randomUUID()}`;
return {
item: { id: localId, kind: 'text', ts, deviceId: '创建者', state: { kind: 'complete' }, text },
entry: { t: 'text', localId, text, ts },
};
}
export async function addShareText(id: string, text: string): Promise<void> {
const { session, resource } = await getSession(id);
if (resource.kind !== 'share' || !session.can('owner') || resource.state !== 'draft') throw new Error('此分享已封存');
await invalidateManifest(session, resource);
const { item, entry } = newTextEntry(text);
resource.draftItems = [...(resource.draftItems ?? []), item];
resource.manifestEntries = [...(resource.manifestEntries ?? []), entry];
await saveResource(resource);
publishShare(id, { t: 'item', item });
}
export async function addShareFiles(id: string, files: File[]): Promise<void> {
const { session, resource } = await getSession(id);
if (resource.kind !== 'share' || !session.can('owner') || resource.state !== 'draft') throw new Error('此分享已封存');
await invalidateManifest(session, resource);
const downloadLimit = resource.maxDownloads ?? 0;
for (const file of files) {
const ts = Date.now();
const plan = session.encryptFile(file, { name: file.name, mime: file.type || 'application/octet-stream', size: file.size }, {
gated: true, maxDownloads: downloadLimit, ts,
});
const item = makeFileItem(plan.itemId, file.name, file.type || 'application/octet-stream', file.size, ts, { kind: 'receiving', progress: 0 }, downloadLimit || null);
const entry: Extract<ManifestEntry, { t: 'file' }> = { t: 'file', id: plan.itemId, name: file.name, mime: file.type || 'application/octet-stream', size: file.size, ts };
blobs.set(`${id}:${item.id}`, file);
try { await saveDraftFile(id, item.id, file); }
catch { publishShare(id, { t: 'error', message: '本机存储空间不足,刷新后可能无法预览此文件。' }); }
resource.draftItems = [...(resource.draftItems ?? []), item];
resource.manifestEntries = [...(resource.manifestEntries ?? []), entry];
await saveResource(resource);
publishShare(id, { t: 'item', item });
await persistPlan(resource, plan);
const uploaded = await flushOutbox(resource, session);
const complete: FileItem = { ...item, state: uploaded ? { kind: 'complete' } : { kind: 'incomplete' } };
resource.draftItems = resource.draftItems.map((draft) => draft.id === item.id ? complete : draft);
await saveResource(resource);
publishShare(id, { t: 'item', item: complete });
if (!uploaded) {
publishShare(id, { t: 'error', message: '上传暂停了。重新打开草稿并连接网络后,可以继续发送已加密帧。' });
continue;
}
}
}
export async function editShareText(id: string, itemId: string, text: string): Promise<void> {
const { session, resource } = await getSession(id);
if (resource.kind !== 'share' || !session.can('owner') || resource.state !== 'draft') throw new Error('此分享已封存');
const item = resource.draftItems?.find((entry): entry is TextItem => entry.id === itemId && entry.kind === 'text');
if (!item) throw fail(404, 'not_found');
await invalidateManifest(session, resource);
const ts = Date.now();
const updated: TextItem = { ...item, text, ts };
resource.draftItems = (resource.draftItems ?? []).map((entry) => entry.id === itemId ? updated : entry);
resource.manifestEntries = (resource.manifestEntries ?? []).map((entry) =>
entry.t === 'text' && entry.localId === itemId ? { ...entry, text, ts } : entry);
await saveResource(resource);
publishShare(id, { t: 'item', item: updated });
}
export async function removeShareItem(id: string, itemId: string): Promise<void> {
const { session, resource } = await getSession(id);
if (resource.kind !== 'share' || !session.can('owner') || resource.state !== 'draft') throw new Error('此分享已封存');
await invalidateManifest(session, resource);
const item = resource.draftItems?.find((entry) => entry.id === itemId);
if (!item) throw fail(404, 'not_found');
if (item.kind === 'file') {
try { await apiFetch(`/shares/${id}/items/${itemId}`, { method: 'DELETE', headers: session.authHeaders('owner') }); }
catch (error) { if ((error as { status?: number }).status !== 404) throw error; }
await dequeueItem(id, itemId);
await removeDraftFile(id, itemId);
blobs.delete(`${id}:${itemId}`);
}
resource.draftItems = (resource.draftItems ?? []).filter((entry) => entry.id !== itemId);
resource.manifestEntries = (resource.manifestEntries ?? []).filter((entry) => entry.t === 'file' ? entry.id !== itemId : entry.localId !== itemId);
await saveResource(resource);
publishShare(id, { t: 'removed', id: itemId });
}
@@ -0,0 +1,70 @@
import { IntegrityError, type ChannelSession } from '@webdrop/core';
import type { ManifestEntry } from '../database';
import type { ShareWire } from './wire';
import { apiFetch } from './http';
export type ManifestFile = { t: 'file'; id: string; name?: string; mime?: string; size: number; ts: number };
export type ManifestText = { t: 'text'; name?: string; text: string; ts: number };
export type ProtocolManifestEntry = ManifestFile | ManifestText;
export interface ShareManifest { v: 1; items: ProtocolManifestEntry[] }
export function encodeManifest(entries: ManifestEntry[]): Uint8Array {
const items: ProtocolManifestEntry[] = entries.map((entry) => entry.t === 'file'
? { t: 'file', id: entry.id, name: entry.name, mime: entry.mime, size: entry.size, ts: entry.ts }
: { t: 'text', ...(entry.name === undefined ? {} : { name: entry.name }), text: entry.text, ts: entry.ts });
return new TextEncoder().encode(JSON.stringify({ v: 1, items }));
}
function safeTime(value: unknown): value is number {
return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0;
}
export async function fetchManifest(
session: ChannelSession,
shareId: string,
info: ShareWire,
signal?: AbortSignal,
): Promise<ShareManifest> {
if (!info.manifest_item_id) throw new IntegrityError('bad_meta');
const response = await apiFetch(`/shares/${shareId}/manifest`, { headers: session.authHeaders('read'), signal });
const events = await session.createItemReader().pushFrame(new Uint8Array(await response.arrayBuffer()));
const rejection = events.find((event) => event.type === 'rejected');
if (rejection?.type === 'rejected') throw new IntegrityError(rejection.reason);
const found = events.find((event) => event.type === 'item');
if (!found || found.type !== 'item' || found.item.kind !== 'manifest' || !found.item.complete
|| found.item.itemId !== info.manifest_item_id) throw new IntegrityError('bad_meta');
let value: unknown;
try { value = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(found.item.content)); }
catch { throw new IntegrityError('bad_meta'); }
if (typeof value !== 'object' || value === null || !('v' in value) || value.v !== 1 || !('items' in value) || !Array.isArray(value.items))
throw new IntegrityError('bad_meta');
const listed = new Set<string>();
const validated: ProtocolManifestEntry[] = [];
for (const raw of value.items) {
if (typeof raw !== 'object' || raw === null || !('t' in raw) || !safeTime('ts' in raw ? raw.ts : undefined))
throw new IntegrityError('bad_meta');
if (raw.t === 'text' && 'text' in raw && typeof raw.text === 'string') {
if ('name' in raw && raw.name !== undefined && typeof raw.name !== 'string') throw new IntegrityError('bad_meta');
validated.push({ t: 'text', ...(typeof raw.name === 'string' ? { name: raw.name } : {}), text: raw.text, ts: raw.ts });
continue;
}
if (raw.t === 'file' && 'id' in raw && typeof raw.id === 'string' && /^[A-Za-z0-9_-]{22}$/.test(raw.id)
&& 'size' in raw && typeof raw.size === 'number' && Number.isSafeInteger(raw.size) && raw.size >= 0) {
if (listed.has(raw.id)) throw new IntegrityError('bad_meta');
if ('name' in raw && raw.name !== undefined && typeof raw.name !== 'string') throw new IntegrityError('bad_meta');
if ('mime' in raw && raw.mime !== undefined && typeof raw.mime !== 'string') throw new IntegrityError('bad_meta');
listed.add(raw.id);
validated.push({
t: 'file', id: raw.id,
...(typeof raw.name === 'string' ? { name: raw.name } : {}),
...(typeof raw.mime === 'string' ? { mime: raw.mime } : {}),
size: raw.size, ts: raw.ts,
});
continue;
}
throw new IntegrityError('bad_meta');
}
return { v: 1, items: validated };
}
+122
View File
@@ -0,0 +1,122 @@
import { ChannelSession, IntegrityError } from '@webdrop/core';
import { getDraftFile, saveResource, type SavedResource } from '../database';
import type { FileItem, TextItem } from '../types';
import { makeFileItem } from './presentation';
import { active, blobs, getSession } from './sessions';
import { fail, readJson } from './http';
import { toShareInfo, type ShareWire } from './wire';
import { flushOutbox, downloadDecrypted } from './transfer';
import { publishShare } from './presentation';
import { fetchManifest, type ManifestFile } from './shareManifest';
import { readShareInfo } from './shareState';
import { errorMessage } from '../errors';
export async function connectShare(id: string): Promise<void> {
active.get(id)?.abort();
const controller = new AbortController();
active.set(id, controller);
try {
const opened = await getSession(id);
let { session } = opened;
const { resource } = opened;
if (controller.signal.aborted) return;
if (resource.kind !== 'share') throw new Error('资源类型不匹配');
const raw = await readShareInfo(session, resource);
session = (await getSession(id)).session;
if (controller.signal.aborted) return;
const share = toShareInfo(raw, resource);
resource.state = raw.state;
resource.expiresAt = raw.expires_at;
await saveResource(resource);
publishShare(id, { t: 'share', share });
if (raw.state === 'draft' && resource.mine) {
const resumed = await flushOutbox(resource, session);
if (resumed) resource.draftItems = resource.draftItems?.map((item) => item.state.kind === 'incomplete' ? { ...item, state: { kind: 'complete' } } : item);
await saveResource(resource);
if (controller.signal.aborted) return;
for (const item of resource.draftItems ?? []) publishShare(id, { t: 'item', item });
publishShare(id, { t: 'ready' });
return;
}
await loadSealedManifest(resource, session, raw, controller.signal);
} catch (error) {
if (controller.signal.aborted) return;
publishShare(id, error instanceof IntegrityError
? { t: 'integrity', message: '分享清单校验失败,内容已拒绝显示。' }
: { t: 'error', message: errorMessage(error) });
publishShare(id, { t: 'ready' });
} finally {
if (active.get(id) === controller) active.delete(id);
}
}
export function stopShare(id: string): void {
active.get(id)?.abort();
active.delete(id);
}
async function loadSealedManifest(resource: SavedResource, session: ChannelSession, raw: ShareWire, signal: AbortSignal): Promise<void> {
try {
const manifest = await fetchManifest(session, resource.id, raw, signal);
const byId = new Map(raw.items.map((item) => [item.item_id, item]));
for (let index = 0; index < manifest.items.length; index++) {
const entry = manifest.items[index];
if (!entry) throw new Error('invalid_manifest_item');
if (entry.t === 'text' && typeof entry.text === 'string' && Number.isSafeInteger(entry.ts)) {
const item: TextItem = { id: `manifest-text-${index}`, kind: 'text', ts: entry.ts, deviceId: '创建者', state: { kind: 'complete' }, text: entry.text };
publishShare(resource.id, { t: 'item', item });
} else if (entry.t === 'file' && typeof entry.id === 'string' && typeof entry.size === 'number' && Number.isSafeInteger(entry.size)) {
const detail = byId.get(entry.id);
const item = makeFileItem(entry.id, entry.name ?? '加密文件', entry.mime ?? 'application/octet-stream', entry.size,
entry.ts, { kind: 'complete' }, detail?.downloads_left ?? (detail ? null : 0));
item.gated = true;
publishShare(resource.id, { t: 'item', item });
} else throw new Error('invalid_manifest_item');
}
} catch {
publishShare(resource.id, { t: 'integrity', message: '分享清单格式不正确,内容已拒绝显示。' });
}
publishShare(resource.id, { t: 'ready' });
}
export async function getShareBlob(id: string, itemId: string): Promise<Blob> {
const cached = blobs.get(`${id}:${itemId}`);
if (cached) return cached;
const { session, resource } = await getSession(id);
if (resource.kind !== 'share') throw new Error('资源类型不匹配');
if (resource.state === 'draft' && resource.mine) {
const draftFile = await getDraftFile(id, itemId);
if (draftFile) {
blobs.set(`${id}:${itemId}`, draftFile);
return draftFile;
}
}
const raw = await readJson<ShareWire>(`/shares/${id}`, session.authHeaders('read'));
const manifest = await fetchManifest(session, id, raw);
if (itemId.startsWith('manifest-text-')) {
const entry = manifest.items[Number(itemId.slice(14))];
if (entry?.t !== 'text') throw fail(404, 'not_found');
return new Blob([entry.text], { type: 'text/plain' });
}
const entry = manifest.items.find((candidate): candidate is ManifestFile => candidate.t === 'file' && candidate.id === itemId);
if (!entry) throw fail(404, 'not_found');
let result: Awaited<ReturnType<typeof downloadDecrypted>>;
try { result = await downloadDecrypted(session, resource, itemId, entry.size); }
catch (error) {
if (error instanceof IntegrityError) publishShare(id, { t: 'integrity', message: '文件校验失败,内容已丢弃。' });
throw error;
}
blobs.set(`${id}:${itemId}`, result.blob);
const fileInfo = raw.items.find((value) => value.item_id === itemId);
const remaining = fileInfo?.downloads_left === null || fileInfo?.downloads_left === undefined
? null
: Math.max(0, fileInfo.downloads_left - 1);
const item: FileItem = makeFileItem(itemId, entry.name ?? '加密文件', entry.mime ?? 'application/octet-stream', entry.size, entry.ts,
{ kind: 'complete' }, remaining);
publishShare(id, { t: 'item', item });
return result.blob;
}
@@ -0,0 +1,108 @@
import { ChannelSession } from '@webdrop/core';
import { getResource, listResources, removeResource, saveResource, type SavedResource } from '../database';
import type { Share, ShareSettings } from '../types';
import { active, blobs, getSession, sessions } from './sessions';
import { apiFetch, fail, readJson } from './http';
import { apiShareSettings, toShareInfo, type ShareWire } from './wire';
import { makeLink, toShareLink } from './links';
export async function createShare(settings: ShareSettings, captchaToken?: string): Promise<Share> {
const endpointSecret = crypto.getRandomValues(new Uint8Array(32));
const { session, request } = await ChannelSession.create({ mode: 'enc', endpointSecret });
try {
const body = {
id: request.body.id,
read_hash: request.body.read_hash,
owner_hash: request.body.owner_hash,
settings: apiShareSettings(settings),
};
const response = await apiFetch('/shares', {
method: 'POST', cache: 'no-store',
headers: { 'Content-Type': 'application/json', 'X-WD-Endpoint': request.headers['X-WD-Endpoint']!, ...(captchaToken ? { 'X-WD-Cap': captchaToken } : {}) },
body: JSON.stringify(body),
});
const raw = await response.json() as ShareWire;
const resource: SavedResource = {
id: raw.id, kind: 'share', mode: 'enc', role: 'owner', mine: true,
link: toShareLink(session.link(self.location.origin)),
ownerLink: toShareLink(session.link(self.location.origin, 'owner')),
endpointSecret: Uint8Array.from(session.endpointSecret),
createdAt: raw.created_at, expiresAt: raw.expires_at, seats: 0, membersCanUpload: false,
state: 'draft', maxVisitors: raw.settings.max_visitors,
ttlSeconds: raw.settings.ttl_seconds,
maxDownloads: raw.settings.file_max_downloads === 0 ? null : raw.settings.file_max_downloads,
draftItems: [], manifestEntries: [],
};
await saveResource(resource);
sessions.set(raw.id, session);
return toShareInfo(raw, resource);
} catch (error) {
session.dispose();
throw error;
}
}
export async function joinShare(input: { link: { id: string; key: string } }): Promise<Share> {
const existing = await getResource(input.link.id);
const endpointSecret = existing?.endpointSecret ?? crypto.getRandomValues(new Uint8Array(32));
const session = await ChannelSession.open(makeLink(self.location.origin, input.link.id, { k: input.link.key }), { endpointSecret });
try {
const raw = await readJson<ShareWire>(`/shares/${input.link.id}`, session.authHeaders('read'));
if (raw.state !== 'sealed') throw fail(404, 'not_found');
const resource: SavedResource = {
id: input.link.id, kind: 'share', mode: 'enc', role: 'member', mine: false,
link: toShareLink(makeLink(self.location.origin, input.link.id, { k: input.link.key })),
endpointSecret: Uint8Array.from(session.endpointSecret), createdAt: raw.created_at, expiresAt: raw.expires_at,
seats: 0, membersCanUpload: false, state: raw.state,
ttlSeconds: raw.settings.ttl_seconds,
maxVisitors: raw.settings.max_visitors, maxDownloads: raw.settings.file_max_downloads === 0 ? null : raw.settings.file_max_downloads,
};
await saveResource(resource);
sessions.set(resource.id, session);
return toShareInfo(raw, resource);
} catch (error) {
session.dispose();
throw error;
}
}
export async function updateShareSettings(id: string, patch: Partial<ShareSettings>): Promise<Share> {
const { session, resource } = await getSession(id);
if (resource.kind !== 'share' || !session.can('owner')) throw new Error('此分享不再可修改');
const body: Record<string, number> = {};
if (patch.ttlSeconds !== undefined) body.ttl_seconds = patch.ttlSeconds;
if (patch.maxVisitors !== undefined) body.max_visitors = patch.maxVisitors;
if (patch.maxDownloads !== undefined) body.file_max_downloads = patch.maxDownloads ?? 0;
const raw = await apiFetch(`/shares/${id}/settings`, {
method: 'PATCH', headers: { ...session.authHeaders('owner'), 'Content-Type': 'application/json' }, body: JSON.stringify(body),
}).then((response) => response.json() as Promise<ShareWire>);
resource.expiresAt = raw.expires_at;
resource.ttlSeconds = raw.settings.ttl_seconds;
resource.maxVisitors = raw.settings.max_visitors;
resource.maxDownloads = raw.settings.file_max_downloads === 0 ? null : raw.settings.file_max_downloads;
await saveResource(resource);
return toShareInfo(raw, resource);
}
export async function removeShare(id: string): Promise<void> {
active.get(id)?.abort();
active.delete(id);
sessions.get(id)?.dispose();
sessions.delete(id);
for (const key of blobs.keys()) if (key.startsWith(`${id}:`)) blobs.delete(key);
await removeResource(id);
}
export async function listSavedShares(): Promise<Share[]> {
const saved = await listResources('share');
return saved.map((resource) => ({
id: resource.id, state: resource.state ?? 'draft', mine: resource.mine ?? false,
link: resource.link, createdAt: resource.createdAt, expiresAt: resource.expiresAt,
ttlSeconds: resource.ttlSeconds ?? 86400, maxVisitors: resource.maxVisitors ?? 0,
maxDownloads: resource.maxDownloads ?? null,
}));
}
+48
View File
@@ -0,0 +1,48 @@
import { saveResource } from '../database';
import type { Item, Share } from '../types';
import { getSession, sessions } from './sessions';
import { apiFetch } from './http';
import { toShareInfo, type ShareWire } from './wire';
import { flushOutbox, persistPlan } from './transfer';
import { publishShare } from './presentation';
import { encodeManifest } from './shareManifest';
import { readShareInfo, releaseOwner } from './shareState';
export async function sealShare(id: string): Promise<Share> {
const { session, resource } = await getSession(id);
if (resource.kind !== 'share' || resource.state !== 'draft' || !session.can('owner')) throw new Error('此分享已封存');
if (!(await flushOutbox(resource, session))) throw new Error('网络连接失败,请稍后重试');
const before = await readShareInfo(session, resource);
if (before.state === 'sealed') return toShareInfo(before, resource);
if ((resource.draftItems ?? []).some((item) => item.state.kind !== 'complete')) throw new Error('还有内容没有上传完成');
if (!resource.pendingManifestId) {
const bytes = encodeManifest(resource.manifestEntries ?? []);
const plan = session.encryptBytes(bytes, { kind: 'manifest' }, { gated: false });
await persistPlan(resource, plan);
resource.pendingManifestId = plan.itemId;
await saveResource(resource);
}
if (!(await flushOutbox(resource, session))) throw new Error('清单上传暂停,请重试');
let raw: ShareWire;
try {
const response = await apiFetch(`/shares/${id}/finish`, {
method: 'POST', headers: { ...session.authHeaders('owner'), 'Content-Type': 'application/json' },
body: JSON.stringify({ manifest_item_id: resource.pendingManifestId }),
});
raw = await response.json() as ShareWire;
} catch (error) {
raw = await readShareInfo(session, resource).catch(() => { throw error; });
if (raw.state !== 'sealed') throw error;
}
const items = resource.draftItems ?? [];
if (sessions.get(id)?.can('owner')) await releaseOwner(session, resource, raw);
const sealed = toShareInfo(raw, resource);
publishShare(id, { t: 'share', share: sealed });
const remaining = new Map(raw.items.map((item) => [item.item_id, item.downloads_left]));
for (const draft of items) {
const item: Item = draft.kind === 'file' ? { ...draft, remaining: remaining.get(draft.id) ?? null } : draft;
publishShare(id, { t: 'item', item });
}
publishShare(id, { t: 'ready' });
return sealed;
}
+49
View File
@@ -0,0 +1,49 @@
import { ChannelSession } from '@webdrop/core';
import { clearDraftFiles, dequeueItem, saveResource, type SavedResource } from '../database';
import { blobs, sessions } from './sessions';
import { apiFetch, readJson } from './http';
import { toCoreLink } from './links';
import type { ShareWire } from './wire';
export async function readShareInfo(session: ChannelSession, resource: SavedResource): Promise<ShareWire> {
const role = resource.state === 'draft' && session.can('owner') ? 'owner' : 'read';
let raw: ShareWire;
try { raw = await readJson<ShareWire>(`/shares/${resource.id}`, session.authHeaders(role)); }
catch (error) {
if (role !== 'owner' || (error as { status?: number }).status !== 401) throw error;
// finish 响应丢失时,读凭据仍可确认服务端已封存。
raw = await readJson<ShareWire>(`/shares/${resource.id}`, session.authHeaders('read'));
}
if (raw.id !== resource.id) throw new Error('分享信息与链接不符');
if (raw.state === 'sealed' && session.can('owner')) await releaseOwner(session, resource, raw);
return raw;
}
export async function releaseOwner(session: ChannelSession, resource: SavedResource, raw: ShareWire): Promise<void> {
const reader = await ChannelSession.open(toCoreLink(resource.link), { endpointSecret: resource.endpointSecret });
for (const key of blobs.keys()) if (key.startsWith(`${resource.id}:`)) blobs.delete(key);
try { await clearDraftFiles(resource.id); }
catch { /* 清理失败不能让已封存资源继续保留 owner 会话 */ }
session.dispose();
sessions.set(resource.id, reader);
resource.ownerLink = undefined;
resource.role = 'member';
resource.state = 'sealed';
resource.expiresAt = raw.expires_at;
resource.createdAt = raw.sealed_at ?? raw.created_at;
resource.ttlSeconds = raw.settings.ttl_seconds;
resource.pendingManifestId = undefined;
resource.draftItems = [];
resource.manifestEntries = [];
await saveResource(resource);
}
export async function invalidateManifest(session: ChannelSession, resource: SavedResource): Promise<void> {
const id = resource.pendingManifestId;
if (!id) return;
try { await apiFetch(`/shares/${resource.id}/items/${id}`, { method: 'DELETE', headers: session.authHeaders('owner') }); }
catch (error) { if ((error as { status?: number }).status !== 404) throw error; }
await dequeueItem(resource.id, id);
resource.pendingManifestId = undefined;
await saveResource(resource);
}
+74
View File
@@ -0,0 +1,74 @@
import { fromBase64Url, IntegrityError, type ChannelSession, type ItemDecodeResult, type ItemPlan } from '@webdrop/core';
import { dequeueFrame, queueFrame, nextQueuedFrame, type SavedResource } from '../database';
import { apiFetch, type ApiError } from './http';
import { serial } from './serial';
export async function persistPlan(resource: SavedResource, plan: ItemPlan): Promise<void> {
const createdAt = Date.now();
for await (const frame of plan.frames) {
await queueFrame({
key: `${resource.id}:${plan.itemId}:${String(frame.index).padStart(10, '0')}`,
resourceId: resource.id,
itemId: plan.itemId,
index: frame.index,
bytes: frame.bytes,
extraHeaders: frame.index === 0 ? { ...plan.frame0Headers } : {},
createdAt: createdAt + frame.index,
});
}
}
function writeRole(session: ChannelSession): 'owner' | 'write' {
if (session.can('owner')) return 'owner';
if (session.can('write')) return 'write';
throw new Error('此链接没有上传权限');
}
/** The outbox contains ciphertext only. Every retry sends these exact bytes. */
export async function flushOutbox(resource: SavedResource, session: ChannelSession): Promise<boolean> {
return serial(`outbox:${resource.id}`, () => flush(resource, session));
}
async function flush(resource: SavedResource, session: ChannelSession): Promise<boolean> {
for (;;) {
const frame = await nextQueuedFrame(resource.id);
if (!frame) return true;
const role = resource.kind === 'share' ? 'owner' : writeRole(session);
const path = resource.kind === 'share' ? `/shares/${resource.id}/frames` : `/channels/${resource.id}/frames`;
try {
await apiFetch(path, {
method: 'POST',
headers: { ...session.authHeaders(role), 'Content-Type': 'application/octet-stream', ...frame.extraHeaders },
body: frame.bytes.slice().buffer as ArrayBuffer,
});
await dequeueFrame(frame.key);
} catch (error) {
const apiError = error as ApiError;
if (apiError.status === undefined || apiError.status === 429 || apiError.status >= 500) return false;
throw error;
}
}
}
export async function downloadDecrypted(
session: ChannelSession,
resource: SavedResource,
itemId: string,
expectedSize?: number,
): Promise<{ blob: Blob; meta: ItemDecodeResult['meta'] }> {
const role = session.can('read') ? 'read' : 'owner';
const path = resource.kind === 'share'
? `/shares/${resource.id}/items/${itemId}/frames?from=0`
: `/channels/${resource.id}/items/${itemId}/frames?from=0`;
const response = await apiFetch(path, { headers: session.authHeaders(role) });
if (!response.body) throw new Error('服务端没有返回文件内容');
const expectedId = fromBase64Url(itemId, 16);
if (!expectedId) throw new IntegrityError('bad_meta');
const decoded = session.decryptItemStream(response.body, { itemId: expectedId });
const blobPromise = new Response(decoded.content).blob();
const [result, blob] = await Promise.all([decoded.result, blobPromise]);
if (!result.complete) throw new Error('文件传输不完整');
if (result.meta?.size !== undefined && result.meta.size !== blob.size) throw new IntegrityError('bad_meta');
if (expectedSize !== undefined && expectedSize !== blob.size) throw new IntegrityError('bad_meta');
return { blob, meta: result.meta };
}
+98
View File
@@ -0,0 +1,98 @@
import { fromBase64Url, toBase64Url, type ChannelSession } from '@webdrop/core';
import { apiFetch, apiPath } from './http';
export interface StreamConsumer {
cursor: number;
record(bytes: Uint8Array): Promise<void>;
control(value: { t?: string; seq?: number }): Promise<void>;
online(): void;
}
export function socketStream(id: string, session: ChannelSession, signal: AbortSignal, consumer: StreamConsumer): Promise<void> {
if (signal.aborted) return Promise.resolve();
const address = new URL(apiPath(`/channels/${id}/ws?from=${consumer.cursor}`), self.location.origin);
address.protocol = address.protocol === 'https:' ? 'wss:' : 'ws:';
const token = session.authHeaders('read').Authorization?.slice(7);
const socket = new WebSocket(address, ['wd.v1', 'wd.role.read', ...(token ? [`wd.auth.${token}`] : []), `wd.ep.${toBase64Url(session.endpointSecret)}`]);
socket.binaryType = 'arraybuffer';
return new Promise((resolve, reject) => {
let chain = Promise.resolve();
let failure: unknown;
let timer: ReturnType<typeof setTimeout>;
const heartbeat = () => {
clearTimeout(timer);
timer = setTimeout(() => { failure = new Error('连接超时'); socket.close(); }, 45_000);
};
const abort = () => socket.close();
signal.addEventListener('abort', abort, { once: true });
heartbeat();
socket.onopen = () => consumer.online();
socket.onmessage = (message) => {
heartbeat();
// 建立连接起就排队,异步解密期间不会错过下一帧。
chain = chain.then(async () => {
if (signal.aborted) return;
if (typeof message.data === 'string') await consumer.control(JSON.parse(message.data));
else await consumer.record(new Uint8Array(message.data as ArrayBuffer));
}).catch((error: unknown) => { failure = error; socket.close(); });
};
socket.onerror = () => { failure = new Error('连接失败'); socket.close(); };
socket.onclose = () => {
clearTimeout(timer);
signal.removeEventListener('abort', abort);
void chain.then(() => { if (failure && !signal.aborted) reject(failure); else resolve(); });
};
});
}
export async function sseStream(id: string, session: ChannelSession, signal: AbortSignal, consumer: StreamConsumer): Promise<void> {
const controller = new AbortController();
const abort = () => controller.abort();
signal.addEventListener('abort', abort, { once: true });
let timer = setTimeout(abort, 45_000);
try {
if (signal.aborted) return;
const response = await apiFetch(`/channels/${id}/events?from=${consumer.cursor}`, { headers: session.authHeaders('read'), signal: controller.signal });
if (!response.body) throw new Error('连接失败');
consumer.online();
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
try {
while (!signal.aborted) {
const chunk = await reader.read();
if (chunk.done) break;
clearTimeout(timer);
timer = setTimeout(abort, 45_000);
buffer += decoder.decode(chunk.value, { stream: true }).replace(/\r/g, '');
let boundary: number;
while ((boundary = buffer.indexOf('\n\n')) !== -1) {
const block = buffer.slice(0, boundary);
buffer = buffer.slice(boundary + 2);
const lines = block.split('\n');
const event = lines.find((line) => line.startsWith('event:'))?.slice(6).trim();
const data = lines.filter((line) => line.startsWith('data:')).map((line) => line.slice(5).trimStart()).join('\n');
if (event === 'log') {
const bytes = fromBase64Url(data);
if (!bytes) throw new Error('内容校验失败');
await consumer.record(bytes);
} else if (event === 'control') await consumer.control(JSON.parse(data));
else if (event === 'error') throw new Error('连接权限已失效');
}
}
} finally { await reader.cancel().catch(() => undefined); reader.releaseLock(); }
} finally {
clearTimeout(timer);
signal.removeEventListener('abort', abort);
controller.abort();
}
}
export function delay(ms: number, signal: AbortSignal): Promise<void> {
return new Promise((resolve) => {
if (signal.aborted) return resolve();
const finish = () => { clearTimeout(timer); signal.removeEventListener('abort', finish); resolve(); };
const timer = setTimeout(finish, ms);
signal.addEventListener('abort', finish, { once: true });
});
}
+54
View File
@@ -0,0 +1,54 @@
import type { SavedResource } from '../database';
import type { ChannelInfo, Share, ShareSettings } from '../types';
export interface ChannelWire {
id: string;
mode: 'enc' | 'raw';
settings: { seats: number; members_can_upload: boolean; delete_policy: 'owner_only' | 'own' | 'anyone'; ttl_seconds: number; max_items: number; file_max_downloads: number };
seats_used: number;
oldest_seq: number;
last_seq: number;
created_at: number;
expires_at: number;
}
export interface ShareWireItem { item_id: string; bytes: number; frames: number; max_downloads: number; downloads_left: number | null }
export interface ShareWire {
id: string;
state: 'draft' | 'sealed';
expires_at: number;
created_at: number;
sealed_at?: number;
manifest_item_id?: string;
settings: { ttl_seconds: number; max_visitors: number; file_max_downloads: number };
items: ShareWireItem[];
}
export function toChannelInfo(raw: ChannelWire, saved: SavedResource): ChannelInfo {
return {
id: raw.id,
mode: raw.mode,
role: saved.role,
canUpload: saved.role === 'owner' || raw.settings.members_can_upload,
seats: raw.settings.seats,
memberLink: saved.link,
ownerLink: saved.ownerLink,
createdAt: raw.created_at,
expiresAt: raw.expires_at,
};
}
export function toShareInfo(raw: ShareWire, saved: SavedResource): Share {
return {
id: raw.id,
state: raw.state,
mine: saved.mine ?? (saved.role === 'owner' || saved.ownerLink !== undefined),
link: saved.link,
createdAt: raw.sealed_at ?? raw.created_at,
expiresAt: raw.expires_at,
ttlSeconds: raw.settings.ttl_seconds,
maxVisitors: raw.settings.max_visitors,
maxDownloads: raw.settings.file_max_downloads === 0 ? null : raw.settings.file_max_downloads,
};
}
export function apiShareSettings(settings: ShareSettings) {
return { ttl_seconds: settings.ttlSeconds, max_visitors: settings.maxVisitors, file_max_downloads: settings.maxDownloads ?? 0 };
}