ai-chatbot/lib/openclaw-chat/task-store.ts

import "server-only";

import {
  deleteMessagesByIds,
  getMessageById,
  getMessagesByChatId,
  saveMessages,
  updateMessageById,
} from "@/lib/db/queries";
import type { DBMessage } from "@/lib/db/schema";
import {
  appendClawdTaskProgress,
  CLAWD_TASK_KIND,
  type ClawdTaskMetadata,
  type ClawdTaskProgressPayload,
  type ClawdTaskStatus,
  normalizeClawdTaskMetadata,
} from "@/lib/openclaw-chat/task-state";

type PendingClawdTaskParams = {
  chatId: string;
  userMessageId: string;
  assistantMessageId: string;
  displayMessage: string;
  attachments: DBMessage["attachments"];
  spreadsheetAttachments: unknown[];
  skillTagId: string;
  skillTagName: string;
  processingHint: string;
};

function buildTaskMetadata(params: {
  userMessageId: string;
  assistantMessageId: string;
  status: ClawdTaskStatus;
  processingHint: string;
  skillTagId?: string | null;
  skillTagName?: string | null;
  progressEvents?: ClawdTaskProgressPayload[];
  requestId?: string | null;
  error?: string | null;
  createdAt?: number;
  updatedAt?: number;
}): ClawdTaskMetadata {
  const now = Date.now();
  return {
    composerMode: "clawd",
    source: "openclaw",
    taskKind: CLAWD_TASK_KIND,
    status: params.status,
    userMessageId: params.userMessageId,
    assistantMessageId: params.assistantMessageId,
    processingHint: params.processingHint,
    progressEvents: params.progressEvents || [],
    skillTagId: params.skillTagId || null,
    skillTagName: params.skillTagName || null,
    requestId: params.requestId || null,
    error: params.error || null,
    createdAt: params.createdAt || now,
    updatedAt: params.updatedAt || now,
  };
}

export async function persistPendingClawdTask(
  params: PendingClawdTaskParams
) {
  const existingRows = await getMessageById({ id: params.userMessageId });
  const existing = existingRows.find((row) => row.chatId === params.chatId);
  if (existing) {
    return normalizeClawdTaskMetadata(existing.videoMetadata);
  }

  const existingMessages = await getMessagesByChatId({ id: params.chatId }).catch(
    () => []
  );
  const latestTimestamp = existingMessages.reduce((latest, item) => {
    const timestamp = item.createdAt?.getTime?.() || 0;
    return Math.max(latest, timestamp);
  }, 0);
  const userCreatedAt = new Date(Math.max(Date.now(), latestTimestamp + 1));
  const assistantCreatedAt = new Date(userCreatedAt.getTime() + 1);
  const taskMetadata = buildTaskMetadata({
    userMessageId: params.userMessageId,
    assistantMessageId: params.assistantMessageId,
    status: "queued",
    processingHint: params.processingHint,
    skillTagId: params.skillTagId,
    skillTagName: params.skillTagName,
  });

  await saveMessages({
    messages: [
      {
        id: params.userMessageId,
        chatId: params.chatId,
        role: "user",
        parts: [{ type: "text", text: params.displayMessage }],
        attachments: params.attachments,
        createdAt: userCreatedAt,
        videoStorageStatus: null,
        temporaryVideoUrl: null,
        permanentVideoUrl: null,
        archivedAt: null,
        videoMetadata: {
          ...taskMetadata,
          spreadsheetAttachmentCount: params.spreadsheetAttachments.length,
          spreadsheetAttachments: params.spreadsheetAttachments,
        },
      },
      {
        id: params.assistantMessageId,
        chatId: params.chatId,
        role: "assistant",
        parts: [],
        attachments: [],
        createdAt: assistantCreatedAt,
        videoStorageStatus: null,
        temporaryVideoUrl: null,
        permanentVideoUrl: null,
        archivedAt: null,
        videoMetadata: taskMetadata,
      },
    ],
  });

  return taskMetadata;
}

async function updateClawdAssistantTask(params: {
  chatId: string;
  assistantMessageId: string;
  status?: ClawdTaskStatus;
  progress?: ClawdTaskProgressPayload;
  parts?: DBMessage["parts"];
  requestId?: string | null;
  error?: string | null;
}) {
  const rows = await getMessageById({ id: params.assistantMessageId });
  const message = rows.find((row) => row.chatId === params.chatId);
  if (!message) return null;
  const metadata = normalizeClawdTaskMetadata(message.videoMetadata);
  if (!metadata) return null;

  const progressEvents = params.progress
    ? appendClawdTaskProgress(metadata.progressEvents, params.progress)
    : metadata.progressEvents;
  const nextMetadata: ClawdTaskMetadata = {
    ...metadata,
    status: params.status || metadata.status,
    progressEvents,
    requestId:
      params.requestId !== undefined ? params.requestId : metadata.requestId,
    error: params.error !== undefined ? params.error : metadata.error,
    updatedAt: Date.now(),
  };
  await updateMessageById({
    messageId: params.assistantMessageId,
    chatId: params.chatId,
    parts: params.parts,
    videoMetadata: nextMetadata,
  });
  return nextMetadata;
}

export async function markClawdTaskRunning(params: {
  chatId: string;
  assistantMessageId: string;
}) {
  return updateClawdAssistantTask({ ...params, status: "running" });
}

export async function appendPersistedClawdTaskProgress(params: {
  chatId: string;
  assistantMessageId: string;
  progress: ClawdTaskProgressPayload;
}) {
  return updateClawdAssistantTask({
    ...params,
    status: "running",
    progress: params.progress,
  });
}

export async function markClawdTaskCompleted(params: {
  chatId: string;
  assistantMessageId: string;
  answer: string;
  requestId?: string | null;
}) {
  return updateClawdAssistantTask({
    chatId: params.chatId,
    assistantMessageId: params.assistantMessageId,
    status: "completed",
    parts: [{ type: "text", text: params.answer }],
    requestId: params.requestId,
    error: null,
  });
}

export async function markClawdTaskFailed(params: {
  chatId: string;
  assistantMessageId: string;
  error: string;
}) {
  return updateClawdAssistantTask({
    ...params,
    status: "failed",
    parts: [{ type: "text", text: params.error }],
    error: params.error,
  });
}

export async function getPersistedClawdTask(params: {
  chatId: string;
  userMessageId?: string | null;
}) {
  const messages = await getMessagesByChatId({ id: params.chatId });
  const candidates = messages
    .filter((message) => message.role === "assistant")
    .map((message) => ({
      message,
      metadata: normalizeClawdTaskMetadata(message.videoMetadata),
    }))
    .filter(
      (
        item
      ): item is { message: DBMessage; metadata: ClawdTaskMetadata } =>
        Boolean(item.metadata) &&
        (!params.userMessageId ||
          item.metadata?.userMessageId === params.userMessageId)
    )
    .sort(
      (left, right) =>
        right.message.createdAt.getTime() - left.message.createdAt.getTime()
    );
  const latest = candidates[0];
  if (!latest) return null;
  const userMessage = messages.find(
    (message) => message.id === latest.metadata.userMessageId
  );
  return {
    metadata: latest.metadata,
    assistantMessage: latest.message,
    userMessage: userMessage || null,
  };
}

export async function deletePersistedClawdTask(params: {
  chatId: string;
  userMessageId: string;
}) {
  const task = await getPersistedClawdTask(params);
  const messageIds = [params.userMessageId];
  if (task?.metadata.assistantMessageId) {
    messageIds.push(task.metadata.assistantMessageId);
  }
  return deleteMessagesByIds({ chatId: params.chatId, messageIds });
}