testPubSub.js

import BosBase from "bosbase";
import * as websocket from "ws";

const baseUrl = process.env.BOSBASE_BASE_URL ?? "http://127.0.0.1:8080";
const authEmail =
  process.env.BOSBASE_EMAIL ??
  process.env.BOSBASE_SUPERUSER_EMAIL ??
  "try@bosbase.com";
const authPassword =
  process.env.BOSBASE_PASSWORD ??
  process.env.BOSBASE_SUPERUSER_PASSWORD ??
  "bosbasepass";

const WebSocket = websocket.WebSocket ?? websocket.default ?? websocket;
global.WebSocket = WebSocket;

function waitForMessage(buffer, label, timeoutMs = 8000) {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(
      () => reject(new Error(`${label} timed out after ${timeoutMs}ms`)),
      timeoutMs
    );

    const poll = () => {
      if (buffer.length > 0) {
        clearTimeout(timer);
        resolve(buffer.shift());
        return;
      }
      setTimeout(poll, 50);
    };

    poll();
  });
}

async function runPublishSubscribe(pb) {
  console.log("\n[INFO] Test 1: publish/subscribe roundtrip...");

  const topic = `test/pubsub/basic-${Date.now()}`;
  const messages = [];

  console.log(`[INFO] Subscribing to topic "${topic}"...`);
  const unsubscribe = await pb.pubsub.subscribe(topic, (msg) => {
    messages.push(msg);
  });

  try {
    const payload = { text: "hello from pubsub", ts: Date.now() };
    console.log("[INFO] Publishing basic message...");
    const ack = await pb.pubsub.publish(topic, payload);
    if (!ack?.id || ack.topic !== topic) {
      throw new Error("Publish ack is missing expected fields");
    }
    console.log("[SUCCESS] Publish ack:", ack);

    const received = await waitForMessage(
      messages,
      "basic pubsub message",
      8000
    );
    if (!received?.data || received.data.text !== payload.text) {
      throw new Error("Received pubsub message payload does not match");
    }
    console.log("[SUCCESS] Received pubsub message:", received);
  } finally {
    await unsubscribe();
    console.log(`[INFO] Unsubscribed from "${topic}"`);
  }
}

async function runRealtimeHelpers(pb) {
  console.log("\n[INFO] Test 2: realtimeSubscribe/realtimePublish helpers...");

  const topic = `test/pubsub/realtime-${Date.now()}`;
  const realtimeMessages = [];

  console.log(`[INFO] realtimeSubscribe on topic "${topic}"...`);
  const unsubscribe = await pb.pubsub.realtimeSubscribe(topic, (msg) => {
    realtimeMessages.push(msg);
  });

  try {
    const ref = `ref-${Date.now()}`;
    const payload = { user: "cli-runner", action: "join" };
    console.log("[INFO] realtimePublish event \"join\"...");
    const ack = await pb.pubsub.realtimePublish(topic, "join", payload, ref);
    if (!ack?.id || ack.topic !== topic) {
      throw new Error("Realtime publish ack is missing expected fields");
    }
    console.log("[SUCCESS] Realtime publish ack:", ack);

    const received = await waitForMessage(
      realtimeMessages,
      "realtime pubsub message",
      8000
    );
    if (received?.event !== "join") {
      throw new Error("Realtime message event did not match \"join\"");
    }
    if (received?.ref !== ref) {
      throw new Error("Realtime message ref did not match");
    }
    if (received?.payload?.action !== payload.action) {
      throw new Error("Realtime message payload did not match");
    }
    console.log("[SUCCESS] Received realtime message:", received);
  } finally {
    await unsubscribe();
    console.log(`[INFO] Unsubscribed from "${topic}"`);
  }
}

async function main() {
  try {
    console.log("[INFO] Starting Pub/Sub tests...");
    console.log(`[INFO] Base URL: ${baseUrl}`);

    const pb = new BosBase(baseUrl);

    console.log("[INFO] Authenticating as superuser...");
    await pb
      .collection("_superusers")
      .authWithPassword(authEmail, authPassword);
    console.log("[SUCCESS] Authenticated");

    await runPublishSubscribe(pb);
    await runRealtimeHelpers(pb);

    await pb.pubsub.disconnect();
    console.log("\n========== Pub/Sub tests completed ==========");
  } catch (error) {
    console.error("[ERROR] Pub/Sub test failed:");
    if (error?.response) {
      console.error("Status:", error.response.status);
      console.error("Data:", JSON.stringify(error.response.data, null, 2));
      if (error.response.data?.message) {
        console.error("Message:", error.response.data.message);
      }
    } else {
      console.error(error);
    }
    process.exit(1);
  }
}

main();