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

import { removeChannelItem } from './channelItems';
import { queueFileCatalog } from './channelThreads';
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 { 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 { publishWorkspace } from './presentation';

export async function sendText(resourceId: string, text: string, threadId?: string,
  event?: { kind: 'chat.create' | 'chat.edit'; targetItemId?: string }, localId?: string): Promise<{ itemId: string; uploaded: boolean }> {
  if (!threadId) throw new Error('请先选择频道');
  const { session, resource } = await getSession(resourceId);
  if (resource.kind !== 'channel' || (!session.can('owner') && !session.can('write'))) throw new Error('此链接没有上传权限');
  const ts = Date.now();
  const author = await senderSnapshot();
  const prepared = await prepareObject(resource, session, threadId, 'chat');
  const plan = session.encryptText(text, { ts, threadId, itemId: prepared.itemIdBytes });
  const item: Item = { id: plan.itemId, kind: 'text', ts, ...(author ? { author } : {}), state: { kind: 'receiving', progress: 0 }, text, ...(threadId ? { threadId } : {}), ...(localId ? { localId } : {}) };
  blobs.set(`${resourceId}:${item.id}`, new Blob([text], { type: 'text/plain' }));
  publishWorkspace(resourceId, { t: 'item', item });
  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);
  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[], suppliedOptions?: FileUploadOptions, threadId?: string, folderId?: string, localIds?: string[]): Promise<void> {
  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<number>();
      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;
    }
  }
}

export async function updateSnapshot(resourceId: string, text: string): Promise<void> {
  const { session, resource } = await getSession(resourceId);
  const name = accountDisplay()?.username ?? '';
  const { structure } = await import('./workspaceStructure').then(module => module.loadWorkspace(resourceId));
  if (!structure.defaultChannelId) throw new Error('请先创建默认频道');
  const prepared = await prepareObject(resource, session, structure.defaultChannelId, 'chat');
  const plan = session.encryptJson({ text, updatedBy: name, updatedAt: Date.now() }, { snapshot: true, itemId: prepared.itemIdBytes });
  await persistPlan(resource, plan, undefined, { channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: 'chat.snapshot' });
  await flushOutbox(resource, session);
}

export async function editChannelText(resourceId: string, itemId: string, text: string, threadId?: string): Promise<void> {
  const created = await sendText(resourceId, text, threadId, { kind: 'chat.edit', targetItemId: itemId });
  if (!created.uploaded) throw new Error('新版本已排队,网络恢复前保留旧内容。');
  await removeChannelItem(resourceId, itemId);
}