Fix ClaudeQQ event API artifact contract

2026-05-21 12:59:54 +00:00
parent 5ebef54936
commit dba4a7fd80
3 changed files with 309 additions and 3 deletions
@@ -23,4 +23,4 @@ COPY --from=build /app/unidesk-adapter.cjs ./scripts/src/server_ts/unidesk-adapt
 WORKDIR /app/scripts/src/server_ts
 RUN npm ci --omit=dev && npm cache clean --force

-CMD ["node", "dist/index.js", "--max-restarts", "0"]
+CMD ["node", "unidesk-adapter.cjs"]
@@ -1,4 +1,6 @@
 const http = require("node:http");
+const fs = require("node:fs");
+const path = require("node:path");
 const { spawn } = require("node:child_process");
 const { URL } = require("node:url");

@@ -10,6 +12,8 @@ const deployCommit = process.env.UNIDESK_DEPLOY_COMMIT || "";
 const deployRequestedCommit = process.env.UNIDESK_DEPLOY_REQUESTED_COMMIT || deployCommit;
 const deployRepo = process.env.UNIDESK_DEPLOY_REPO || "";
 const serviceId = process.env.UNIDESK_DEPLOY_SERVICE_ID || "claudeqq";
+const workspaceDir = process.env.CLAUDEQQ_WORKSPACE_DIR || "/bot_workspace";
+const stateDir = process.env.CLAUDEQQ_STATE_DIR || "/app/.state";
 const startedAt = new Date().toISOString();
 const endpoints = [
  "/health",
@@ -24,7 +28,11 @@ const endpoints = [

 const child = spawn("node", ["dist/index.js", "--max-restarts", "0"], {
  cwd: process.env.CLAUDEQQ_SERVER_CWD || process.cwd(),
-  env: process.env,
+  env: {
+    ...process.env,
+    CLAUDEQQ_HOST: upstreamHost,
+    CLAUDEQQ_PORT: String(upstreamPort),
+  },
  stdio: "inherit",
 });

@@ -51,6 +59,148 @@ function readBody(req) {
  });
 }

+function asRecord(value) {
+  return value && typeof value === "object" && !Array.isArray(value) ? value : {};
+}
+
+function numberOrUndefined(value) {
+  const numberValue = typeof value === "number" ? value : Number.parseInt(String(value || ""), 10);
+  return Number.isFinite(numberValue) ? numberValue : undefined;
+}
+
+function parseJson(value, fallback) {
+  try {
+    return JSON.parse(value);
+  } catch {
+    return fallback;
+  }
+}
+
+function clampLimit(value) {
+  const parsed = Number.parseInt(String(value || "50"), 10);
+  if (!Number.isFinite(parsed) || parsed <= 0) return 50;
+  return Math.min(parsed, 200);
+}
+
+function textFromMessage(message) {
+  if (typeof message === "string") return message;
+  if (!Array.isArray(message)) return "";
+  return message
+    .map((segment) => {
+      const data = asRecord(asRecord(segment).data);
+      if (typeof data.text === "string") return data.text;
+      return "";
+    })
+    .join("");
+}
+
+function eventFromRecord(record, fallbackIndex) {
+  const item = asRecord(record);
+  const raw = asRecord(item.raw_message || item.raw || item);
+  const messageId = numberOrUndefined(raw.message_id ?? item.message_id);
+  const time = numberOrUndefined(raw.time ?? item.time);
+  const userId = raw.user_id ?? item.user_id;
+  const groupId = raw.group_id ?? item.group_id;
+  const eventId = typeof item.id === "string" && item.id.length > 0
+    ? item.id
+    : [time || "unknown", messageId || fallbackIndex, userId || "unknown", groupId || "private"].join(":");
+
+  return {
+    id: eventId,
+    type: typeof item.type === "string" ? item.type : "message",
+    source: typeof item.source === "string" ? item.source : "messages-file",
+    timestamp: typeof item.timestamp === "string"
+      ? item.timestamp
+      : time
+        ? new Date(time * 1000).toISOString()
+        : new Date(0).toISOString(),
+    messageType: typeof raw.message_type === "string" ? raw.message_type : (groupId ? "group" : "private"),
+    userId: userId === undefined ? undefined : String(userId),
+    groupId: groupId === undefined ? undefined : String(groupId),
+    messageId,
+    text: textFromMessage(raw.message ?? item.message ?? item.text),
+    raw,
+  };
+}
+
+function readRecentEvents(searchParams) {
+  const limit = clampLimit(searchParams.get("limit"));
+  const since = searchParams.get("since");
+  const messagesFile = path.join(workspaceDir, "messages", "messages.jsonl");
+  try {
+    if (!fs.existsSync(messagesFile)) return [];
+    const lines = fs.readFileSync(messagesFile, "utf8").split(/\r?\n/).filter(Boolean);
+    const events = lines
+      .slice(-Math.max(limit * 2, limit))
+      .map((line, index) => eventFromRecord(parseJson(line, {}), index))
+      .filter((event) => !since || event.timestamp > since);
+    return events.slice(-limit).reverse();
+  } catch {
+    return [];
+  }
+}
+
+function readSubscriptions() {
+  const subscriptionsFile = path.join(stateDir, "event_subscriptions.json");
+  try {
+    if (!fs.existsSync(subscriptionsFile)) return [];
+    const parsed = parseJson(fs.readFileSync(subscriptionsFile, "utf8"), []);
+    return Array.isArray(parsed) ? parsed.map(asRecord) : [];
+  } catch {
+    return [];
+  }
+}
+
+function fallbackEventResponse(pathname, searchParams, upstream) {
+  if (pathname === "/api/events/recent") {
+    const events = readRecentEvents(searchParams);
+    return {
+      ok: true,
+      events,
+      count: events.length,
+      nextCursor: null,
+      source: "adapter-readonly-fallback",
+      degraded: true,
+      reason: "upstream-event-api-unavailable",
+      upstream: { ok: upstream.ok, status: upstream.status },
+    };
+  }
+  const subscriptions = readSubscriptions();
+  return {
+    ok: true,
+    subscriptions,
+    count: subscriptions.length,
+    source: "adapter-readonly-fallback",
+    degraded: true,
+    reason: "upstream-event-api-unavailable",
+    upstream: { ok: upstream.ok, status: upstream.status },
+  };
+}
+
+function normalizeEventApiBody(pathname, body, upstream) {
+  const record = asRecord(body);
+  if (record.ok === true) return record;
+  if (pathname === "/api/events/recent" && Array.isArray(record.events)) {
+    return { ...record, ok: true, count: record.events.length, source: "upstream", upstream: { ok: upstream.ok, status: upstream.status } };
+  }
+  if (pathname === "/api/events/subscriptions" && Array.isArray(record.subscriptions)) {
+    return { ...record, ok: true, count: record.subscriptions.length, source: "upstream", upstream: { ok: upstream.ok, status: upstream.status } };
+  }
+  return null;
+}
+
+async function eventApiRead(res, url) {
+  const upstream = await upstreamFetch(`${url.pathname}${url.search}`);
+  if (upstream.ok && typeof upstream.body === "string") {
+    const normalized = normalizeEventApiBody(url.pathname, parseJson(upstream.body, null), upstream);
+    if (normalized) {
+      json(res, 200, normalized);
+      return;
+    }
+  }
+  json(res, 200, fallbackEventResponse(url.pathname, url.searchParams, upstream));
+}
+
 function normalizePushPayload(raw) {
  const body = JSON.parse(raw.length > 0 ? raw.toString("utf8") : "{}");
  const message = typeof body.message === "string" ? body.message : "";
@@ -70,7 +220,10 @@ function upstreamFetch(path, method = "GET") {
    const req = http.request({ host: upstreamHost, port: upstreamPort, path, method, timeout: 2000 }, (res) => {
      const chunks = [];
      res.on("data", (chunk) => chunks.push(Buffer.from(chunk)));
-      res.on("end", () => resolve({ ok: (res.statusCode || 500) < 500, status: res.statusCode || 0, body: Buffer.concat(chunks).toString("utf8") }));
+      res.on("end", () => {
+        const status = res.statusCode || 0;
+        resolve({ ok: status >= 200 && status < 300, status, body: Buffer.concat(chunks).toString("utf8") });
+      });
    });
    req.on("timeout", () => req.destroy(new Error("timeout")));
    req.on("error", (error) => resolve({ ok: false, status: 0, body: String(error?.message || error) }));
@@ -140,6 +293,12 @@ const server = http.createServer(async (req, res) => {
          wsPort: statusNapcat.wsPort || 3001,
        },
        adapter: "unidesk-claudeqq-adapter",
+        eventApi: {
+          contract: "claudeqq-event-api-v1",
+          readonlyFallback: true,
+          validationPaths: ["/api/events/recent", "/api/events/subscriptions"],
+          mutationPaths: ["POST /api/events/subscriptions", "DELETE /api/events/subscriptions/{id}"],
+        },
        startedAt,
        upstream: { url: `http://${upstreamHost}:${upstreamPort}`, ok: upstreamHealth.ok, status: upstreamHealth.status },
        deploy: {
@@ -155,6 +314,10 @@ const server = http.createServer(async (req, res) => {
      await proxy({ ...req, method: "POST" }, res, "/api/send/text", body);
      return;
    }
+    if ((url.pathname === "/api/events/recent" || url.pathname === "/api/events/subscriptions") && req.method === "GET") {
+      await eventApiRead(res, url);
+      return;
+    }
    await proxy(req, res, `${url.pathname}${url.search}`, undefined);
  } catch (error) {
    json(res, 400, { ok: false, success: false, service: serviceId, error: String(error?.message || error) });