From 4fb2e47ca99d581e65b71f8ade12b835a1ab5d9e Mon Sep 17 00:00:00 2001 From: Theo Zourzouvillys Date: Sun, 8 Mar 2026 15:38:58 -0700 Subject: [PATCH] Add missing server features: exclusion filters, decoded values, query, health, replication status Catch up with lplex Go server changes since the initial TypeScript client. --- packages/lplex/src/client.ts | 79 +++++++++- packages/lplex/src/index.ts | 5 + packages/lplex/src/types.ts | 44 ++++++ packages/lplex/test/client.test.ts | 241 +++++++++++++++++++++++++++++ 4 files changed, 368 insertions(+), 1 deletion(-) diff --git a/packages/lplex/src/client.ts b/packages/lplex/src/client.ts index 5330cf3..1e4d181 100644 --- a/packages/lplex/src/client.ts +++ b/packages/lplex/src/client.ts @@ -2,10 +2,15 @@ import { HttpError } from "./errors.js"; import { Session } from "./session.js"; import { parseSSE } from "./sse.js"; import type { + DecodedDeviceValues, Device, DeviceValues, Event, Filter, + Frame, + HealthStatus, + QueryParams, + ReplicationStatus, SendParams, SessionConfig, SessionInfo, @@ -83,6 +88,24 @@ export class Client { return parseSSE(resp.body); } + /** Fetch the last-seen decoded values for each (device, PGN) pair. */ + async decodedValues( + filter?: Filter, + signal?: AbortSignal, + ): Promise { + let url = `${this.#baseURL}/values/decoded`; + const qs = filterToQueryString(filter); + if (qs) url += `?${qs}`; + 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; + } + /** Transmit a CAN frame through the server. */ async send(params: SendParams, signal?: AbortSignal): Promise { const url = `${this.#baseURL}/send`; @@ -99,6 +122,53 @@ 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. + */ + async query(params: QueryParams, signal?: AbortSignal): Promise { + const url = `${this.#baseURL}/query`; + const resp = await this.#fetch(url, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(params), + signal, + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new HttpError("POST", url, resp.status, body); + } + + return resp.json() as Promise; + } + + /** Check server health. */ + async health(signal?: AbortSignal): Promise { + 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; + } + + /** Fetch boat-side replication status (only available when replication is configured). */ + async replicationStatus(signal?: AbortSignal): Promise { + const url = `${this.#baseURL}/replication/status`; + 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; + } + /** Create or reconnect a buffered session on the server. */ async createSession( config: SessionConfig, @@ -133,9 +203,11 @@ export class Client { function filterIsEmpty(f: Filter): boolean { return ( !f.pgn?.length && + !f.exclude_pgn?.length && !f.manufacturer?.length && !f.instance?.length && - !f.name?.length + !f.name?.length && + !f.exclude_name?.length ); } @@ -144,17 +216,22 @@ function filterToQueryString(f?: Filter): string { 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); return params.toString(); } function filterToJSON(f: Filter): Record { const m: Record = {}; if (f.pgn?.length) m.pgn = f.pgn; + if (f.exclude_pgn?.length) m.exclude_pgn = f.exclude_pgn; if (f.manufacturer?.length) m.manufacturer = f.manufacturer; 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; return m; } diff --git a/packages/lplex/src/index.ts b/packages/lplex/src/index.ts index 2f780af..4b56838 100644 --- a/packages/lplex/src/index.ts +++ b/packages/lplex/src/index.ts @@ -11,6 +11,11 @@ export type { Filter, PGNValue, DeviceValues, + DecodedPGNValue, + DecodedDeviceValues, + QueryParams, + HealthStatus, + ReplicationStatus, SessionConfig, SessionInfo, SendParams, diff --git a/packages/lplex/src/types.ts b/packages/lplex/src/types.ts index 67735f0..654a484 100644 --- a/packages/lplex/src/types.ts +++ b/packages/lplex/src/types.ts @@ -41,9 +41,11 @@ export type Event = */ export interface Filter { pgn?: number[]; + exclude_pgn?: number[]; manufacturer?: string[]; instance?: number[]; name?: string[]; + exclude_name?: string[]; } /** Configuration for creating a buffered session. */ @@ -89,6 +91,48 @@ export interface DeviceValues { values: PGNValue[]; } +/** A single PGN's decoded value for a device. */ +export interface DecodedPGNValue { + pgn: number; + ts: string; + data: string; + seq: number; + decoded: Record; +} + +/** Decoded values grouped by device. */ +export interface DecodedDeviceValues { + name: string; + src: number; + manufacturer?: string; + model_id?: string; + values: DecodedPGNValue[]; +} + +/** Parameters for an ISO Request query (POST /query). */ +export interface QueryParams { + pgn: number; + dst: number; + timeout?: string; +} + +/** Health check response from GET /healthz. */ +export interface HealthStatus { + status: string; +} + +/** Boat-side replication status from GET /replication/status. */ +export interface ReplicationStatus { + connected: boolean; + instance_id: string; + local_head_seq: number; + cloud_cursor: number; + holes: SeqRange[]; + live_lag: number; + backfill_remaining_seqs: number; + last_ack: string; +} + // --- Cloud types --- /** Summary of a cloud instance, returned by GET /instances. */ diff --git a/packages/lplex/test/client.test.ts b/packages/lplex/test/client.test.ts index ed8ce1b..db6c356 100644 --- a/packages/lplex/test/client.test.ts +++ b/packages/lplex/test/client.test.ts @@ -414,3 +414,244 @@ describe("Session", () => { await expect(session.ack(42)).rejects.toThrow(HttpError); }); }); + +describe("Client.decodedValues", () => { + it("fetches and returns decoded values", async () => { + const values = [ + { + name: "0x00deadbeef123456", + src: 1, + manufacturer: "Garmin", + model_id: "GPS 19x", + values: [ + { + pgn: 129025, + ts: "2026-03-04T10:00:00Z", + data: "aabbccdd", + seq: 100, + decoded: { latitude: 47.6, longitude: -122.3 }, + }, + ], + }, + ]; + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("http://localhost:8089/values/decoded"); + return jsonResponse(values); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.decodedValues(); + expect(result).toHaveLength(1); + expect(result[0].values[0].decoded.latitude).toBe(47.6); + }); + + it("encodes filter as query params", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return jsonResponse([]); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.decodedValues({ pgn: [129025] }); + const parsed = new URL(capturedURL); + expect(parsed.pathname).toBe("/values/decoded"); + expect(parsed.searchParams.getAll("pgn")).toEqual(["129025"]); + }); + + it("throws HttpError on non-200", async () => { + const mockFetch = async () => errorResponse(500, "internal error"); + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await expect(client.decodedValues()).rejects.toThrow(HttpError); + }); +}); + +describe("Client.query", () => { + it("sends an ISO request and returns the response frame", async () => { + const frame = { ...sampleFrame, pgn: 60928 }; + let capturedBody = ""; + const mockFetch = async ( + url: string | URL | Request, + init?: RequestInit, + ) => { + expect(String(url)).toBe("http://localhost:8089/query"); + capturedBody = init?.body as string; + return jsonResponse(frame); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.query({ pgn: 60928, dst: 10, timeout: "PT5S" }); + expect(result.pgn).toBe(60928); + const body = JSON.parse(capturedBody); + expect(body.pgn).toBe(60928); + expect(body.dst).toBe(10); + expect(body.timeout).toBe("PT5S"); + }); + + it("throws HttpError on 504 timeout", async () => { + const mockFetch = async () => errorResponse(504, "gateway timeout"); + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await expect(client.query({ pgn: 60928, dst: 10 })).rejects.toThrow( + HttpError, + ); + }); +}); + +describe("Client.health", () => { + it("returns health status", async () => { + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("http://localhost:8089/healthz"); + return jsonResponse({ status: "ok" }); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.health(); + expect(result.status).toBe("ok"); + }); + + it("throws HttpError when unhealthy", async () => { + const mockFetch = async () => errorResponse(503, "bus silent"); + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await expect(client.health()).rejects.toThrow(HttpError); + }); +}); + +describe("Client.replicationStatus", () => { + it("returns replication status", async () => { + const status = { + connected: true, + instance_id: "boat-001", + local_head_seq: 50000, + cloud_cursor: 49950, + holes: [{ start: 100, end: 200 }], + live_lag: 50, + backfill_remaining_seqs: 500, + last_ack: "2026-03-06T10:15:30Z", + }; + const mockFetch = async (url: string | URL | Request) => { + expect(url).toBe("http://localhost:8089/replication/status"); + return jsonResponse(status); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + const result = await client.replicationStatus(); + expect(result.connected).toBe(true); + expect(result.instance_id).toBe("boat-001"); + expect(result.holes).toHaveLength(1); + expect(result.live_lag).toBe(50); + }); + + it("throws HttpError when not configured", async () => { + const mockFetch = async () => errorResponse(404, "not found"); + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await expect(client.replicationStatus()).rejects.toThrow(HttpError); + }); +}); + +describe("Filter exclusions", () => { + it("encodes exclude_pgn in query params", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return jsonResponse([]); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.values({ exclude_pgn: [59904, 60928] }); + + const parsed = new URL(capturedURL); + expect(parsed.searchParams.getAll("exclude_pgn")).toEqual([ + "59904", + "60928", + ]); + }); + + it("encodes exclude_name in query params", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return jsonResponse([]); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.values({ exclude_name: ["0x00deadbeef123456"] }); + + const parsed = new URL(capturedURL); + expect(parsed.searchParams.getAll("exclude_name")).toEqual([ + "0x00deadbeef123456", + ]); + }); + + it("includes exclusion filters in session creation body", async () => { + let capturedBody = ""; + const mockFetch = async ( + url: string | URL | Request, + init?: RequestInit, + ) => { + if (init?.method === "PUT") { + capturedBody = init.body as string; + return jsonResponse({ + client_id: "excl-test", + seq: 1, + cursor: 0, + devices: [], + }); + } + return errorResponse(404, "not found"); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.createSession({ + clientId: "excl-test", + bufferTimeout: "PT1M", + filter: { + pgn: [129025], + exclude_pgn: [59904], + exclude_name: ["0x00deadbeef123456"], + }, + }); + + const body = JSON.parse(capturedBody); + expect(body.filter.pgn).toEqual([129025]); + expect(body.filter.exclude_pgn).toEqual([59904]); + expect(body.filter.exclude_name).toEqual(["0x00deadbeef123456"]); + }); + + it("treats only-exclusion filter as non-empty", async () => { + let capturedURL = ""; + const mockFetch = async (url: string | URL | Request) => { + capturedURL = String(url); + return jsonResponse([]); + }; + + const client = new Client("http://localhost:8089", { + fetch: mockFetch as typeof fetch, + }); + await client.values({ exclude_pgn: [59904] }); + expect(capturedURL).toContain("exclude_pgn=59904"); + }); +});