Add full boat and cloud server API coverage

CLAUDE.md

@@ -31,15 +31,17 @@ npx tsx src/main.ts --server http://inuc1.local:8089
 
 | File | Owns |
 |---|---|
-| `src/types.ts` | `Frame`, `Device`, `Event` (discriminated union), `Filter`, `PGNValue`, `DeviceValues`, `SessionConfig`, `SessionInfo`, `SendParams` |
+| `src/types.ts` | `Frame`, `Device`, `DeviceRemoved`, `Event` (discriminated union), `Filter`, `SubscribeOptions`, `PGNValue`, `DeviceValues`, `DecodedPGNValue`, `DecodedDeviceValues`, `QueryParams`, `HistoryParams`, `BrokerHealth`, `ReplicationHealth`, `HealthStatus`, `ReplicationStatus`, `SessionConfig`, `SessionInfo`, `SendParams`, `InstanceSummary`, `InstanceStatus`, `SeqRange`, `ReplicationEvent`, `ReplicationEventType` |
 | `src/errors.ts` | `LplexError`, `HttpError` |
-| `src/sse.ts` | `parseSSE` async generator: reads `data:` lines from `ReadableStream<Uint8Array>`, yields `Event` objects |
-| `src/client.ts` | `Client` class: `devices()`, `values()`, `subscribe()`, `send()`, `createSession()`. Injectable `fetch`. |
+| `src/sse.ts` | `parseSSE` async generator: reads `data:` lines from `ReadableStream<Uint8Array>`, yields `Event` objects (frame, device, device_removed) |
+| `src/client.ts` | `Client` class: `devices()`, `values()`, `decodedValues()`, `subscribe()`, `send()`, `query()`, `history()`, `health()`, `liveness()`, `readiness()`, `replicationStatus()`, `createSession()`. Injectable `fetch`. |
+| `src/cloud.ts` | `CloudClient` class: `client()`, `instances()`, `status()`, `replicationEvents()`, `health()`, `liveness()`, `readiness()`. |
 | `src/session.ts` | `Session` class: `subscribe()`, `ack()`, `info`, `lastAckedSeq` |
 | `src/index.ts` | Barrel exports |
 | `README.md` | npm package page README |
-| `test/sse.test.ts` | SSE parser unit tests (8 tests) |
-| `test/client.test.ts` | Client + Session tests with injected fetch (14 tests) |
+| `test/sse.test.ts` | SSE parser unit tests (9 tests) |
+| `test/client.test.ts` | Client + Session tests with injected fetch (40 tests) |
+| `test/cloud.test.ts` | CloudClient tests with injected fetch (12 tests) |
 
 ### packages/lplex-cli/ File Map
 

packages/lplex-cli/src/main.ts

@@ -197,6 +197,8 @@ async function runEphemeral(client: Client, devices: DeviceMap): Promise<void> {
     if (aborted) break;
     if (event.type === "device") {
       handleDevice(event.device, devices);
+    } else if (event.type === "device_removed") {
+      devices.delete(event.deviceRemoved.src);
     } else {
       writeFrame(event, devices);
     }
@@ -242,6 +244,8 @@ async function runBuffered(client: Client, devices: DeviceMap): Promise<void> {
       if (aborted) break;
       if (event.type === "device") {
         handleDevice(event.device, devices);
+      } else if (event.type === "device_removed") {
+        devices.delete(event.deviceRemoved.src);
       } else {
         lastSeq = event.frame.seq;
         writeFrame(event, devices);

packages/lplex/src/client.ts

@@ -9,11 +9,13 @@ import type {
   Filter,
   Frame,
   HealthStatus,
+  HistoryParams,
   QueryParams,
   ReplicationStatus,
   SendParams,
   SessionConfig,
   SessionInfo,
+  SubscribeOptions,
 } from "./types.js";
 
 type FetchFn = typeof globalThis.fetch;
@@ -62,18 +64,33 @@ export class Client {
   /**
    * Open an ephemeral SSE stream with optional filtering.
    * No session, no replay, no ACK.
+   *
+   * Accepts either a Filter (for backwards compatibility) or
+   * SubscribeOptions for additional control (decode, signal).
    */
   async subscribe(
-    filter?: Filter,
+    filterOrOptions?: Filter | SubscribeOptions,
     signal?: AbortSignal,
   ): Promise<AsyncIterable<Event>> {
+    let filter: Filter | undefined;
+    let decode: boolean | undefined;
+    let sig: AbortSignal | undefined = signal;
+
+    if (filterOrOptions && isSubscribeOptions(filterOrOptions)) {
+      filter = filterOrOptions.filter;
+      decode = filterOrOptions.decode;
+      sig = filterOrOptions.signal ?? sig;
+    } else {
+      filter = filterOrOptions as Filter | undefined;
+    }
+
     let url = `${this.#baseURL}/events`;
-    const qs = filterToQueryString(filter);
+    const qs = filterToQueryString(filter, decode);
     if (qs) url += `?${qs}`;
 
     const resp = await this.#fetch(url, {
       headers: { Accept: "text/event-stream" },
-      signal,
+      signal: sig,
     });
 
     if (!resp.ok) {
@@ -124,7 +141,7 @@ export class Client {
 
   /**
    * Send an ISO Request (PGN 59904) and wait for the response frame.
-   * Returns the response frame, or throws HttpError with status 504 on timeout.
+   * Returns the response frame, or throws HttpError on timeout (408).
    */
   async query(params: QueryParams, signal?: AbortSignal): Promise<Frame> {
     const url = `${this.#baseURL}/query`;
@@ -143,7 +160,32 @@ export class Client {
     return resp.json() as Promise<Frame>;
   }
 
-  /** Check server health. */
+  /**
+   * Query historical frames (requires journaling on the server).
+   * Returns an array of frames matching the query parameters.
+   */
+  async history(params: HistoryParams, signal?: AbortSignal): Promise<Frame[]> {
+    const qs = new URLSearchParams();
+    qs.set("from", params.from);
+    if (params.to) qs.set("to", params.to);
+    for (const p of params.pgn ?? []) qs.append("pgn", p.toString());
+    for (const s of params.src ?? []) qs.append("src", s.toString());
+    if (params.limit !== undefined) qs.set("limit", params.limit.toString());
+    if (params.interval) qs.set("interval", params.interval);
+    if (params.decode) qs.set("decode", "true");
+
+    const url = `${this.#baseURL}/history?${qs.toString()}`;
+    const resp = await this.#fetch(url, { signal });
+
+    if (!resp.ok) {
+      const body = await resp.text();
+      throw new HttpError("GET", url, resp.status, body);
+    }
+
+    return resp.json() as Promise<Frame[]>;
+  }
+
+  /** Check server health (GET /healthz). */
   async health(signal?: AbortSignal): Promise<HealthStatus> {
     const url = `${this.#baseURL}/healthz`;
     const resp = await this.#fetch(url, { signal });
@@ -156,6 +198,32 @@ export class Client {
     return resp.json() as Promise<HealthStatus>;
   }
 
+  /** Liveness probe (GET /livez). */
+  async liveness(signal?: AbortSignal): Promise<HealthStatus> {
+    const url = `${this.#baseURL}/livez`;
+    const resp = await this.#fetch(url, { signal });
+
+    if (!resp.ok) {
+      const body = await resp.text();
+      throw new HttpError("GET", url, resp.status, body);
+    }
+
+    return resp.json() as Promise<HealthStatus>;
+  }
+
+  /** Readiness probe (GET /readyz). */
+  async readiness(signal?: AbortSignal): Promise<HealthStatus> {
+    const url = `${this.#baseURL}/readyz`;
+    const resp = await this.#fetch(url, { signal });
+
+    if (!resp.ok) {
+      const body = await resp.text();
+      throw new HttpError("GET", url, resp.status, body);
+    }
+
+    return resp.json() as Promise<HealthStatus>;
+  }
+
   /** Fetch boat-side replication status (only available when replication is configured). */
   async replicationStatus(signal?: AbortSignal): Promise<ReplicationStatus> {
     const url = `${this.#baseURL}/replication/status`;
@@ -200,28 +268,39 @@ export class Client {
   }
 }
 
+function isSubscribeOptions(
+  obj: Filter | SubscribeOptions,
+): obj is SubscribeOptions {
+  return "decode" in obj || "signal" in obj || "filter" in obj;
+}
+
 function filterIsEmpty(f: Filter): boolean {
   return (
     !f.pgn?.length &&
     !f.exclude_pgn?.length &&
     !f.manufacturer?.length &&
     !f.instance?.length &&
     !f.name?.length &&
-    !f.exclude_name?.length
+    !f.exclude_name?.length &&
+    !f.bus?.length
   );
 }
 
-function filterToQueryString(f?: Filter): string {
-  if (!f || filterIsEmpty(f)) return "";
+function filterToQueryString(f?: Filter, decode?: boolean): string {
+  if ((!f || filterIsEmpty(f)) && !decode) return "";
 
   const params = new URLSearchParams();
-  for (const p of f.pgn ?? []) params.append("pgn", p.toString());
-  for (const p of f.exclude_pgn ?? [])
-    params.append("exclude_pgn", p.toString());
-  for (const m of f.manufacturer ?? []) params.append("manufacturer", m);
-  for (const i of f.instance ?? []) params.append("instance", i.toString());
-  for (const n of f.name ?? []) params.append("name", n);
-  for (const n of f.exclude_name ?? []) params.append("exclude_name", n);
+  if (f) {
+    for (const p of f.pgn ?? []) params.append("pgn", p.toString());
+    for (const p of f.exclude_pgn ?? [])
+      params.append("exclude_pgn", p.toString());
+    for (const m of f.manufacturer ?? []) params.append("manufacturer", m);
+    for (const i of f.instance ?? []) params.append("instance", i.toString());
+    for (const n of f.name ?? []) params.append("name", n);
+    for (const n of f.exclude_name ?? []) params.append("exclude_name", n);
+    for (const b of f.bus ?? []) params.append("bus", b);
+  }
+  if (decode) params.set("decode", "true");
   return params.toString();
 }
 
@@ -233,5 +312,6 @@ function filterToJSON(f: Filter): Record<string, unknown> {
   if (f.instance?.length) m.instance = f.instance;
   if (f.name?.length) m.name = f.name;
   if (f.exclude_name?.length) m.exclude_name = f.exclude_name;
+  if (f.bus?.length) m.bus = f.bus;
   return m;
 }

packages/lplex/src/cloud.ts

@@ -1,6 +1,7 @@
 import { Client, type ClientOptions } from "./client.js";
 import { HttpError } from "./errors.js";
 import type {
+  HealthStatus,
   InstanceStatus,
   InstanceSummary,
   ReplicationEvent,
@@ -88,4 +89,43 @@ export class CloudClient {
 
     return resp.json() as Promise<ReplicationEvent[]>;
   }
+
+  /** Check cloud server health (GET /healthz). */
+  async health(signal?: AbortSignal): Promise<HealthStatus> {
+    const url = `${this.#baseURL}/healthz`;
+    const resp = await this.#fetch(url, { signal });
+
+    if (!resp.ok) {
+      const body = await resp.text();
+      throw new HttpError("GET", url, resp.status, body);
+    }
+
+    return resp.json() as Promise<HealthStatus>;
+  }
+
+  /** Liveness probe (GET /livez). */
+  async liveness(signal?: AbortSignal): Promise<HealthStatus> {
+    const url = `${this.#baseURL}/livez`;
+    const resp = await this.#fetch(url, { signal });
+
+    if (!resp.ok) {
+      const body = await resp.text();
+      throw new HttpError("GET", url, resp.status, body);
+    }
+
+    return resp.json() as Promise<HealthStatus>;
+  }
+
+  /** Readiness probe (GET /readyz). */
+  async readiness(signal?: AbortSignal): Promise<HealthStatus> {
+    const url = `${this.#baseURL}/readyz`;
+    const resp = await this.#fetch(url, { signal });
+
+    if (!resp.ok) {
+      const body = await resp.text();
+      throw new HttpError("GET", url, resp.status, body);
+    }
+
+    return resp.json() as Promise<HealthStatus>;
+  }
 }

packages/lplex/src/index.ts

@@ -7,13 +7,18 @@ export { LplexError, HttpError } from "./errors.js";
 export type {
   Frame,
   Device,
+  DeviceRemoved,
   Event,
   Filter,
+  SubscribeOptions,
   PGNValue,
   DeviceValues,
   DecodedPGNValue,
   DecodedDeviceValues,
   QueryParams,
+  HistoryParams,
+  BrokerHealth,
+  ReplicationHealth,
   HealthStatus,
   ReplicationStatus,
   SessionConfig,

packages/lplex/src/sse.ts

@@ -1,4 +1,4 @@
-import type { Device, Event, Frame } from "./types.js";
+import type { Device, DeviceRemoved, Event, Frame } from "./types.js";
 
 /**
  * Parse an SSE stream from a fetch Response into an async iterable of Events.
@@ -67,8 +67,16 @@ export async function* parseSSE(
 }
 
 function classify(obj: Record<string, unknown>): Event | null {
-  if ("type" in obj && obj.type === "device") {
-    return { type: "device", device: obj as unknown as Device };
+  if ("type" in obj) {
+    if (obj.type === "device") {
+      return { type: "device", device: obj as unknown as Device };
+    }
+    if (obj.type === "device_removed") {
+      return {
+        type: "device_removed",
+        deviceRemoved: obj as unknown as DeviceRemoved,
+      };
+    }
   }
   if ("seq" in obj) {
     return { type: "frame", frame: obj as unknown as Frame };