Blixify Server

Shared server template: MongoDB, Firebase, Auth, Upload, and real-time Pub/Sub→SSE live-stream.

SocketWrapper

Generalises the per-delivery Pub/Sub → SSE pattern for any collection.

Write → MongoWrapper.afterWrite → SocketWrapper.emit
                                        │
                              Pub/Sub topic (global bus)
                                   attributes.room
                                        │
                  ┌─────────────────────┴─────────────────────┐
             room="robots"                            room="robots-r001"
         (list-level SSE clients)              (doc-level SSE clients)

Every GAE instance publishes to the same Pub/Sub topic. Each SSE client gets its own temporary pull subscription filtered by attributes.room. A client on instance 2 receives events published by instance 1 — no Redis adapter, no sticky sessions.

Room naming

| Room | Who subscribes | What it receives | | --------------- | --------------------------------------------------------- | ------------------------------------ | | "robots" | DTW list view (bareSocketName="robots") | insert / update / delete for any doc | | "robots-r001" | DTW read/update view (bareSocketName="robots-r001") | update / delete for doc r001 only |

On every write, emit("robots", "r001", "update", payload) publishes to both rooms.

Install

yarn add @google-cloud/pubsub

GCP setup

1. Create a Pub/Sub topic — e.g. blixify-live-events-prod (separate from any existing topics).
2. Grant pubsub.topics.publish to the App Engine service account.
3. Set PUBSUB_LIVE_TOPIC=blixify-live-events-prod in .env / App Engine env vars.

Server setup

import { PubSub } from "@google-cloud/pubsub";
import { SocketWrapper } from "blixify-server/dist/apis";

// INFO: null → no-op in local dev (mirrors the getPubSubClient pattern)
const pubsub = process.env.PUBSUB_LIVE_TOPIC ? new PubSub() : null;

export const socketWrapper = new SocketWrapper(
  pubsub,
  process.env.PUBSUB_LIVE_TOPIC ?? "",
  ["robots", "vehicles"], // INFO: opt-in whitelist
  {
    checkTopicAuth: async (room, req) =>
      req.query?.bm_apiToken === process.env.SECURITY_TOKEN ||
      req.body?.bm_apiToken === process.env.SECURITY_TOKEN,
  },
);

Wire to MongoWrapper

// afterWrite is optional — when unset, MongoWrapper behaviour is unchanged
wrapper.afterWrite = (type, id, payload) =>
  socketWrapper.emit("robots", id, type, payload);
//                             ↑
//              id="" for batch ops → only collection room receives it

SSE endpoint

Next.js:

// pages/api/live/[room].ts
export const config = { api: { bodyParser: false } };
export default socketWrapper.createSSEHandler();

Express:

app.get("/api/live/:room", socketWrapper.createSSEHandler());

Snapshot on reconnect

socketWrapper.registerSnapshotProvider("robots", async () => {
  const db = await mongoPromise;
  return db.db("robots-prod").collection("robots").find({}).toArray();
});

On each new SSE connection the handler immediately sends a snapshot envelope so the client reconciles rows that drifted while offline.

SocketEnvelope

interface SocketEnvelope<T = any> {
  topic: string; // collection name
  type: "insert" | "update" | "delete" | "snapshot";
  payload: T | T[]; // snapshot = T[]
  ts: number; // Unix ms
}

DTW client usage

Set workspace.sseEndpoint (or devSettings.sseEndpoint) to the base of your /api/live route:

// workspaceDev
export const workspaceDev = {
  apiEndpoint: `${apiPrefix}/api/data`,
  assetEndpoint: storageAPI,
  assetUploadEndpoint: `${apiPrefix}/api/asset`,
  apiEndpointToken: securityToken,
  sseEndpoint: `${apiPrefix}/api/live`,   // ← add this
};

// List view — subscribes room "robots"
<DataTemplateWrapper collectionId="robots" type="list"
  bareSettings={{ bareSocketName: "robots" }}
  workspace={workspaceDev} />

// Read/update view — subscribes room "robots-{id}"
<DataTemplateWrapper collectionId="robots" type="read" id={robotId}
  bareSettings={{ bareSocketName: `robots-${robotId}` }}
  workspace={workspaceDev} />

// MRQ polling fallback (1.5 s) — no Pub/Sub needed
<DataTemplateWrapper collectionId="robots" type="list"
  bareSettings={{ barePollingInterval: 1500 }}
  workspace={workspaceDev} />

The DTW opens EventSource at {sseEndpoint}/{room}?bm_apiToken={token}. No socket.io-client package needed.

Capacitor

EventSource is a plain HTTP long-lived GET. Works on iOS and Android WebView with no extra configuration — same as any other API call. No sticky sessions or instance_affinity required on GAE.

Local dev

pubsub = null → emit() is a silent no-op. The SSE endpoint stays open and sends periodic : ping keep-alives. The app saves correctly; other devices just don't receive a live push.

Tests

yarn test:unit

Covers: whitelist gating, dual-room publish (collection + doc), batch-op single-room publish, null-pubsub no-op, SSE auth rejection, snapshot on connect, MongoWrapper.afterWrite → emit full chain.