WebDrop/web/src/core-bridge/worker/channelEvents.ts

import { parseLogRecords, parseEnvelope, toBase64Url, type ChannelSession, type LogRecord, type ReceivedItem } from '@webdrop/core';
import { canDecryptRecord, objectEventProjection, readDataKeys, verifyObjectDeletion, verifyObjectEvent } from './channelDataKeys';
import { getResource, cacheResource, type SavedResource } from '../database';
import { channelCache } from './resourceMetadata';
import { loadWorkspace } from './workspaceStructure';
import type { FileItem, Item, Snapshot } from '../types';
import { previewKindOf } from '@/lib/fileKind';
import { blobs, readers } from './sessions';
import { showRejected, publishWorkspace } from './presentation';
import { type ChannelWire, toChannelInfo } from './wire';
import { apiFetch, readJson } from './http';
import { readResourceRecord } from './workspaceRecords';
import { clearFileCatalog, fileCatalogs, visibleFiles } from './fileCatalog';

export function mapReceived(item: ReceivedItem): Item {
  const timestamp = item.meta?.ts ?? Date.now();
  const state = item.complete ? { kind: 'complete' as const } : { kind: 'incomplete' as const };
  if (item.kind === 'file') {
    const name = item.meta?.name ?? '加密文件';
    return {
      id: item.itemId, kind: 'file', ts: timestamp, state, name,
      ...(item.meta?.threadId ? { threadId: item.meta.threadId } : {}),
      ...(item.meta?.folderId ? { folderId: item.meta.folderId } : {}),
      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, state,
    ...(item.meta?.threadId ? { threadId: item.meta.threadId } : {}),
    text: new TextDecoder().decode(item.content) };
}

export async function applyChannelRecord(resource: SavedResource, session: ChannelSession, input: Uint8Array | LogRecord): Promise<void> {
  if (!(input instanceof Uint8Array)) {
    const itemId = input.kind === 'frame' ? toBase64Url(parseEnvelope(input.envelope).itemId) : input.itemId;
    if (!(await canDecryptRecord(resource, session, itemId))) return;
  }
  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': {
        if (!event.item.complete || !event.item.ciphertextManifestHash
            || !await verifyObjectEvent(resource, session, event.item.itemId, event.item.ciphertextManifestHash)) {
          showRejected(resource.id, { type: 'rejected', itemId: event.item.itemId, seq: event.item.seq, reason: 'auth_failed' });
          break;
        }
        const signedEvent = await objectEventProjection(resource, session, event.item.itemId);
        if (signedEvent?.kind === 'chat.edit' && signedEvent.targetItemId)
          publishWorkspace(resource.id, { t: 'removed', id: signedEvent.targetItemId });
        if (signedEvent?.kind.startsWith('file.')) void loadWorkspace(resource.id).catch(() => {});
        const control = readResourceRecord(event.item);
        if (control !== false) {
          if (control && control.t === 'file-catalog') {
            const key = `${resource.id}:${control.item.id}`;
            fileCatalogs.set(key, control.item);
            if (visibleFiles.has(key)) publishWorkspace(resource.id, control);
            break;
          }
          if (control) publishWorkspace(resource.id, control);
          break;
        }
        const item = mapReceived(event.item);
        if (item.kind === 'file') visibleFiles.add(`${resource.id}:${item.id}`);
        if (event.item.snapshot) {
          let snapshot: Snapshot = { text: '', updatedBy: '', 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 ?? '', 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 key = `${resource.id}:${event.itemId}`;
        visibleFiles.add(key);
        const catalog = fileCatalogs.get(key);
        if (catalog) { publishWorkspace(resource.id, { t: 'file-catalog', item: catalog }); break; }
        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, state: { kind: 'incomplete' },
          name: '受限下载内容', mime: 'application/octet-stream', size: 0, remaining: null, preview: 'none',
          gated: true, metadataHidden: true,
        };
        publishWorkspace(resource.id, { t: 'item', item: placeholder });
        break;
      }
      case 'removed':
        if (event.reason === 'deleted' && !await verifyObjectDeletion(resource, session, event.itemId)) {
          showRejected(resource.id, { type: 'rejected', itemId: event.itemId, seq: event.seq, reason: 'auth_failed' });
          break;
        }
        visibleFiles.delete(`${resource.id}:${event.itemId}`);
        fileCatalogs.delete(`${resource.id}:${event.itemId}`);
        blobs.delete(`${resource.id}:${event.itemId}`);
        publishWorkspace(resource.id, { t: 'removed', id: event.itemId });
        void loadWorkspace(resource.id).catch(() => {});
        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(session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read'));
  session.verifyChannelInfo(info);
  Object.assign(resource, await cacheResource(resource, channelCache(info)));
  publishWorkspace(resource.id, { t: 'channel', info: toChannelInfo(info, resource) });
  await loadWorkspace(resource.id);
}

export async function syncChannel(resource: SavedResource, session: ChannelSession, signal: AbortSignal): Promise<number> {
  // reset 必须同步发出:connectChannel 随后并行拉取频道信息,它们的结果要落在 reset 之后
  publishWorkspace(resource.id, { t: 'reset' });
  clearFileCatalog(resource.id);
  readers.set(resource.id, session.createItemReader());
  // 密钥与第一页日志互不依赖,同时发出,首次打开少等一个往返
  const keys = resource.mode === 'enc' ? readDataKeys(resource, session) : undefined;
  keys?.catch(() => undefined);
  let from = 0;
  for (let first = true; ; first = false) {
    const response = await apiFetch(`/channels/${resource.id}/log?from=${from}&limit=200`, {
      headers: session.authHeaders('read'), signal,
    });
    if (first) await keys;
    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;
}