865 lines
26 KiB
TypeScript

import { after, NextResponse } from "next/server";
import { z } from "zod";
import { auth } from "@/app/(auth)/auth";
import {
getChatById,
getSystemSetting,
getUserById,
saveChat,
} from "@/lib/db/queries";
import {
buildClawdCancelKey,
clearClawdRequestCancellation,
isClawdRequestCanceled,
registerActiveClawdRequest,
unregisterActiveClawdRequest,
} from "@/lib/openclaw-chat/cancel-registry";
import {
normalizeOpenClawChatConfig,
OPENCLAW_CHAT_CONFIG_KEY,
} from "@/lib/openclaw-chat/config";
import {
getOpenClawTaskFailureMessage,
OPENCLAW_AUTH_REQUIRED_CODE,
OPENCLAW_AUTH_REQUIRED_MESSAGE,
OPENCLAW_UPGRADE_RETRY_MESSAGE,
} from "@/lib/openclaw-chat/errors";
import {
normalizeOpenClawChatMediaAttachments,
OPENCLAW_CHAT_MEDIA_MAX_ATTACHMENTS,
type OpenClawChatMediaAttachment,
} from "@/lib/openclaw-chat/media-attachments";
import {
buildOpenClawDisplayMessage,
buildOpenClawMessageWithSpreadsheetMarkdown,
buildOpenClawSpreadsheetMarkdown,
} from "@/lib/openclaw-chat/spreadsheet-markdown";
import {
appendPersistedClawdTaskProgress,
getPersistedClawdTask,
markClawdTaskCompleted,
markClawdTaskFailed,
markClawdTaskRunning,
persistPendingClawdTask,
} from "@/lib/openclaw-chat/task-store";
import { generateUUID } from "@/lib/utils";
const DEFAULT_PROCESSING_HINT =
"已收到,这个任务稍微繁重一些,处理时间可能会长一点,请耐心等候。";
export const maxDuration = 1800;
const requestSchema = z.object({
id: z.string().uuid(),
message: z.string().trim().max(8000).optional().default(""),
displayMessage: z.string().trim().max(8000).optional(),
userMessageId: z.string().uuid().optional(),
metadata: z.record(z.unknown()).optional(),
sensitiveInput: z
.discriminatedUnion("type", [
z.object({
type: z.literal("taobao_credentials"),
username: z.string().min(1).max(160),
password: z.string().min(1).max(256),
}),
z.object({
type: z.literal("taobao_verification"),
verification_code: z.string().min(4).max(12),
}),
])
.optional(),
attachments: z
.array(z.unknown())
.max(OPENCLAW_CHAT_MEDIA_MAX_ATTACHMENTS)
.optional()
.default([]),
});
type ParsedClawdChatRequest = Omit<
z.infer<typeof requestSchema>,
"attachments"
> & {
attachments: OpenClawChatMediaAttachment[];
tableFiles: File[];
};
type ClawdSpreadsheetAttachment = {
name: string;
size: number;
contentType: string;
sheetCount: number;
rowCount: number;
truncated: boolean;
};
type OpenClawBackgroundTask = {
jobId: string;
state: string;
stage: string;
phase: string;
displayText: string;
updatedAt: string;
completed: number;
total: number | null;
progressPercent: number | null;
pollEndpoint: string;
pollAfterMs: number;
maxWaitSeconds: number;
};
type ExecuteClawdTaskParams = {
assistantMessageId: string;
cancelKey: string;
chatId: string;
config: { baseUrl: string; token: string; timeoutMs: number };
conversationId: string;
departmentName: string;
metadata: Record<string, unknown> | undefined;
sensitiveInput?: z.infer<typeof requestSchema>["sensitiveInput"];
mediaAttachments: OpenClawChatMediaAttachment[];
skillTagId: string;
skillTagName: string;
spreadsheetFiles: unknown[];
spreadsheetMarkdownTruncated: boolean;
upstreamMessage: string;
userId: string;
userMessageId: string;
userName: string;
};
function buildChatTitle(message: string) {
const normalized = message.replace(/\s+/g, " ").trim();
if (!normalized) return "Claw 对话";
return normalized.length > 80 ? `${normalized.slice(0, 80)}...` : normalized;
}
function createConversationId(userId: string, chatId: string) {
return `site-user-${userId}-canvas-${chatId}`;
}
function getOpenClawUserName(user: {
id: string;
name?: string | null;
email?: string | null;
}) {
return String(user.name || user.email || user.id).trim();
}
function getMetadataString(
metadata: Record<string, unknown> | undefined,
key: string
) {
const value = metadata?.[key];
return typeof value === "string" ? value.trim() : "";
}
function isRequestFile(value: FormDataEntryValue): value is File {
return (
typeof value === "object" &&
value !== null &&
"arrayBuffer" in value &&
"name" in value &&
"size" in value
);
}
async function parseClawdChatRequest(
request: Request
): Promise<ParsedClawdChatRequest | null> {
const contentType = request.headers.get("content-type") || "";
if (!contentType.includes("multipart/form-data")) {
const parsed = requestSchema.safeParse(await request.json().catch(() => null));
if (!parsed.success) return null;
const attachments = normalizeOpenClawChatMediaAttachments(
parsed.data.attachments
);
if (attachments.length !== parsed.data.attachments.length) return null;
return { ...parsed.data, attachments, tableFiles: [] };
}
const formData = await request.formData().catch(() => null);
if (!formData) return null;
let metadata: Record<string, unknown> | undefined;
let attachments: unknown[] = [];
let sensitiveInput: unknown;
const metadataValue = formData.get("metadata");
const attachmentsValue = formData.get("attachments");
const sensitiveInputValue = formData.get("sensitiveInput");
try {
if (typeof metadataValue === "string" && metadataValue.trim()) {
const parsedMetadata = JSON.parse(metadataValue);
if (!parsedMetadata || typeof parsedMetadata !== "object") return null;
metadata = parsedMetadata as Record<string, unknown>;
}
if (typeof attachmentsValue === "string" && attachmentsValue.trim()) {
const parsedAttachments = JSON.parse(attachmentsValue);
if (!Array.isArray(parsedAttachments)) return null;
attachments = parsedAttachments;
}
if (typeof sensitiveInputValue === "string" && sensitiveInputValue.trim()) {
sensitiveInput = JSON.parse(sensitiveInputValue);
}
} catch {
return null;
}
const parsed = requestSchema.safeParse({
id: formData.get("id"),
message: formData.get("message") || "",
displayMessage: formData.get("displayMessage") || undefined,
userMessageId: formData.get("userMessageId") || undefined,
metadata,
attachments,
sensitiveInput,
});
if (!parsed.success) return null;
const normalizedAttachments = normalizeOpenClawChatMediaAttachments(
parsed.data.attachments
);
if (normalizedAttachments.length !== parsed.data.attachments.length) {
return null;
}
return {
...parsed.data,
attachments: normalizedAttachments,
tableFiles: formData
.getAll("tableFiles")
.filter((entry): entry is File => isRequestFile(entry) && entry.size > 0),
};
}
async function readOpenClawConfig() {
const setting = await getSystemSetting({
key: OPENCLAW_CHAT_CONFIG_KEY,
}).catch(() => null);
return normalizeOpenClawChatConfig(setting?.value);
}
function parseSseBlock(block: string) {
let event = "message";
const dataLines: string[] = [];
for (const line of block.split(/\r?\n/)) {
if (line.startsWith("event:")) event = line.slice(6).trim() || "message";
if (line.startsWith("data:")) dataLines.push(line.slice(5).trimStart());
}
if (dataLines.length === 0) return null;
try {
return { event, data: JSON.parse(dataLines.join("\n")) as unknown };
} catch {
return null;
}
}
function parseOpenClawBackgroundTask(
payload: unknown
): OpenClawBackgroundTask | null {
if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
return null;
}
const rawPayload = payload as Record<string, unknown>;
const rawTask =
rawPayload.background_task &&
typeof rawPayload.background_task === "object" &&
!Array.isArray(rawPayload.background_task)
? (rawPayload.background_task as Record<string, unknown>)
: null;
if (!rawTask) return null;
const jobId = String(rawTask.job_id || "").trim();
const state = String(rawTask.state || "").trim();
if (!jobId || !state) return null;
const completed = Math.max(0, Number(rawTask.completed) || 0);
const parsedTotal = Number(rawTask.total);
const total =
Number.isFinite(parsedTotal) && parsedTotal > 0 ? parsedTotal : null;
const parsedPercent = Number(rawTask.progress_percent);
return {
jobId,
state,
stage: String(rawTask.stage || "").trim(),
phase: String(rawTask.phase || "").trim(),
displayText: String(
rawTask.display_text || rawTask.phase || rawPayload.answer || ""
).trim(),
updatedAt: String(rawTask.updated_at || "").trim(),
completed,
total,
progressPercent:
rawTask.progress_percent != null && Number.isFinite(parsedPercent)
? Math.max(0, Math.min(100, parsedPercent))
: total
? Math.round((completed * 100) / total)
: null,
pollEndpoint: String(rawTask.poll_endpoint || "").trim(),
pollAfterMs: Math.max(
800,
Math.min(10_000, Number(rawTask.poll_after_ms) || 1500)
),
maxWaitSeconds: Math.max(
1,
Math.min(25, Number(rawTask.max_wait_seconds) || 20)
),
};
}
function buildOpenClawTaskProgress(task: OpenClawBackgroundTask) {
const message =
task.displayText ||
task.phase ||
(task.state === "processing" ? "任务正在处理" : "任务状态已更新");
return {
step_code: task.stage || "background_task",
step_state:
task.state === "failed"
? "failed"
: task.state === "processing"
? "started"
: "completed",
message,
display: message,
completed: task.completed,
...(task.total != null ? { total: task.total } : {}),
...(task.progressPercent != null
? { progress_percent: task.progressPercent }
: {}),
stage: task.stage,
state: task.state,
updated_at: task.updatedAt,
};
}
function buildOpenClawTaskStatusUrl(params: {
baseUrl: string;
task: OpenClawBackgroundTask;
userId: string;
conversationId: string;
}) {
const base = new URL(params.baseUrl);
const endpoint = params.task.pollEndpoint.startsWith("/")
? params.task.pollEndpoint
: "/api/chat/task-status";
const url = new URL(endpoint, base.origin);
url.searchParams.set("job_id", params.task.jobId);
url.searchParams.set("user_id", params.userId);
url.searchParams.set("conversation_id", params.conversationId);
url.searchParams.set("wait_seconds", String(params.task.maxWaitSeconds));
if (params.task.updatedAt) {
url.searchParams.set("known_updated_at", params.task.updatedAt);
}
return url;
}
function getAnswer(payload: unknown) {
if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
return "";
}
return String((payload as Record<string, unknown>).answer || "").trim();
}
function getRequestId(payload: unknown) {
if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
return null;
}
const value = (payload as Record<string, unknown>).request_id;
return value ? String(value) : null;
}
async function readOpenClawSse(params: {
response: Response;
chatId: string;
assistantMessageId: string;
}) {
const reader = params.response.body?.getReader();
if (!reader) throw new Error("OpenClaw stream is unavailable");
const decoder = new TextDecoder();
let buffer = "";
let finalPayload: Record<string, unknown> | null = null;
const processBlock = async (block: string) => {
const event = parseSseBlock(block);
if (!event) return;
if (event.event === "progress") {
if (event.data && typeof event.data === "object") {
await appendPersistedClawdTaskProgress({
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
progress: event.data as Record<string, unknown>,
});
}
return;
}
if (
event.event === "final" &&
event.data &&
typeof event.data === "object" &&
!Array.isArray(event.data)
) {
finalPayload = event.data as Record<string, unknown>;
}
if (event.event === "error") {
throw new Error("OpenClaw stream returned an error");
}
};
while (true) {
const { done, value } = await reader.read();
buffer += decoder.decode(value || new Uint8Array(), { stream: !done });
const blocks = buffer.split(/\r?\n\r?\n/);
buffer = blocks.pop() || "";
for (const block of blocks) {
await processBlock(block);
}
if (done) break;
}
if (buffer.trim()) await processBlock(buffer);
if (!finalPayload) throw new Error("OpenClaw stream ended without a result");
return finalPayload;
}
async function pollOpenClawBackgroundTask(params: {
config: ExecuteClawdTaskParams["config"];
initialPayload: Record<string, unknown>;
initialTask: OpenClawBackgroundTask;
controller: AbortController;
chatId: string;
assistantMessageId: string;
userId: string;
conversationId: string;
}) {
let task = params.initialTask;
let finalPayload = params.initialPayload;
let failureCount = 0;
await appendPersistedClawdTaskProgress({
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
progress: buildOpenClawTaskProgress(task),
});
while (task.state === "processing") {
if (!task.pollEndpoint) {
throw new Error("OpenClaw background task has no status endpoint");
}
try {
const response = await fetch(
buildOpenClawTaskStatusUrl({
baseUrl: params.config.baseUrl,
task,
userId: params.userId,
conversationId: params.conversationId,
}),
{
headers: {
Authorization: `Bearer ${params.config.token}`,
Accept: "application/json",
},
cache: "no-store",
signal: params.controller.signal,
}
);
if (!response.ok) {
throw new Error(`OpenClaw task status failed: ${response.status}`);
}
const payload = (await response.json().catch(() => null)) as
| Record<string, unknown>
| null;
const nextTask = parseOpenClawBackgroundTask(payload);
if (!payload || !nextTask) {
throw new Error("OpenClaw task status payload is invalid");
}
failureCount = 0;
finalPayload = {
...payload,
request_id:
payload.request_id || params.initialPayload.request_id || null,
};
task = nextTask;
await appendPersistedClawdTaskProgress({
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
progress: buildOpenClawTaskProgress(task),
});
} catch (error) {
if (params.controller.signal.aborted) throw error;
failureCount += 1;
if (failureCount >= 5) throw error;
await appendPersistedClawdTaskProgress({
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
progress: {
step_code: task.stage || "background_task",
step_state: "waiting",
message: "任务仍在运行,正在重新连接进度服务",
display: "任务仍在运行,正在重新连接进度服务",
},
});
await new Promise((resolve) => setTimeout(resolve, task.pollAfterMs));
}
}
return finalPayload;
}
async function executeClawdTask(params: ExecuteClawdTaskParams) {
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), params.config.timeoutMs);
registerActiveClawdRequest(
params.cancelKey,
controller,
Math.max(params.config.timeoutMs, 10 * 60 * 1000)
);
try {
if (isClawdRequestCanceled(params.cancelKey)) return;
await markClawdTaskRunning({
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
});
const isSensitiveSubmission = Boolean(params.sensitiveInput);
const upstreamBody = JSON.stringify({
conversation_id: params.conversationId,
user_id: params.userId,
user_name: params.userName,
username: params.userName,
department: params.departmentName,
department_name: params.departmentName,
user_department: params.departmentName,
skill_tag_id: params.skillTagId || undefined,
skill_tag_name: params.skillTagName || undefined,
message: params.upstreamMessage,
attachments:
params.mediaAttachments.length > 0
? params.mediaAttachments
: undefined,
metadata: {
source: "ai_chatbot_canvas",
page: "/canvas",
tenant: "default",
chat_id: params.chatId,
user_id: params.userId,
user_name: params.userName,
username: params.userName,
department: params.departmentName,
department_name: params.departmentName,
user_department: params.departmentName,
skill_tag_id: params.skillTagId || undefined,
skill_tag_name: params.skillTagName || undefined,
media_attachments:
params.mediaAttachments.length > 0
? params.mediaAttachments
: undefined,
spreadsheet_files:
params.spreadsheetFiles.length > 0
? params.spreadsheetFiles
: undefined,
spreadsheet_markdown_truncated:
params.spreadsheetMarkdownTruncated || undefined,
...params.metadata,
},
stream_progress: !isSensitiveSubmission,
sensitive_input: params.sensitiveInput,
});
const response = await fetch(`${params.config.baseUrl}/api/chat`, {
method: "POST",
headers: {
Authorization: `Bearer ${params.config.token}`,
"Content-Type": "application/json",
Accept: isSensitiveSubmission ? "application/json" : "text/event-stream",
},
body: upstreamBody,
signal: controller.signal,
cache: "no-store",
});
if (!response.ok) {
throw new Error(`OpenClaw request failed: ${response.status}`);
}
const contentType = response.headers.get("content-type") || "";
let payload = contentType.includes("text/event-stream")
? await readOpenClawSse({
response,
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
})
: ((await response.json().catch(() => null)) as Record<
string,
unknown
> | null);
if (!payload) throw new Error("OpenClaw returned an invalid response");
const backgroundTask = parseOpenClawBackgroundTask(payload);
if (backgroundTask?.state === "processing") {
payload = await pollOpenClawBackgroundTask({
config: params.config,
initialPayload: payload,
initialTask: backgroundTask,
controller,
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
userId: params.userId,
conversationId: params.conversationId,
});
}
if (isClawdRequestCanceled(params.cancelKey)) return;
const answer = getAnswer(payload);
if (!answer) throw new Error("OpenClaw returned an empty answer");
await markClawdTaskCompleted({
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
answer,
requestId: getRequestId(payload),
});
clearClawdRequestCancellation(params.cancelKey);
} catch (error) {
if (isClawdRequestCanceled(params.cancelKey)) return;
console.error("[ClawdChat API] Durable task failed:", error);
await markClawdTaskFailed({
chatId: params.chatId,
assistantMessageId: params.assistantMessageId,
error: getOpenClawTaskFailureMessage(error),
}).catch(() => null);
} finally {
if (params.sensitiveInput?.type === "taobao_credentials") {
params.sensitiveInput.username = "";
params.sensitiveInput.password = "";
} else if (params.sensitiveInput?.type === "taobao_verification") {
params.sensitiveInput.verification_code = "";
}
params.sensitiveInput = undefined;
clearTimeout(timeout);
unregisterActiveClawdRequest(params.cancelKey);
}
}
function buildTaskResponse(params: {
chatId: string;
userMessageId: string;
assistantMessageId: string;
status: string;
processingHint: string;
}) {
const statusUrl = `/api/canvas/clawd-chat/status?id=${encodeURIComponent(
params.chatId
)}&userMessageId=${encodeURIComponent(params.userMessageId)}`;
return {
accepted: true,
task: {
chatId: params.chatId,
userMessageId: params.userMessageId,
assistantMessageId: params.assistantMessageId,
status: params.status,
processingHint: params.processingHint,
statusUrl,
},
};
}
export async function POST(request: Request) {
const session = await auth().catch(() => null);
if (!session?.user?.id) {
return NextResponse.json(
{
error: OPENCLAW_AUTH_REQUIRED_MESSAGE,
code: OPENCLAW_AUTH_REQUIRED_CODE,
},
{ status: 401 }
);
}
const parsed = await parseClawdChatRequest(request);
if (!parsed) {
return NextResponse.json({ error: "请求参数不正确" }, { status: 400 });
}
if (
!parsed.message &&
parsed.tableFiles.length === 0 &&
parsed.attachments.length === 0
) {
return NextResponse.json(
{ error: "请输入问题或上传附件" },
{ status: 400 }
);
}
const config = await readOpenClawConfig();
if (!config.enabled || !config.baseUrl || !config.token) {
return NextResponse.json(
{ error: OPENCLAW_UPGRADE_RETRY_MESSAGE },
{ status: 503 }
);
}
const existingChat = await getChatById({ id: parsed.id }).catch(() => null);
if (existingChat && existingChat.userId !== session.user.id) {
return NextResponse.json({ error: "Forbidden" }, { status: 403 });
}
const userMessageId = parsed.userMessageId || generateUUID();
const existingTask = await getPersistedClawdTask({
chatId: parsed.id,
userMessageId,
}).catch(() => null);
if (existingTask) {
return NextResponse.json(
buildTaskResponse({
chatId: parsed.id,
userMessageId,
assistantMessageId: existingTask.metadata.assistantMessageId,
status: existingTask.metadata.status,
processingHint:
existingTask.metadata.processingHint || DEFAULT_PROCESSING_HINT,
}),
{ status: 202, headers: { "Retry-After": "2" } }
);
}
let spreadsheetMarkdownResult: Awaited<
ReturnType<typeof buildOpenClawSpreadsheetMarkdown>
> | null = null;
if (parsed.tableFiles.length > 0) {
try {
spreadsheetMarkdownResult = await buildOpenClawSpreadsheetMarkdown(
parsed.tableFiles
);
} catch (error) {
return NextResponse.json(
{
error:
error instanceof Error ? error.message : "表格解析失败,请检查文件格式",
},
{ status: 400 }
);
}
}
const displayMessage = buildOpenClawDisplayMessage({
message: parsed.displayMessage || parsed.message,
files: spreadsheetMarkdownResult?.files || [],
});
const upstreamMessage = buildOpenClawMessageWithSpreadsheetMarkdown({
message: parsed.message,
spreadsheetMarkdown: spreadsheetMarkdownResult?.markdown || "",
});
const spreadsheetAttachments: ClawdSpreadsheetAttachment[] = (
spreadsheetMarkdownResult?.files || []
).map((summary, index) => {
const sourceFile = parsed.tableFiles[index];
return {
name: summary.fileName,
size: sourceFile?.size || 0,
contentType:
sourceFile?.type ||
(summary.fileName.toLowerCase().endsWith(".xlsx")
? "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
: "text/csv"),
sheetCount: summary.sheetCount,
rowCount: summary.rowCount,
truncated: summary.truncated,
};
});
if (!existingChat) {
await saveChat({
id: parsed.id,
userId: session.user.id,
title: buildChatTitle(displayMessage),
visibility: "private",
});
}
const assistantMessageId = generateUUID();
const skillTagId = getMetadataString(parsed.metadata, "skillTagId");
const skillTagName = getMetadataString(parsed.metadata, "skillTagName");
const processingHint =
getMetadataString(parsed.metadata, "processingHint") ||
DEFAULT_PROCESSING_HINT;
const messageAttachments = parsed.attachments.map((attachment) => ({
name:
attachment.name ||
(attachment.type === "video"
? "视频附件"
: attachment.type === "file"
? "文件夹压缩包.zip"
: "图片附件"),
url: attachment.url,
contentType:
attachment.content_type ||
(attachment.type === "video"
? "video/mp4"
: attachment.type === "file"
? "application/zip"
: "image/jpeg"),
}));
await persistPendingClawdTask({
chatId: parsed.id,
userMessageId,
assistantMessageId,
displayMessage,
attachments: messageAttachments,
spreadsheetAttachments,
skillTagId,
skillTagName,
processingHint,
});
const userProfile = await getUserById({ id: session.user.id }).catch(
() => null
);
const userName = getOpenClawUserName({
id: session.user.id,
name: userProfile?.name ?? session.user.name,
email: userProfile?.email ?? session.user.email,
});
const cancelKey = buildClawdCancelKey({
userId: session.user.id,
chatId: parsed.id,
userMessageId,
});
after(async () => {
await executeClawdTask({
assistantMessageId,
cancelKey,
chatId: parsed.id,
config,
conversationId: createConversationId(session.user.id, parsed.id),
departmentName: String(userProfile?.department || "").trim(),
metadata: parsed.metadata,
sensitiveInput: parsed.sensitiveInput,
mediaAttachments: parsed.attachments,
skillTagId,
skillTagName,
spreadsheetFiles: spreadsheetMarkdownResult?.files || [],
spreadsheetMarkdownTruncated:
spreadsheetMarkdownResult?.truncated || false,
upstreamMessage,
userId: session.user.id,
userMessageId,
userName,
});
});
const responseBody = buildTaskResponse({
chatId: parsed.id,
userMessageId,
assistantMessageId,
status: "queued",
processingHint,
});
return NextResponse.json(responseBody, {
status: 202,
headers: {
"Cache-Control": "no-store, max-age=0",
Location: responseBody.task.statusUrl,
"Retry-After": "2",
},
});
}