feat(files): store files on the host when a channel has no file domain
Chat stays usable when the file service is off or down. Files up to 256 MB are hosted like chat; downloads try the host first and fall back to file-service tickets. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> [CI]
This commit is contained in:
1 parent
7e6ef5e587
commit
4b4ab4ee4d
8 files changed
+47
-16
No files matched your search
@@ -18,9 +18,9 @@ public sealed partial class ChannelService(
|
||||
private IFileGrants Files => fileGrants ?? NoFileGrants.Instance;
|
||||
/// <summary>帧内容在文件域里的帧,BlobKey 以它开头;主后端只有头部、长度和摘要</summary>
|
||||
public const string FilesPrefix = "files:";
|
||||
/// <summary>这个条目是文件域的对象(不含目录控制帧)</summary>
|
||||
/// <summary>这个条目是文件域的对象(不含目录控制帧)。所在频道没有分配文件域(文件服务没开、小文件)时,文件和聊天一样存在主后端</summary>
|
||||
private static bool IsFileObject(IResourceTx tx, Id16 channelId, Envelope env) =>
|
||||
!env.IsControl && tx.GetWorkspaceCrypto(channelId).Objects.Exists(o => o.ItemId == env.ItemId.ToBase64Url() && o.Domain == CryptoDomain.Files);
|
||||
!env.IsControl && FileDomainOf(tx, channelId, env.ItemId) is not null;
|
||||
|
||||
private static bool IsRemote(string? blobKey) => blobKey is not null && blobKey.StartsWith(FilesPrefix, StringComparison.Ordinal);
|
||||
|
||||
|
||||
@@ -86,6 +86,7 @@ public sealed class ChannelCryptoTests
|
||||
using var env = new TestEnv(); var ownerEndpoint = TestEnv.Endpoint(); var memberEndpoint = TestEnv.Endpoint();
|
||||
var space = env.Create(ownerEndpoint, ChannelMode.Raw); var owner = space.Owner(ownerEndpoint);
|
||||
var channelId = env.Service.GetWorkspace(space.Id, owner).Channels.Single().Id;
|
||||
env.Service.SetWorkspaceChannelFileDomain(space.Id, owner, channelId, "domain-1");
|
||||
var oldFile = TestEnv.NewId();
|
||||
env.Service.DeclareObject(space.Id, owner, new ObjectEnvelope { ItemId = oldFile.ToBase64Url(), ChannelId = channelId, Domain = CryptoDomain.Files, Epoch = 0, KeyEnvelope = "" });
|
||||
await Register(env.Service, space.Id, owner, TestEnv.Frame(oldFile, 0, FrameFlags.Final, 16));
|
||||
|
||||
@@ -167,7 +167,16 @@ public class ChannelServiceTests : IDisposable
|
||||
var item = TestEnv.NewId();
|
||||
Svc.DeclareObject(space.Id, owner, new ObjectEnvelope { ItemId = item.ToBase64Url(), ChannelId = channelId, Domain = CryptoDomain.Files, Epoch = 0, KeyEnvelope = "" });
|
||||
|
||||
// 文件域的对象不能把字节直接交给主后端
|
||||
// 频道没有文件域(文件服务没开、小文件):文件和聊天一样存在主后端,不能整个登记
|
||||
var hosted = TestEnv.NewId();
|
||||
Svc.DeclareObject(space.Id, owner, new ObjectEnvelope { ItemId = hosted.ToBase64Url(), ChannelId = channelId, Domain = CryptoDomain.Files, Epoch = 0, KeyEnvelope = "" });
|
||||
Assert.True((await Svc.AppendFrameAsync(space.Id, owner, TestEnv.Frame(hosted, 0, FrameFlags.Gated | FrameFlags.Final, 60), 3)).Created);
|
||||
var hostedFrame = TestEnv.Frame(hosted, 0, FrameFlags.Gated | FrameFlags.Final, 60);
|
||||
Assert.Equal(ErrorCode.BadRequest, Assert.Throws<WebDropException>(() => Svc.RegisterFileItem(space.Id, owner, hostedFrame.AsMemory(0, Envelope.HeaderSize),
|
||||
[new ChannelService.FrameEntry(60, System.Security.Cryptography.SHA256.HashData(hostedFrame))], 3)).Code);
|
||||
|
||||
// 分配了文件域的频道:文件域的对象不能把字节直接交给主后端
|
||||
Svc.SetWorkspaceChannelFileDomain(space.Id, owner, channelId, "domain-1");
|
||||
byte[] first = TestEnv.Frame(item, 0, FrameFlags.Gated, 120);
|
||||
await AssertCode(ErrorCode.UseFileApi, () => Svc.AppendFrameAsync(space.Id, owner, first, null));
|
||||
|
||||
|
||||
@@ -88,8 +88,8 @@ export function Workspace({ onCreateChannel }: { onCreateChannel: () => void })
|
||||
// 文件服务现在能不能用、创建频道时可选的文件空间档位(文件服务公布)
|
||||
const { status: fileService, options: fileOptions } = useFileService(channel?.id);
|
||||
const caps = useCapabilities();
|
||||
// 服务端没有开启文件服务时不能上传(连不上只是暂时的,上传会排队,恢复后继续)
|
||||
const enabled = Boolean(activeThreadId && caps.canUpload && fileService !== 'disabled');
|
||||
// 聊天和小文件不依赖文件服务(没开启时小文件存在主后端,大文件由上传时提示);连不上文件服务只是暂时的,上传会排队
|
||||
const enabled = Boolean(activeThreadId && caps.canUpload);
|
||||
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);
|
||||
|
||||
@@ -91,7 +91,7 @@ export function FileExplorer(props: Props) {
|
||||
</div>
|
||||
</div>
|
||||
</header>
|
||||
{props.fileService === 'disabled' ? <p className={styles.notice} role="status">服务端没有开启文件服务,暂时不能上传和下载文件。聊天不受影响。</p> : null}
|
||||
{props.fileService === 'disabled' ? <p className={styles.notice} role="status">服务端没有开启文件服务,暂时只能上传 256 MB 以内的文件。聊天不受影响。</p> : null}
|
||||
{props.fileService === 'down' ? <p className={styles.notice} role="status">文件服务暂时连不上(可能还在部署或所在机器离线)。聊天不受影响;上传会排队,恢复后继续。</p> : null}
|
||||
<input type="file" multiple hidden ref={input} onChange={(event) => { if (event.target.files?.length) props.onFiles([...event.target.files], props.folderId); event.target.value = ''; }} />
|
||||
{props.filterBar}
|
||||
|
||||
@@ -66,21 +66,34 @@ export async function createFileDomain(resource: SavedResource, session: Channel
|
||||
* 保证这个频道有文件域:没有就创建一个并记到频道上(创建频道时没带的、旧频道)。
|
||||
* 只有所有者和管理员能分配;其他人遇到没有文件域的频道只能等管理员分配。
|
||||
*/
|
||||
export async function ensureThreadFileDomain(resource: SavedResource, session: ChannelSession, threadId: string): Promise<void> {
|
||||
if (resource.kind !== 'channel') return;
|
||||
export async function ensureThreadFileDomain(resource: SavedResource, session: ChannelSession, threadId: string, size: number): Promise<boolean> {
|
||||
if (resource.kind !== 'channel') return false;
|
||||
const thread = (await cachedWorkspace(resource.id)).structure.channels.find((channel) => channel.id === threadId);
|
||||
if (!thread) throw new Error('频道不存在');
|
||||
if (thread.fileDomain) return;
|
||||
if (!await filesEnabled()) throw new Error('服务端没有开启文件服务,不能上传文件');
|
||||
if (thread.fileDomain) return true;
|
||||
// 文件服务没开启或没部署好:小文件照样存在主后端(和聊天一样),大文件要等文件服务
|
||||
const hosted = (reason: string): false => {
|
||||
if (size > HOSTED_FILE_MAX_BYTES) throw new Error(`${reason},暂时只能上传不超过 ${HOSTED_FILE_MAX_BYTES / 1024 / 1024} MB 的文件`);
|
||||
return false;
|
||||
};
|
||||
if (!await filesEnabled()) return hosted('服务端没有开启文件服务');
|
||||
let domain: string;
|
||||
try { domain = await createFileDomain(resource, session); }
|
||||
catch (error) {
|
||||
if (hasCode(error, ErrorCode.Forbidden)) throw new Error('这个频道还没有文件空间,请让管理员先为它分配');
|
||||
if (hasCode(error, ErrorCode.Forbidden)) {
|
||||
if (size <= HOSTED_FILE_MAX_BYTES) return false;
|
||||
throw new Error('这个频道还没有文件空间,请让管理员先为它分配');
|
||||
}
|
||||
if (error instanceof Error && error.message === FILE_SERVICE_DOWN) return hosted('文件服务暂时不可用');
|
||||
throw error;
|
||||
}
|
||||
await changeWorkspace(resource.id, `channels/${threadId}/file-domain`, 'PUT', { fileDomain: domain });
|
||||
return true;
|
||||
}
|
||||
|
||||
/** 没有文件域时,文件存在主后端能收的上限 */
|
||||
export const HOSTED_FILE_MAX_BYTES = 256 * 1024 * 1024;
|
||||
|
||||
/** 频道文件域的容量和已用字节 */
|
||||
export async function fileDomainUsage(resourceId: string, threadId: string): Promise<FileDomainUsage | undefined> {
|
||||
const { session, resource } = await getSession(resourceId);
|
||||
@@ -141,6 +154,10 @@ export async function deleteFileBytes(grant: FileGrantBody, itemId: string): Pro
|
||||
*/
|
||||
export async function openItemFrames(resource: SavedResource, session: ChannelSession, itemId: string, from: number, signal?: AbortSignal): Promise<Response> {
|
||||
if (resource.kind === 'channel') {
|
||||
// 先问主后端:没有文件域的频道里,文件和聊天一样存在这里;存在文件服务的文件它会回 use_file_api,再去要票据
|
||||
try {
|
||||
return await apiFetch(`/channels/${resource.id}/items/${itemId}/frames?from=${from}`, { headers: session.authHeaders(tokenRole(session)), ...(signal ? { signal } : {}) });
|
||||
} catch (error) { if (!hasCode(error, ErrorCode.UseFileApi)) throw error; }
|
||||
const grant = await requestGrant(resource, session, { op: 'get', item: itemId, from });
|
||||
return externalFetch(`${base(grant)}/file/v1/domains/${grant.domain}/items/${itemHex(itemId)}/frames?from=${from}`, { 'X-WD-Ticket': grant.ticket }, signal);
|
||||
}
|
||||
|
||||
@@ -30,11 +30,15 @@ const REGISTER_ENTRY_SIZE = 36;
|
||||
/** 登记整个文件的请求(见 QueuedFrame.register):请求体 = 末帧头部 + 每帧(长度,摘要),头部带下载次数和末帧的签名事件 */
|
||||
export interface FileRegistration { createdAt: number; body: Uint8Array; headers: Record<string, string> }
|
||||
|
||||
function isFileDomainUpload(resource: SavedResource, signedEvent?: { domain: string; eventKind: string; hosted?: boolean }): boolean {
|
||||
return resource.kind === 'channel' && signedEvent?.domain === 'files' && signedEvent.eventKind === 'file.publish' && signedEvent.hosted !== true;
|
||||
}
|
||||
|
||||
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 },
|
||||
signedEvent?: { channelId: string; domain: 'chat' | 'files'; epoch: number; eventKind: string; targetItemId?: string; /** 文件存在主后端(频道没有文件域):不走文件服务,也不用整个文件登记 */ hosted?: boolean },
|
||||
/** 每一帧写入离线队列之后调用;上传流水线借此边加密边上传,返回的 Promise 用来给加密端施加背压 */
|
||||
onFrame?: (frame: QueuedFrame, final: boolean) => Promise<void>,
|
||||
): Promise<{ frames: number; bytes: number; registration?: FileRegistration }> {
|
||||
@@ -54,7 +58,7 @@ export async function persistPlan(
|
||||
bytes: frame.bytes,
|
||||
extraHeaders: { ...(frame.index === 0 ? plan.frame0Headers : {}), ...frame.extraHeaders, ...extraHeaders },
|
||||
createdAt: createdAt + frame.index,
|
||||
...(resource.kind === 'channel' && signedEvent?.domain === 'files' && signedEvent.eventKind === 'file.publish' ? { remote: true } : {}),
|
||||
...(isFileDomainUpload(resource, signedEvent) ? { remote: true } : {}),
|
||||
};
|
||||
// 提供了 onFrame 的调用方(上传流水线)自己决定何时落盘,见 frameStash
|
||||
if (!onFrame) await queueFrame(queued);
|
||||
@@ -77,7 +81,7 @@ export async function persistPlan(
|
||||
: {};
|
||||
await persistFrame(pending, headers, true);
|
||||
// 文件域的文件:字节只去文件服务,主后端一次登记整个文件(逐帧登记要为每个 1 MiB 往返一次主后端,大文件被它拖慢)
|
||||
if (resource.kind === 'channel' && signedEvent?.domain === 'files' && signedEvent.eventKind === 'file.publish') {
|
||||
if (isFileDomainUpload(resource, signedEvent)) {
|
||||
const body = new Uint8Array(FRAME_HEADER_SIZE + frameDigests.length * REGISTER_ENTRY_SIZE);
|
||||
body.set(pending.bytes.subarray(0, FRAME_HEADER_SIZE), 0);
|
||||
const view = new DataView(body.buffer);
|
||||
|
||||
@@ -88,7 +88,7 @@ async function runJob(job: Job): Promise<void> {
|
||||
try {
|
||||
if (signal.aborted) return;
|
||||
// 文件存在频道的文件域里:没有就先分配一个(所有者/管理员),成员遇到没有文件域的频道会收到提示
|
||||
await ensureThreadFileDomain(resource, session, threadId);
|
||||
const hosted = !await ensureThreadFileDomain(resource, session, threadId, file.size);
|
||||
const prepared = await prepareObject(resource, session, threadId, 'files');
|
||||
const plan = session.encryptFile(file, {}, {
|
||||
itemId: prepared.itemIdBytes,
|
||||
@@ -123,7 +123,7 @@ async function runJob(job: Job): Promise<void> {
|
||||
onBytes: progress.sent,
|
||||
});
|
||||
const planned = await persistPlan(resource, plan, undefined,
|
||||
{ channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: 'file.publish' },
|
||||
{ channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: 'file.publish', hosted },
|
||||
async (frame: QueuedFrame, final) => {
|
||||
if (frame.index === 0) progress.estimate(estimateCipherBytes(file.size, maxFrameData, frame.bytes.byteLength));
|
||||
await queued.add(frame);
|
||||
|
||||
Reference in new issue
Block a user