feat(cloud-agent): 云端 Agent - 完善运行时与资源持久化
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
import { getActiveUserScope } from "@/lib/user-scope";
|
||||
import { localForageStorageForScope } from "@/lib/localforage-storage";
|
||||
import { http, apiBaseURL, ApiError } from "@/services/api/request";
|
||||
import type { OSSConnectionTestInput, OSSConnectionTestResult, OSSProvider, S3Preset } from "@/lib/oss-settings";
|
||||
|
||||
@@ -117,8 +118,59 @@ export type ResourceAccess = {
|
||||
fallbackReason?: string;
|
||||
};
|
||||
|
||||
const accessCache = new Map<string, { value: ResourceAccess; expiresAt: number }>();
|
||||
type ResourceAccessBatchItem = { resourceId: string; access?: ResourceAccess; error?: { msg?: string } };
|
||||
type CachedResourceAccess = { value: ResourceAccess; expiresAt: number };
|
||||
const accessCache = new Map<string, CachedResourceAccess>();
|
||||
const accessRequests = new Map<string, Promise<ResourceAccess>>();
|
||||
const PERSISTED_DISPLAY_ACCESS_PREFIX = "resource-access-v1:";
|
||||
|
||||
function persistedDisplayAccessKey(resourceId: string, variant: ResourceAccessVariant) {
|
||||
return `${PERSISTED_DISPLAY_ACCESS_PREFIX}${resourceId}:${variant}`;
|
||||
}
|
||||
|
||||
function shouldPersistDisplayAccess(value: ResourceAccess) {
|
||||
// 只跨页面缓存 OSS/CDN 直连地址;平台本地/代理地址仍只在内存中保留,避免把平台签名凭据落到持久存储。
|
||||
return value.delivery === "cdn" || value.delivery === "origin";
|
||||
}
|
||||
|
||||
function resourceAccessCacheExpiry(value: ResourceAccess, now = Date.now()) {
|
||||
const expiresAt = value.expiresAt ? new Date(value.expiresAt).getTime() : Number.NaN;
|
||||
if (Number.isFinite(expiresAt) && expiresAt <= now + 1_000) return now;
|
||||
const refreshAt = value.refreshAt ? new Date(value.refreshAt).getTime() : Number.NaN;
|
||||
const candidates = [
|
||||
Number.isFinite(refreshAt) && refreshAt > now ? refreshAt : Number.POSITIVE_INFINITY,
|
||||
Number.isFinite(expiresAt) && expiresAt > now ? expiresAt - 15_000 : Number.POSITIVE_INFINITY,
|
||||
].filter(Number.isFinite);
|
||||
if (candidates.length) return Math.max(now, Math.min(...candidates));
|
||||
return now + 5 * 60_000;
|
||||
}
|
||||
|
||||
async function readPersistedDisplayAccess(scope: string, resourceId: string, variant: ResourceAccessVariant): Promise<CachedResourceAccess | undefined> {
|
||||
try {
|
||||
const raw = await localForageStorageForScope(scope).getItem(persistedDisplayAccessKey(resourceId, variant));
|
||||
if (!raw) return undefined;
|
||||
const parsed = JSON.parse(raw) as Partial<CachedResourceAccess>;
|
||||
if (!parsed.value?.url || !shouldPersistDisplayAccess(parsed.value) || typeof parsed.expiresAt !== "number") return undefined;
|
||||
const actualExpiresAt = parsed.value.expiresAt ? new Date(parsed.value.expiresAt).getTime() : Number.NaN;
|
||||
if (parsed.expiresAt <= Date.now() || (Number.isFinite(actualExpiresAt) && actualExpiresAt <= Date.now() + 1_000)) {
|
||||
await localForageStorageForScope(scope).removeItem(persistedDisplayAccessKey(resourceId, variant));
|
||||
return undefined;
|
||||
}
|
||||
return parsed as CachedResourceAccess;
|
||||
} catch {
|
||||
// 签名地址缓存只是性能优化,存储不可用或数据损坏时继续走后端签发。
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
async function persistDisplayAccess(scope: string, resourceId: string, variant: ResourceAccessVariant, entry: CachedResourceAccess) {
|
||||
if (!shouldPersistDisplayAccess(entry.value) || entry.expiresAt <= Date.now()) return;
|
||||
try {
|
||||
await localForageStorageForScope(scope).setItem(persistedDisplayAccessKey(resourceId, variant), JSON.stringify(entry));
|
||||
} catch {
|
||||
// 不把本地缓存失败升级为资源访问失败。
|
||||
}
|
||||
}
|
||||
let accessGeneration = 0;
|
||||
|
||||
/**
|
||||
@@ -335,15 +387,27 @@ export async function getResourceAccess(storageKey: string | undefined, purpose:
|
||||
let request!: Promise<ResourceAccess>;
|
||||
request = (async () => {
|
||||
try {
|
||||
const data = await http.post<{ items: Array<{ resourceId: string; access?: ResourceAccess; error?: { msg?: string } }> }>("/resources/access", [{ resourceId: id, purpose, variant, ...(downloadName ? { downloadName } : {}) }]);
|
||||
// 只跨页面复用普通展示地址。下载和模型输入地址可能包含更敏感的权限/文件名,仍只保留内存缓存。
|
||||
if (purpose === "display") {
|
||||
const persisted = await readPersistedDisplayAccess(scope, id, variant);
|
||||
assertResourceRequestCurrent(generation, scope);
|
||||
if (persisted) {
|
||||
accessCache.set(key, persisted);
|
||||
return persisted.value;
|
||||
}
|
||||
}
|
||||
const data = await http.post<{ items: ResourceAccessBatchItem[] }>("/resources/access", [{ resourceId: id, purpose, variant, ...(downloadName ? { downloadName } : {}) }]);
|
||||
if (generation !== accessGeneration || scope !== getActiveUserScope()) {
|
||||
throw new DOMException("资源访问请求已因账号切换失效", "AbortError");
|
||||
}
|
||||
const item = data.items?.[0];
|
||||
if (!item?.access?.url) throw new Error(item?.error?.msg || "后端未返回资源访问地址");
|
||||
const value = item.access;
|
||||
const ttl = value.expiresAt ? Math.max(10_000, new Date(value.expiresAt).getTime() - Date.now() - 15_000) : 5 * 60_000;
|
||||
accessCache.set(key, { value, expiresAt: Date.now() + ttl });
|
||||
const entry = { value, expiresAt: resourceAccessCacheExpiry(value) } satisfies CachedResourceAccess;
|
||||
if (entry.expiresAt > Date.now()) {
|
||||
accessCache.set(key, entry);
|
||||
if (purpose === "display") await persistDisplayAccess(scope, id, variant, entry);
|
||||
}
|
||||
return value;
|
||||
} catch (error) {
|
||||
if (error instanceof ApiError) throw new Error(error.message || "获取对象存储地址失败");
|
||||
|
||||
@@ -63,7 +63,19 @@ export async function loadCreationConversations<T extends StoredCreationConversa
|
||||
return parsed as T[];
|
||||
}
|
||||
|
||||
function persistableCreationConversations<T extends StoredCreationConversation>(conversations: T[]) {
|
||||
return conversations.map((conversation) => ({
|
||||
...conversation,
|
||||
messages: conversation.messages.map((message) => {
|
||||
const candidate = message as PendingCreationMessage & { resultUrls?: unknown; resultStorageKeys?: unknown };
|
||||
if (!Array.isArray(candidate.resultStorageKeys) || candidate.resultStorageKeys.length === 0) return message;
|
||||
const { resultUrls: _transientResultUrls, ...persistedMessage } = candidate;
|
||||
return persistedMessage as typeof message;
|
||||
}),
|
||||
})) as T[];
|
||||
}
|
||||
|
||||
export async function saveCreationConversations<T extends StoredCreationConversation>(conversations: T[]) {
|
||||
const storage = localForageStorageForScope(getActiveUserScope());
|
||||
await storage.setItem(CREATION_CONVERSATIONS_KEY, JSON.stringify(conversations));
|
||||
await storage.setItem(CREATION_CONVERSATIONS_KEY, JSON.stringify(persistableCreationConversations(conversations)));
|
||||
}
|
||||
|
||||
@@ -520,12 +520,13 @@ export async function consumeGenerationTaskAgent(
|
||||
export async function consumeGenerationTaskMessage(
|
||||
task: GenerationTask,
|
||||
messageId: string,
|
||||
consumer: (input: { task: GenerationTask; resultUrls: string[]; effectKey: string; signal?: AbortSignal }) => Promise<void> | void,
|
||||
consumer: (input: { task: GenerationTask; resultUrls: string[]; resultStorageKeys: string[]; effectKey: string; signal?: AbortSignal }) => Promise<void> | void,
|
||||
dependencies: {
|
||||
signal?: AbortSignal;
|
||||
managed?: true;
|
||||
materialize?: typeof materializeGenerationTaskAssets;
|
||||
materializedUrls?: typeof generationTaskMaterializedUrls;
|
||||
materializedStorageKeys?: typeof generationTaskMaterializedStorageKeys;
|
||||
attachMessage?: typeof attachGenerationTaskMessage;
|
||||
} = {},
|
||||
): Promise<GenerationTask> {
|
||||
@@ -534,6 +535,7 @@ export async function consumeGenerationTaskMessage(
|
||||
}
|
||||
const materialized = await (dependencies.materialize ?? materializeGenerationTaskAssets)(task, dependencies.signal);
|
||||
const resultUrls = (dependencies.materializedUrls ?? generationTaskMaterializedUrls)(materialized);
|
||||
const resultStorageKeys = (dependencies.materializedStorageKeys ?? generationTaskMaterializedStorageKeys)(materialized);
|
||||
const attach = dependencies.attachMessage ?? attachGenerationTaskMessage;
|
||||
const outputs = materialized.outputs?.filter((output) => output.materializedAssetId) ?? [];
|
||||
for (const output of outputs) {
|
||||
@@ -542,7 +544,7 @@ export async function consumeGenerationTaskMessage(
|
||||
messageId,
|
||||
output.outputIndex,
|
||||
async ({ effectKey, signal }) => {
|
||||
await consumer({ task: materialized, resultUrls, effectKey, signal });
|
||||
await consumer({ task: materialized, resultUrls, resultStorageKeys, effectKey, signal });
|
||||
},
|
||||
dependencies.signal,
|
||||
);
|
||||
@@ -555,8 +557,21 @@ export function generationTaskMaterializedUrls(task: GenerationTask): string[] {
|
||||
return (task.outputs || []).flatMap((output) => {
|
||||
const asset = output.materializedAssetId ? assets.find((candidate) => candidate.id === output.materializedAssetId) : undefined;
|
||||
if (!asset) return [];
|
||||
if (asset.kind === "image") return [asset.data.dataUrl || asset.coverUrl];
|
||||
if (asset.kind === "video" || asset.kind === "audio") return [asset.data.url];
|
||||
if (asset.kind === "image") return [asset.data.dataUrl || asset.coverUrl].filter(Boolean);
|
||||
if (asset.kind === "video" || asset.kind === "audio") return [asset.data.url].filter(Boolean);
|
||||
return [];
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 返回生成结果的稳定素材定位符。对于远程资源是 resource:<id>,对于本地降级素材是
|
||||
* 本地 IndexedDB storageKey。它可以持久化到历史记录,不能被短时效签名 URL 替代。
|
||||
*/
|
||||
export function generationTaskMaterializedStorageKeys(task: GenerationTask): string[] {
|
||||
const assets = useAssetStore.getState().assets;
|
||||
return (task.outputs || []).flatMap((output) => {
|
||||
const asset = output.materializedAssetId ? assets.find((candidate) => candidate.id === output.materializedAssetId) : undefined;
|
||||
if (!asset || (asset.kind !== "image" && asset.kind !== "video" && asset.kind !== "audio")) return [];
|
||||
return asset.data.storageKey ? [asset.data.storageKey] : [];
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user