bridge: add in-memory /logs endpoint to broker bridge

slack-bridge/broker-bridge.mjs

@@ -51,21 +51,64 @@ const DEDUPE_TTL_MS = clampInt(
 const MAX_BACKOFF_MS = 30_000;
 const INBOX_PROTOCOL_VERSION = "2026-02-1";
 const BROKER_HEALTH_PATH = path.join(homedir(), ".pi", "agent", "broker-health.json");
+const LOG_BUFFER_MAX_LINES = 1000;
+
+const logLineBuffer = [];
 
 function ts() {
   return new Date().toISOString();
 }
 
+function formatLogArg(arg) {
+  if (typeof arg === "string") return arg;
+  if (arg instanceof Error) return arg.stack || arg.message;
+  try {
+    return JSON.stringify(arg);
+  } catch {
+    return String(arg);
+  }
+}
+
+function pushLogLine(line) {
+  const lines = String(line).split(/\r?\n/);
+  for (const rawLine of lines) {
+    const normalizedLine = rawLine.trimEnd();
+    if (!normalizedLine) continue;
+    logLineBuffer.push(normalizedLine);
+  }
+
+  const overflow = logLineBuffer.length - LOG_BUFFER_MAX_LINES;
+  if (overflow > 0) {
+    logLineBuffer.splice(0, overflow);
+  }
+}
+
+function logWithLevel(level, ...args) {
+  const timestampPrefix = `[${ts()}]`;
+  const line = [timestampPrefix, ...args.map(formatLogArg)].join(" ");
+  pushLogLine(line);
+
+  if (level === "error") {
+    console.error(timestampPrefix, ...args);
+    return;
+  }
+  if (level === "warn") {
+    console.warn(timestampPrefix, ...args);
+    return;
+  }
+  console.log(timestampPrefix, ...args);
+}
+
 function logInfo(...args) {
-  console.log(`[${ts()}]`, ...args);
+  logWithLevel("info", ...args);
 }
 
 function logError(...args) {
-  console.error(`[${ts()}]`, ...args);
+  logWithLevel("error", ...args);
 }
 
 function logWarn(...args) {
-  console.warn(`[${ts()}]`, ...args);
+  logWithLevel("warn", ...args);
 }
 
 for (const key of [
@@ -669,13 +712,37 @@ async function processPulledMessage(message) {
   return true;
 }
 
+function getLogLinesForResponse(url) {
+  const nParam = url.searchParams.get("n");
+  const filterParam = url.searchParams.get("filter");
+
+  let requestedLineCount = null;
+  if (nParam !== null) {
+    const parsedN = Number.parseInt(nParam, 10);
+    if (!Number.isFinite(parsedN) || parsedN < 1) {
+      throw new Error("n must be a positive integer");
+    }
+    requestedLineCount = Math.min(parsedN, LOG_BUFFER_MAX_LINES);
+  }
+
+  let lines = logLineBuffer.slice();
+
+  const normalizedFilter = filterParam?.trim().toLowerCase();
+  if (normalizedFilter) {
+    lines = lines.filter((line) => line.toLowerCase().includes(normalizedFilter));
+  }
+
+  if (requestedLineCount !== null) {
+    lines = lines.slice(-requestedLineCount);
+  }
+
+  return lines;
+}
+
 function startApiServer() {
   const server = createServer(async (req, res) => {
-    if (req.method !== "POST") {
-      res.writeHead(405, { "Content-Type": "application/json" });
-      res.end(JSON.stringify({ error: "Method not allowed" }));
-      return;
-    }
+    const url = new URL(req.url, `http://localhost:${API_PORT}`);
+    const pathname = url.pathname;
 
     const remoteAddr = req.socket.remoteAddress;
     if (remoteAddr !== "127.0.0.1" && remoteAddr !== "::1" && remoteAddr !== "::ffff:127.0.0.1") {
@@ -690,6 +757,31 @@ function startApiServer() {
       return;
     }
 
+    if (pathname === "/logs") {
+      if (req.method !== "GET") {
+        res.writeHead(405, { "Content-Type": "application/json" });
+        res.end(JSON.stringify({ error: "Method not allowed" }));
+        return;
+      }
+
+      try {
+        const lines = getLogLinesForResponse(url);
+        res.writeHead(200, { "Content-Type": "text/plain; charset=utf-8" });
+        res.end(lines.length > 0 ? `${lines.join("\n")}\n` : "");
+        return;
+      } catch (err) {
+        res.writeHead(400, { "Content-Type": "application/json" });
+        res.end(JSON.stringify({ error: err instanceof Error ? err.message : "invalid query params" }));
+        return;
+      }
+    }
+
+    if (req.method !== "POST") {
+      res.writeHead(405, { "Content-Type": "application/json" });
+      res.end(JSON.stringify({ error: "Method not allowed" }));
+      return;
+    }
+
     let rawApiRequestBody = "";
     for await (const chunk of req) rawApiRequestBody += chunk;
 
@@ -703,9 +795,6 @@ function startApiServer() {
     }
 
     try {
-      const url = new URL(req.url, `http://localhost:${API_PORT}`);
-      const pathname = url.pathname;
-
       if (pathname === "/send") {
         const validationError = validateSendParams(apiRequestBody);
         if (validationError) {
@@ -715,7 +804,7 @@ function startApiServer() {
         }
 
         const { channel, text, thread_ts } = apiRequestBody;
-        
+
         const result = await sendViaBroker({
           action: "chat.postMessage",
           routing: { channel, ...(thread_ts ? { thread_ts } : {}) },
@@ -767,7 +856,7 @@ function startApiServer() {
         }
 
         const { channel, timestamp, emoji } = apiRequestBody;
-        
+
         await sendViaBroker({
           action: "reactions.add",
           routing: { channel, timestamp, emoji },
@@ -780,7 +869,7 @@ function startApiServer() {
       }
 
       res.writeHead(404, { "Content-Type": "application/json" });
-      res.end(JSON.stringify({ error: "Not found. Endpoints: POST /send, POST /reply, POST /react" }));
+      res.end(JSON.stringify({ error: "Not found. Endpoints: POST /send, POST /reply, POST /react, GET /logs" }));
     } catch (err) {
       res.writeHead(500, { "Content-Type": "application/json" });
       res.end(JSON.stringify({ error: err instanceof Error ? err.message : "unknown error" }));

test/broker-bridge.integration.test.mjs

@@ -70,6 +70,108 @@ describe("broker pull bridge semi-integration", () => {
     tempDirs.length = 0;
   });
 
+  it("serves in-memory recent logs via GET /logs", async () => {
+    await sodium.ready;
+
+    const apiPort = await reserveFreePort();
+
+    const broker = createServer(async (req, res) => {
+      if (req.method === "POST" && req.url === "/api/inbox/pull") {
+        res.writeHead(200, { "Content-Type": "application/json" });
+        res.end(JSON.stringify({ ok: true, messages: [] }));
+        return;
+      }
+
+      if (req.method === "POST" && req.url === "/api/inbox/ack") {
+        res.writeHead(200, { "Content-Type": "application/json" });
+        res.end(JSON.stringify({ ok: true, acked: 0 }));
+        return;
+      }
+
+      if (req.method === "POST" && req.url === "/api/send") {
+        res.writeHead(200, { "Content-Type": "application/json" });
+        res.end(JSON.stringify({ ok: true, ts: "1234.5678" }));
+        return;
+      }
+
+      res.writeHead(404, { "Content-Type": "application/json" });
+      res.end(JSON.stringify({ ok: false, error: "not found" }));
+    });
+
+    await new Promise((resolve) => broker.listen(0, "127.0.0.1", resolve));
+    servers.push(broker);
+
+    const brokerAddress = broker.address();
+    if (!brokerAddress || typeof brokerAddress === "string") {
+      throw new Error("failed to get broker test server address");
+    }
+
+    const testFileDir = path.dirname(fileURLToPath(import.meta.url));
+    const repoRoot = path.dirname(testFileDir);
+    const bridgePath = path.join(repoRoot, "slack-bridge", "broker-bridge.mjs");
+    const bridgeCwd = path.join(repoRoot, "slack-bridge");
+
+    let bridgeStdout = "";
+    let bridgeStderr = "";
+
+    const bridge = spawn("node", [bridgePath], {
+      cwd: bridgeCwd,
+      env: {
+        ...process.env,
+        SLACK_BROKER_URL: `http://127.0.0.1:${brokerAddress.port}`,
+        SLACK_BROKER_WORKSPACE_ID: "T123BROKER",
+        SLACK_BROKER_SERVER_PRIVATE_KEY: b64(32, 11),
+        SLACK_BROKER_SERVER_PUBLIC_KEY: b64(32, 12),
+        SLACK_BROKER_SERVER_SIGNING_PRIVATE_KEY: b64(32, 13),
+        SLACK_BROKER_PUBLIC_KEY: b64(32, 14),
+        SLACK_BROKER_SIGNING_PUBLIC_KEY: b64(32, 15),
+        SLACK_BROKER_ACCESS_TOKEN: "test-broker-token",
+        SLACK_ALLOWED_USERS: "U_ALLOWED",
+        SLACK_BROKER_POLL_INTERVAL_MS: "100",
+        BRIDGE_API_PORT: String(apiPort),
+      },
+      stdio: ["ignore", "pipe", "pipe"],
+    });
+
+    bridge.stdout.on("data", (chunk) => {
+      bridgeStdout += chunk.toString();
+    });
+    bridge.stderr.on("data", (chunk) => {
+      bridgeStderr += chunk.toString();
+    });
+
+    children.push(bridge);
+
+    await waitFor(
+      () => bridgeStdout.includes("Outbound API listening"),
+      10_000,
+      50,
+      `timeout waiting for startup log; stdout=${bridgeStdout}; stderr=${bridgeStderr}`,
+    );
+
+    const allLogsResponse = await fetch(`http://127.0.0.1:${apiPort}/logs`);
+    expect(allLogsResponse.status).toBe(200);
+    expect(allLogsResponse.headers.get("content-type")).toContain("text/plain");
+    const allLogsText = await allLogsResponse.text();
+    expect(allLogsText).toContain("Outbound API listening");
+
+    const filteredLogsResponse = await fetch(`http://127.0.0.1:${apiPort}/logs?filter=outbound`);
+    expect(filteredLogsResponse.status).toBe(200);
+    const filteredLogsText = await filteredLogsResponse.text();
+    expect(filteredLogsText.toLowerCase()).toContain("outbound api listening");
+
+    const limitedLogsResponse = await fetch(`http://127.0.0.1:${apiPort}/logs?n=1`);
+    expect(limitedLogsResponse.status).toBe(200);
+    const limitedLogsText = await limitedLogsResponse.text();
+    const limitedLines = limitedLogsText.trim() ? limitedLogsText.trim().split("\n") : [];
+    expect(limitedLines.length).toBeLessThanOrEqual(1);
+
+    const invalidNResponse = await fetch(`http://127.0.0.1:${apiPort}/logs?n=0`);
+    expect(invalidNResponse.status).toBe(400);
+    const invalidNBody = await invalidNResponse.json();
+    expect(invalidNBody.error).toContain("positive integer");
+  });
+
   it("acks poison messages from broker to avoid infinite retry loops", async () => {
     await sodium.ready;