@msgmesh/sdk

@msgmesh/sdk

MsgMesh 的 TypeScript SDK(Node / 瀏覽器通用,基於 fetch)。是整個前端生態的唯一 HTTP 真相源——MCP server 與面板都複用它。

快速開始

先在面板註冊帳號、簽發一把 API key(明文僅顯示一次),再:

import { MsgMesh } from "@msgmesh/sdk";

const mq = new MsgMesh({
  apiKey: process.env.MSGMESH_KEY,          // 伺服器端用長期 key
  controlPlaneUrl: "https://cp.example.com",
  gatewayUrl: "https://gw.example.com",
  realtimeUrl: "https://rt.example.com",
});

await mq.createTopic("orders");
await mq.publish("orders", { hello: 1 });
const msgs = await mq.poll("orders", { group: "g1" });

瀏覽器/不可信端:用 getToken,不要放 API key

瀏覽器裡的東西都會外洩,不要把長期 API key 放進前端。改採 token-broker:後端(持 key)呼叫 POST /v1/tokens 換一張短期 dp token,前端只拿 token;SDK 會自動快取、將過期前重取,SSE 重連時亦換新。

const mq = new MsgMesh({
  // 不放 apiKey;給一個「去我後端拿短期 token」的函式
  getToken: async () => (await fetch("/api/mm-token")).then((r) => r.json()), // { token, expires_in }
  gatewayUrl: "https://gw.example.com",
  realtimeUrl: "https://rt.example.com",
});
mq.stream("room.42", (data) => console.log(data));   // SSE;token 過期自動換新後重連
mq.streamWs("room.42", (data) => console.log(data)); // WebSocket;同介面,SDK 自管重連

即時接收兩種選擇,介面一致、皆回傳停止函式:

• stream(SSE):靠瀏覽器原生 EventSource(有原生自動重連),走 /…/sse。僅瀏覽器。
• streamWs(WebSocket):全域 WebSocket(瀏覽器原生;Node ≥ 22 內建),走 /…/ws。WebSocket 無原生重連,故由 SDK 接管:每次斷線退避 1s 重連,成功連上即重置失敗計數;連續失敗達上限(未曾連上)即停止(避免對已撤銷的憑證或持續不可用的端點無限重連),getToken 模式另在重連前換新 token。撤權若在連線中發生為 CLOSE 1008 authorization revoked(立即停);若在握手期(HTTP 401→CloseEvent 1006)則由上限收口。適合 SSE 被中間層擋掉、或已有 WS 基礎設施的場景。Node < 22 無全域 WebSocket 會拋錯,改用 subscribe(長輪詢)。

多房間(rooms)

一個 topic 內可再切多個房間(room = Kafka record key),脫鉤「房間數」與「topic 數」。分兩層:

① 路由——發佈時用 publish(topic, body, { key: roomId }) 指定房間,訂閱時傳選用 room(第四參數 opts,stream/streamWs 皆同)只收該房間:

mq.stream("chat", (data) => console.log(data), undefined, { room: "room-42" });   // 只收 room-42
mq.streamWs("chat", (data) => console.log(data), undefined, { room: "room-42" });
await mq.publish("chat", { text: "hi" }, { key: "room-42" });                     // 發到 room-42

省略 room=收該 topic 全部訊息(向後相容)。路由本身只做過濾、無強制隔離——惡意 client 可改成別人的 room 偷聽同 topic 其他房間。要真隔離看 ②。

② 隔離(平台強制)——把憑證的 capabilities 加上選用 rooms,平台即強制該憑證只能收發指定房間(逾越 403)。rooms 省略/空 = 所有房間(向後相容);非空 = 僅限這些。典型作法是後端持一把全房間金鑰,向 POST /v1/tokens 降權簽出「只准某房間」的短期 token 給前端(降權只准更窄、須為金鑰能力子集,逾越 403):

// 後端 token-broker:用全房間 key 降權鑄「只准 chat / room-42」的短期 token,回給前端當 getToken
const r = await fetch(`${controlPlaneUrl}/v1/tokens`, {
  method: "POST",
  headers: { Authorization: `Bearer ${process.env.MSGMESH_KEY}`, "Content-Type": "application/json" },
  body: JSON.stringify({
    ttl_seconds: 600,
    capabilities: [{ ops: ["subscribe", "publish"], topics: ["chat"], rooms: ["room-42"] }],
  }),
});
const { token, expires_in } = await r.json();   // 回給前端;前端以 getToken 用它連 SSE/WS

也可用 createKey("key", { capabilities: [{ ops, topics, rooms }] }) 簽一把常駐 room-scoped 鍵。平台強制點:訂閱(SSE/WS)必須帶允許集內的 ?room(不帶=想收全部房間,一樣 403);發佈的 ?key 必須 ∈ 允許集。

⚠️ room-scoped 憑證只能走即時(SSE/WS)+ 對其房間 publish;不能 poll / consume / DLQ。後者是整個 topic 的 firehose(consumer-group offset 會吃掉別房間、每房一 group = 讀取放大),無法乾淨 per-room 過濾,受限房間憑證一律 403(use realtime SSE/WS ?room=)。需要 poll/consume 時請改用不限房間的憑證。

房間隔離的安全須知(必讀)

• 隔離強度 = 你發的 token 範圍。 只有在「後端用全房間金鑰降權鑄 room-scoped token 給前端」時才有隔離。別把不限房間的憑證(全權 key、或沒有 rooms 的 token)放進前端——那樣 client 改個 room 就能看到所有房間,隔離形同虛設。
• 平台不驗「發訊者是誰」。 房間隔離管的是「能收發哪些房間」,不是「你是房裡的誰」。同一房內,任何持該房 token 的人都能在 payload 裡冒充任何 sender。要防房內冒名:後端為每個使用者各自鑄 token、並由後端戳上 / 驗證 sender,別讓前端自報身分。
• 附帶:presence(在線數)目前是 per-topic 非 per-room(只洩漏聚合數字);短期 token 為 bearer,洩漏 = 該房 ≤TTL 可用(故 TTL 短、勿記進 log)。

生產環境設定(必讀)

• 務必明確指定服務 URL:controlPlaneUrl(治理 API)、gatewayUrl(收發)、realtimeUrl(SSE/WS/presence)。 未指定時 SDK 退回本機開發預設值(http://localhost:8080/8081/8082),只適合本機; 生產環境連不上時錯誤訊息會附「請指定 controlPlaneUrl/gatewayUrl/realtimeUrl」提示。
• 憑證保管:API key 只在建立時回傳一次明文,不要寫進 repo 或日誌。瀏覽器端勿放任何 API key,改用 getToken。
• scope:發訊需 producer、收訊需 consumer、管理需 admin(通吃);又推又收/細粒度的資料面鍵可用中性 key scope(強制附 capabilities)。getToken 模式僅供資料面(收發),呼叫治理端點會 401/403。

錯誤處理

非 2xx 回應會依狀態碼拋型別化錯誤(都繼承 MsgMeshError,帶 status/code/path):

import { ValidationError, AuthError, NotFoundError, RateLimitError } from "@msgmesh/sdk";

try {
  await mq.createTopic("Bad Name!");
} catch (e) {
  if (e instanceof ValidationError) console.error("參數不合法:", e.message);
  else if (e instanceof RateLimitError) console.error("被限流,稍後重試");
  else throw e;
}

狀態碼	型別	code
400 / 422	ValidationError	validation
401 / 403	AuthError	auth
404	NotFoundError	not_found
429	RateLimitError	rate_limit
其他	MsgMeshError	server

API 一覽

• Topics:createTopic / listTopics / deleteTopic
• 收發:publish / poll / subscribe(輪詢)/ stream(SSE,瀏覽器)/ streamWs(WebSocket,瀏覽器 + Node ≥ 22)/ getPresence
• Keys:listKeys(回傳含 capabilities/name)/ createKey(可帶 scope+capabilities)/ deleteKey
• Webhooks:listWebhooks / createWebhook / deleteWebhook / reactivateWebhook
• Schemas:registerSchema / listSchemas / getLatestSchema / deleteSchema
• Functions:registerFunction / getFunction / deleteFunction(JavaScript / WASM)
• 方案:getPlan / setPlan;用量:getUsage
• 設定:getSettings / setStrictTopics(資料面 topic 閘門開關)
• 帳務(加密貨幣 PAYG 預付):getBilling / getDepositAddresses / getDeposits / getLedger / getUsageDebits / getDepositStatus
• 其他:getSnippet / getDocs / getAudit、DLQ dlqPeek / dlqReplay

註冊、超管(finance/租戶治理)走面板 session,不在本 SDK。

開發

npm test -w @msgmesh/sdk && npm run build -w @msgmesh/sdk