diff --git a/.changeset/curly-poets-run.md b/.changeset/curly-poets-run.md new file mode 100644 index 00000000..53947fe4 --- /dev/null +++ b/.changeset/curly-poets-run.md @@ -0,0 +1,5 @@ +--- +"@elgato/streamdeck": minor +--- + +Add connection bridge to enable communication with the VS Code extension. diff --git a/.gitignore b/.gitignore index d34c33fd..1965213f 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,6 @@ dist/ # Shared package files packages/plugin/README.md packages/plugin/LICENSE + +# OS +.DS_Store diff --git a/packages/plugin/src/plugin/__tests__/index.test.ts b/packages/plugin/src/plugin/__tests__/index.test.ts index ceff8b40..9d567071 100644 --- a/packages/plugin/src/plugin/__tests__/index.test.ts +++ b/packages/plugin/src/plugin/__tests__/index.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest"; import { BarSubType, DeviceType, Target } from "../../api/index.js"; import { SingletonAction } from "../actions/singleton-action.js"; import { connection } from "../connection.js"; +import { bridge } from "../bridge/adapter.js"; import streamDeckAsDefaultExport, { streamDeck } from "../index.js"; import { logger } from "../logging/index.js"; @@ -10,6 +11,7 @@ vi.mock("../../common/i18n.js"); vi.mock("../logging/index.js"); vi.mock("../manifest.js"); vi.mock("../connection.js"); +vi.mock("../bridge/adapter.js"); describe("index", () => { /** @@ -47,10 +49,12 @@ describe("index", () => { it("connects", async () => { // Arrange. const spyOnConnect = vi.spyOn(connection, "connect"); + const spyOnStart = vi.spyOn(bridge, "start"); // Act, assert. await streamDeck.connect(); expect(spyOnConnect).toHaveBeenCalledTimes(1); + expect(spyOnStart).toHaveBeenCalledTimes(1); }); /** diff --git a/packages/plugin/src/plugin/bridge/__mocks__/adapter.ts b/packages/plugin/src/plugin/bridge/__mocks__/adapter.ts new file mode 100644 index 00000000..9935d7d6 --- /dev/null +++ b/packages/plugin/src/plugin/bridge/__mocks__/adapter.ts @@ -0,0 +1,5 @@ +import { vi } from "vitest"; + +export const bridge = { + start: vi.fn().mockResolvedValue(undefined), +}; diff --git a/packages/plugin/src/plugin/bridge/__tests__/adapter.test.ts b/packages/plugin/src/plugin/bridge/__tests__/adapter.test.ts new file mode 100644 index 00000000..bc95d29c --- /dev/null +++ b/packages/plugin/src/plugin/bridge/__tests__/adapter.test.ts @@ -0,0 +1,342 @@ +import type { JsonRpcRequest, JsonRpcResponse } from "@elgato/utils/rpc"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { DeviceType, type DeviceDidChange, type DidReceiveSettings, type WillAppear, type WillDisappear } from "../../../api/index.js"; + +vi.mock("../../connection.js"); +vi.mock("../../manifest.js"); + +describe("debug adapter", () => { + let adapter: Awaited["bridge"]; + let connection: Awaited["connection"]; + let sent: Array; + + beforeEach(async () => { + vi.resetModules(); + sent = []; + + ({ connection } = await import("../../connection.js")); + ({ bridge: adapter } = await import("../adapter.js")); + adapter.attachRpc(async (value) => { + sent.push(value); + }); + }); + + afterEach(() => { + vi.clearAllMocks(); + }); + + it("builds plugin state from manifest actions and visible instances", () => { + connection.emit("willAppear", createKeyWillAppear()); + + expect(adapter.getPluginState()).toEqual({ + actions: [ + { + instances: [ + { + context: "ctx_001", + controller: "Keypad", + coordinates: { + column: 2, + row: 3, + }, + device: "Device One", + isInMultiAction: false, + resources: { + thumbnail: "imgs/key.png", + }, + settings: { + count: 42, + }, + }, + ], + name: "Action One", + uuid: "com.elgato.test.key", + }, + { + instances: [], + name: "Action Two", + uuid: "com.elgato.test.dial", + }, + ], + plugin: { + name: "Test Plugin", + uuid: "com.elgato.test", + version: "1.0.0", + }, + }); + }); + + it("publishes plugin state changes for visible instances", async () => { + connection.emit("willAppear", createKeyWillAppear()); + await waitForMessages(sent, 1); + + expect(sent).toEqual([ + { + jsonrpc: "2.0", + method: "streamDeck.bridge.pluginStateChanged", + params: adapter.getPluginState(), + }, + ]); + }); + + it("preserves coordinates and multi-action state", () => { + connection.emit("willAppear", createDialWillAppear()); + + expect(adapter.getPluginState().actions[1].instances[0]?.coordinates).toEqual({ + column: 1, + row: 0, + }); + + connection.emit("willAppear", createMultiActionWillAppear()); + + const instance = adapter.getPluginState().actions[0].instances[0]; + expect(instance?.isInMultiAction).toBe(true); + expect(instance?.coordinates).toEqual({ + column: 4, + row: 2, + }); + }); + + it("updates settings, device details, and removals from connection events", async () => { + connection.emit("willAppear", createKeyWillAppear()); + await waitForMessages(sent, 1); + clearMessages(sent); + + connection.emit("didReceiveSettings", createDidReceiveSettings()); + await waitForMessages(sent, 1); + expect(adapter.getPluginState().actions[0].instances[0]?.settings).toEqual({ + count: 7, + }); + expect(adapter.getPluginState().actions[0].instances[0]?.resources).toEqual({ + thumbnail: "imgs/updated.png", + }); + expect(sent).toHaveLength(1); + + clearMessages(sent); + connection.emit("deviceDidChange", createDeviceDidChange()); + await waitForMessages(sent, 1); + expect(adapter.getPluginState().actions[0].instances[0]?.device).toBe("Renamed Device"); + expect(sent).toHaveLength(1); + + clearMessages(sent); + connection.emit("willDisappear", createWillDisappear()); + await waitForMessages(sent, 1); + expect(adapter.getPluginState().actions[0].instances).toEqual([]); + expect(sent).toHaveLength(1); + }); + + it("responds to getPluginState RPC requests", async () => { + connection.emit("willAppear", createKeyWillAppear()); + await waitForMessages(sent, 1); + clearMessages(sent); + + const handled = await adapter.receive({ + id: "request-1", + jsonrpc: "2.0", + method: "streamDeck.bridge.getPluginState", + }); + + expect(handled).toBe(true); + expect(sent).toEqual([ + { + id: "request-1", + jsonrpc: "2.0", + result: adapter.getPluginState(), + }, + ]); + }); + + it("does not start the socket transport outside debug mode", async () => { + vi.resetModules(); + vi.doMock("../../common/utils.js", async () => { + const actual = await vi.importActual("../../common/utils.js"); + + return { + ...actual, + isDebugMode: vi.fn().mockReturnValue(false), + }; + }); + vi.doMock("../socket.js", () => ({ + socket: { + start: vi.fn().mockResolvedValue(undefined), + }, + })); + + const { bridge } = await import("../adapter.js"); + const { socket } = await import("../socket.js"); + + await bridge.start(); + + expect(socket.start).not.toHaveBeenCalled(); + }); +}); + +/** + * Clears recorded JSON-RPC messages between assertions. + * @param messages Recorded messages. + */ +function clearMessages(messages: Array): void { + messages.splice(0, messages.length); +} + +/** + * Waits for the adapter's queued JSON-RPC messages to be sent. + * @param messages Recorded messages. + * @param length Expected message count. + */ +async function waitForMessages(messages: Array, length: number): Promise { + await vi.waitFor(() => { + expect(messages).toHaveLength(length); + }); +} + +/** + * Creates a device-change event for the seeded mock device. + * @returns Device-change event. + */ +function createDeviceDidChange(): DeviceDidChange { + return { + device: "DEV1", + deviceInfo: { + name: "Renamed Device", + size: { + columns: 5, + rows: 3, + }, + type: DeviceType.StreamDeckXL, + }, + event: "deviceDidChange", + }; +} + +/** + * Creates a settings update for the visible test key. + * @returns Did-receive-settings event. + */ +function createDidReceiveSettings(): DidReceiveSettings<{ count: number }> { + return { + action: "com.elgato.test.key", + context: "ctx_001", + device: "DEV1", + event: "didReceiveSettings", + payload: { + controller: "Keypad", + coordinates: { + column: 2, + row: 3, + }, + isInMultiAction: false, + resources: { + thumbnail: "imgs/updated.png", + }, + settings: { + count: 7, + }, + }, + }; +} + +/** + * Creates a visible dial instance event. + * @returns Will-appear event for a dial. + */ +function createDialWillAppear(): WillAppear<{ target: string }> { + return { + action: "com.elgato.test.dial", + context: "ctx_002", + device: "DEV1", + event: "willAppear", + payload: { + controller: "Encoder", + coordinates: { + column: 1, + row: 0, + }, + isInMultiAction: false, + resources: { + thumbnail: "imgs/dial.png", + }, + settings: { + target: "master", + }, + }, + }; +} + +/** + * Creates a visible key instance event. + * @returns Will-appear event for a key. + */ +function createKeyWillAppear(): WillAppear<{ count: number }> { + return { + action: "com.elgato.test.key", + context: "ctx_001", + device: "DEV1", + event: "willAppear", + payload: { + controller: "Keypad", + coordinates: { + column: 2, + row: 3, + }, + isInMultiAction: false, + resources: { + thumbnail: "imgs/key.png", + }, + settings: { + count: 42, + }, + }, + }; +} + +/** + * Creates a visible multi-action key instance event. + * @returns Will-appear event for a multi-action key. + */ +function createMultiActionWillAppear(): WillAppear<{ count: number }> { + return { + action: "com.elgato.test.key", + context: "ctx_003", + device: "DEV1", + event: "willAppear", + payload: { + controller: "Keypad", + coordinates: { + column: 4, + row: 2, + }, + isInMultiAction: true, + resources: {}, + settings: { + count: 99, + }, + }, + }; +} + +/** + * Creates a disappearance event for the visible test key. + * @returns Will-disappear event for a key. + */ +function createWillDisappear(): WillDisappear<{ count: number }> { + return { + action: "com.elgato.test.key", + context: "ctx_001", + device: "DEV1", + event: "willDisappear", + payload: { + controller: "Keypad", + coordinates: { + column: 2, + row: 3, + }, + isInMultiAction: false, + resources: {}, + settings: { + count: 7, + }, + }, + }; +} \ No newline at end of file diff --git a/packages/plugin/src/plugin/bridge/__tests__/socket.test.ts b/packages/plugin/src/plugin/bridge/__tests__/socket.test.ts new file mode 100644 index 00000000..ab670b0b --- /dev/null +++ b/packages/plugin/src/plugin/bridge/__tests__/socket.test.ts @@ -0,0 +1,272 @@ +import { type JsonValue, withResolvers } from "@elgato/utils"; +import type { RpcSender } from "@elgato/utils/rpc"; +import { EventEmitter } from "node:events"; +import { createConnection, type Server, type Socket } from "node:net"; +import { platform } from "node:os"; +import { access } from "node:fs/promises"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { WillAppear } from "../../../api/index.js"; +import { getPipePath } from "../pipe-path.js"; + +vi.mock("../../connection.js"); +vi.mock("../../common/utils.js", async () => { + const actual = await vi.importActual("../../common/utils.js"); + + return { + ...actual, + isDebugMode: vi.fn().mockReturnValue(true), + }; +}); +vi.mock("../../manifest.js"); +vi.mock("../../logging/index.js"); + +const pipePath = getPipePath("com.elgato.test"); + +describe("bridge socket", () => { + let bridge: Awaited["bridge"]; + let connection: Awaited["connection"]; + let socket: Awaited["socket"]; + + beforeEach(async () => { + vi.resetModules(); + ({ bridge } = await import("../adapter.js")); + ({ connection } = await import("../../connection.js")); + ({ socket } = await import("../socket.js")); + }); + + afterEach(async () => { + await socket.stop(); + vi.clearAllMocks(); + }); + + it("serves plugin state requests over the debug socket", async () => { + await bridge.start(); + const client = await connect(); + + const response = await getPluginState(client); + expect(JSON.parse(response)).toEqual({ + id: "request-1", + jsonrpc: "2.0", + result: bridge.getPluginState(), + }); + + client.destroy(); + }); + + it("forwards plugin state change notifications to the connected client", async () => { + await bridge.start(); + const client = await connect(); + await getPluginState(client); // Wait for the initial plugin state response to ensure the RPC connection is ready. + + const notification = receive(client); + connection.emit("willAppear", createKeyWillAppear()); + + expect(JSON.parse(await notification)).toEqual({ + jsonrpc: "2.0", + method: "streamDeck.bridge.pluginStateChanged", + params: bridge.getPluginState(), + }); + + client.destroy(); + }); + + it("rebinds the RPC host for an active socket when started again", async () => { + await bridge.start(); + const client = await connect(); + let send: RpcSender | undefined; + + await socket.start( + { + attachRpc: (nextSend) => { + send = nextSend; + }, + receive: async (value: JsonValue) => { + const id = getRequestId(value); + if (!id || !send) { + return false; + } + + await send({ + id, + jsonrpc: "2.0", + result: "rebound", + }); + return true; + }, + }, + "com.elgato.test", + ); + + const response = receive(client); + client.write( + `${JSON.stringify({ + id: "request-1", + jsonrpc: "2.0", + method: "test.rebound", + })}\n`, + ); + + expect(JSON.parse(await response)).toEqual({ + id: "request-1", + jsonrpc: "2.0", + result: "rebound", + }); + + client.destroy(); + }); + + it("retries listening after a startup error", async () => { + vi.resetModules(); + const listen = vi.fn(); + + vi.doMock("node:net", () => ({ + createServer: vi.fn(() => { + const server = new EventEmitter() as EventEmitter & Pick; + server.close = ((callback?: (err?: Error) => void) => { + callback?.(); + return server; + }) as Server["close"]; + server.listen = ((path: string) => { + listen(path); + const attempt = listen.mock.calls.length; + + queueMicrotask(() => { + if (attempt === 1) { + server.emit("error", new Error("address in use")); + } else { + server.emit("listening"); + } + }); + + return server; + }) as Server["listen"]; + + return server; + }), + })); + + try { + const { socket: retrySocket } = await import("../socket.js"); + const rpcHost = { + attachRpc: vi.fn(), + receive: vi.fn().mockResolvedValue(false), + }; + + await expect(retrySocket.start(rpcHost, "com.elgato.test")).rejects.toThrow("address in use"); + await expect(retrySocket.start(rpcHost, "com.elgato.test")).resolves.toBeUndefined(); + await retrySocket.stop(); + } finally { + vi.doUnmock("node:net"); + } + + expect(listen).toHaveBeenCalledTimes(2); + }); + + it("removes the socket file when stopped", async () => { + await bridge.start(); + if (platform() !== "win32") { + await expect(access(pipePath)).resolves.toBeUndefined(); + } + + await socket.stop(); + if (platform() !== "win32") { + await expect(access(pipePath)).rejects.toThrow(); + } + }); +}); + +/** + * Creates a visible key instance event. + * @returns Will-appear event for a key. + */ +function createKeyWillAppear(): WillAppear<{ count: number }> { + return { + action: "com.elgato.test.key", + context: "ctx_001", + device: "DEV1", + event: "willAppear", + payload: { + controller: "Keypad", + coordinates: { + column: 2, + row: 3, + }, + isInMultiAction: false, + resources: {}, + settings: { + count: 42, + }, + }, + }; +} + +/** + * Connects a client to the debug socket. + * @returns Connected socket client. + */ +async function connect(): Promise { + const client = createConnection(pipePath); + client.setEncoding("utf-8"); + + const opened = withResolvers(); + client.once("connect", () => opened.resolve()); + client.once("error", (err) => opened.reject(err)); + await opened.promise; + + return client; +} + +/** + * Requests plugin state over the socket client. + * @param client Socket client. + * @returns JSON-RPC response message. + */ +function getPluginState(client: Socket): Promise { + const response = receive(client); + client.write( + `${JSON.stringify({ + id: "request-1", + jsonrpc: "2.0", + method: "streamDeck.bridge.getPluginState", + })}\n`, + ); + + return response; +} + +/** + * Gets a JSON-RPC request id from a received message. + * @param value Received JSON value. + * @returns Request id when present. + */ +function getRequestId(value: JsonValue): string | undefined { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + return undefined; + } + + return typeof value.id === "string" ? value.id : undefined; +} + +/** + * Receives the next newline-delimited message from the client. + * @param client Socket client. + * @returns Received message, without its trailing newline. + */ +function receive(client: Socket): Promise { + const message = withResolvers(); + + let buffer = ""; + const onData = (chunk: string): void => { + buffer += chunk; + const newline = buffer.indexOf("\n"); + if (newline !== -1) { + client.off("data", onData); + message.resolve(buffer.slice(0, newline)); + } + }; + + client.on("data", onData); + client.once("error", (err) => message.reject(err)); + return message.promise; +} diff --git a/packages/plugin/src/plugin/bridge/adapter.ts b/packages/plugin/src/plugin/bridge/adapter.ts new file mode 100644 index 00000000..8ab29fed --- /dev/null +++ b/packages/plugin/src/plugin/bridge/adapter.ts @@ -0,0 +1,492 @@ +import type { IDisposable, JsonObject, JsonValue } from "@elgato/utils"; +import { createRpcServerClient, type RpcSender } from "@elgato/utils/rpc"; + +import type { + Coordinates, + DeviceDidChange, + DeviceDidConnect, + DidReceiveSettings, + Manifest, + Resources, + WillAppear, + WillDisappear, +} from "../../api/index.js"; +import { isDebugMode } from "../common/utils.js"; +import { connection } from "../connection.js"; +import { logger } from "../logging/index.js"; +import { getManifest } from "../manifest.js"; +import { socket } from "./socket.js"; + +/** + * Internal JSON-RPC adapter that projects connection events into serialized plugin state. + */ +class BridgeAdapter { + /** + * Tracked device names indexed by device identifier. + */ + readonly #devices = new Map(); + + /** + * Tracked visible action instances indexed by action context. + */ + readonly #instances = new Map(); + + /** + * Registered disposables associated with adapter listeners and methods. + */ + readonly #disposables: IDisposable[] = []; + + /** + * Parsed plugin manifest, when available. + */ + #manifest: Manifest | null | undefined; + + /** + * Internal RPC client/server used by the adapter. + */ + #rpc: ReturnType | undefined; + + /** + * Last serialized plugin state emitted by the adapter. + */ + #lastPluginState = ""; + + /** + * Tail of the serialized publication promise chain. + */ + #publishQueue: Promise = Promise.resolve(); + + /** + * Determines whether registration devices have been seeded into the internal device map. + */ + #seededDevices = false; + + /** + * Initializes a new bridge adapter instance. + */ + constructor() { + this.#disposables.push( + connection.disposableOn("deviceDidChange", (ev: DeviceDidChange) => { + this.#devices.set(ev.device, ev.deviceInfo.name); + this.#publishChanges(); + }), + ); + this.#disposables.push( + connection.disposableOn("deviceDidConnect", (ev: DeviceDidConnect) => { + this.#devices.set(ev.device, ev.deviceInfo.name); + this.#publishChanges(); + }), + ); + this.#disposables.push( + connection.disposableOn("didReceiveSettings", (ev: DidReceiveSettings) => { + const instance = this.#instances.get(ev.context); + if (!instance) { + return; + } + const payload = ev.payload as PayloadWithCoordinates; + + instance.controller = ev.payload.controller; + instance.deviceId = ev.device; + instance.isInMultiAction = payload.isInMultiAction ?? false; + instance.coordinates = payload.coordinates; + instance.resources = payload.resources ?? {}; + instance.settings = ev.payload.settings; + this.#publishChanges(); + }), + ); + this.#disposables.push( + connection.disposableOn("willAppear", (ev: WillAppear) => { + const payload = ev.payload as PayloadWithCoordinates; + + this.#instances.set(ev.context, { + context: ev.context, + controller: ev.payload.controller, + deviceId: ev.device, + isInMultiAction: payload.isInMultiAction ?? false, + manifestId: ev.action, + coordinates: payload.coordinates, + resources: payload.resources ?? {}, + settings: ev.payload.settings, + }); + + this.#publishChanges(); + }), + ); + this.#disposables.push( + connection.disposableOn("willDisappear", (ev: WillDisappear) => { + if (this.#instances.delete(ev.context)) { + this.#publishChanges(); + } + }), + ); + } + + /** + * Attaches an RPC transport to the adapter. + * @param send Function responsible for sending JSON-RPC requests and responses. + */ + public attachRpc(send: RpcSender): void { + this.#rpc = createRpcServerClient(send); + this.#rpc.addMethod("streamDeck.bridge.getPluginState", () => this.getPluginState()); + this.#rpc.addMethod("streamDeck.bridge.setSettings", (params) => this.#setSettings(params)); + } + + /** + * Disposes the adapter and unregisters all connection listeners. + */ + public dispose(): void { + this.#disposables.forEach((disposable) => disposable.dispose()); + } + + /** + * Builds the current plugin state. + * @returns Current plugin state. + */ + public getPluginState(): PluginState { + this.#seedDevices(); + + const manifestActions = this.#getManifest()?.Actions ?? []; + const groups = new Map(); + + manifestActions.forEach((action) => { + groups.set(action.UUID, { + instances: [], + name: action.Name, + uuid: action.UUID, + }); + }); + + this.#instances.forEach((instance) => { + const action = groups.get(instance.manifestId) ?? { + instances: [], + name: instance.manifestId, + uuid: instance.manifestId, + }; + + action.instances.push({ + context: instance.context, + controller: instance.controller, + coordinates: instance.coordinates, + device: this.#getDeviceName(instance.deviceId), + isInMultiAction: instance.isInMultiAction, + resources: instance.resources, + settings: instance.settings, + }); + + groups.set(instance.manifestId, action); + }); + + return { + actions: [...groups.values()], + plugin: { + name: this.#getManifest()?.Name ?? connection.registrationParameters.info.plugin.uuid, + uuid: this.#getManifest()?.UUID ?? connection.registrationParameters.info.plugin.uuid, + version: this.#getManifest()?.Version ?? connection.registrationParameters.info.plugin.version, + }, + }; + } + + /** + * Attempts to process the specified JSON-RPC message. + * @param value Value to process. + * @returns `true` when the value was handled by the adapter. + */ + public receive(value: JsonValue): Promise { + return this.#rpc ? this.#rpc.receive(value) : Promise.resolve(false); + } + + /** + * Starts the bridge transport when the plugin is running in debug mode. + * @returns A promise resolved when startup is complete. + */ + public async start(): Promise { + if (!isDebugMode()) { + return; + } + + const uuid = this.#getManifest()?.UUID ?? connection.registrationParameters.info.plugin.uuid; + try { + await socket.start(this, uuid); + } catch (err) { + logger.warn("Failed to start bridge socket", err); + } + } + + /** + * Persists settings for a tracked action instance and re-publishes plugin state. + * @param params Context and settings to persist. + * @returns `true` when the instance was known and the update was sent. + */ + async #setSettings(params: SetSettingsParams | undefined): Promise { + const { context, settings } = params ?? {}; + if (typeof context !== "string" || !this.#isJsonObject(settings)) { + return false; + } + + const instance = this.#instances.get(context); + if (!instance) { + return false; + } + + await connection.send({ + event: "setSettings", + context, + payload: settings, + }); + + // Stream Deck does not echo a didReceiveSettings for setSettings, so update the + // tracked instance optimistically to keep the published plugin state in sync. + instance.settings = settings; + this.#publishChanges(); + return true; + } + + /** + * Determines whether a value is a JSON object. + * @param value Value to inspect. + * @returns `true` when the value is a non-array object. + */ + #isJsonObject(value: unknown): value is JsonObject { + return typeof value === "object" && value !== null && !Array.isArray(value); + } + + /** + * Gets the name associated with a device identifier. + * @param id Device identifier. + * @returns Device name when known; otherwise the identifier. + */ + #getDeviceName(id: string): string { + return this.#devices.get(id) ?? id; + } + + /** + * Enqueues a plugin state publication, ensuring notifications are emitted in call order. + */ + #publishChanges(): void { + this.#publishQueue = this.#publishQueue.then(() => this.#doPublish()); + } + + /** + * Publishes the current plugin state when it differs from the last emitted state. + */ + async #doPublish(): Promise { + try { + const pluginState = this.getPluginState(); + const serialized = JSON.stringify(pluginState); + + if (serialized === this.#lastPluginState) { + return; + } + + this.#lastPluginState = serialized; + if (this.#rpc) { + await this.#rpc.notify("streamDeck.bridge.pluginStateChanged", pluginState); + } + } catch { + // Swallow transport errors to avoid disrupting the plugin; + } + } + + /** + * Seeds the internal device map from Stream Deck registration parameters. + */ + #seedDevices(): void { + if (this.#seededDevices) { + return; + } + + connection.registrationParameters.info.devices.forEach((device) => { + this.#devices.set(device.id, device.name); + }); + + this.#seededDevices = true; + } + + /** + * Gets the parsed manifest, loading it on first access. + * @returns Parsed manifest, or `null` when unavailable. + */ + #getManifest(): Manifest | null { + return (this.#manifest ??= getManifest()); + } +} + +/** + * Singleton bridge adapter. + */ +export const bridge = new BridgeAdapter(); + +/** + * Parameters for the `streamDeck.bridge.setSettings` RPC method. + */ +type SetSettingsParams = { + /** + * Context identifier of the action instance to update. + */ + context: string; + + /** + * Settings to persist for the action instance. + */ + settings: JsonObject; +}; + +/** + * Serializable state of the plugin's visible actions. + */ +type PluginState = { + /** + * Actions known to the plugin, including currently visible instances. + */ + actions: ActionState[]; + + /** + * Plugin metadata associated with the state. + */ + plugin: PluginMetadata; +}; + +/** + * State of a manifest action and its currently visible instances. + */ +type ActionState = { + /** + * Visible instances associated with the action. + */ + instances: ActionInstanceState[]; + + /** + * Human-readable action name. + */ + name: string; + + /** + * Manifest action UUID. + */ + uuid: string; +}; + +/** + * State of a visible action instance. + */ +type ActionInstanceState = { + /** + * Unique context identifier for the action instance. + */ + context: string; + + /** + * Controller type associated with the instance. + */ + controller: "Encoder" | "Keypad"; + + /** + * Coordinates associated with the instance. + */ + coordinates: Coordinates; + + /** + * Name of the device the instance is currently shown on. + */ + device: string; + + /** + * Determines whether the instance is part of a multi-action. + */ + isInMultiAction: boolean; + + /** + * Resources associated with the instance. + */ + resources: Resources; + + /** + * Persisted action settings. + */ + settings: JsonObject; +}; + +/** + * Serializable plugin metadata exposed by the state. + */ +type PluginMetadata = { + /** + * Human-readable plugin name. + */ + name: string; + + /** + * Plugin UUID. + */ + uuid: string; + + /** + * Plugin version. + */ + version: string; +}; + +/** + * Runtime payload shape used by the bridge for visible action positions. + */ +type PayloadWithCoordinates = { + /** + * Coordinates reported by Stream Deck. + */ + coordinates: Coordinates; + + /** + * Determines whether the instance is part of a multi-action. + */ + isInMultiAction?: boolean; + + /** + * Resources associated with the instance. + */ + resources?: Resources; +}; + +/** + * Internal state of a visible action instance. + */ +type InternalInstanceState = { + /** + * Unique action context identifier. + */ + context: string; + + /** + * Controller associated with the instance. + */ + controller: "Encoder" | "Keypad"; + + /** + * Coordinates associated with the instance. + */ + coordinates: Coordinates; + + /** + * Device identifier associated with the instance. + */ + deviceId: string; + + /** + * Determines whether the instance is part of a multi-action. + */ + isInMultiAction: boolean; + + /** + * Manifest action UUID associated with the instance. + */ + manifestId: string; + + /** + * Last known resources associated with the instance. + */ + resources: Resources; + + /** + * Last known settings associated with the instance. + */ + settings: JsonObject; +}; diff --git a/packages/plugin/src/plugin/bridge/pipe-path.ts b/packages/plugin/src/plugin/bridge/pipe-path.ts new file mode 100644 index 00000000..3cc068ba --- /dev/null +++ b/packages/plugin/src/plugin/bridge/pipe-path.ts @@ -0,0 +1,24 @@ +import { createHash } from "node:crypto"; +import { platform, tmpdir } from "node:os"; +import { join } from "node:path"; + +// NOTE: This derivation MUST stay byte-for-byte identical to the VS Code extension's copy +// (vscode-streamdeck: src/bridge/pipe-path.ts). Both sides compute the rendezvous path from the +// plugin UUID independently; if they diverge, the extension can no longer find the plugin. + +/** + * Computes the pipe path for the bridge transport, derived deterministically from the plugin UUID. + * + * The UUID is hashed to a short, filesystem-safe token so the resulting Unix domain socket path + * stays within the platform's path length limit (~104 bytes on macOS). + * @param uuid Plugin UUID. + * @returns Named pipe path (Windows) or Unix domain socket path (macOS/Linux). + */ +export function getPipePath(uuid: string): string { + const token = `sd-${createHash("sha1").update(uuid).digest("hex").slice(0, 16)}`; + if (platform() === "win32") { + return `\\\\.\\pipe\\${token}`; + } + + return join(tmpdir(), `${token}.sock`); +} diff --git a/packages/plugin/src/plugin/bridge/socket.ts b/packages/plugin/src/plugin/bridge/socket.ts new file mode 100644 index 00000000..9dcd43a9 --- /dev/null +++ b/packages/plugin/src/plugin/bridge/socket.ts @@ -0,0 +1,198 @@ +import { type JsonValue, withResolvers } from "@elgato/utils"; +import type { RpcSender } from "@elgato/utils/rpc"; +import { rm } from "node:fs/promises"; +import { createServer, type Server, type Socket } from "node:net"; +import { platform } from "node:os"; + +import { logger } from "../logging/index.js"; +import { getPipePath } from "./pipe-path.js"; + +/** + * Hosts the bridge adapter over a named pipe (Windows) or Unix domain socket + * (macOS/Linux) whose path is derived from the plugin UUID, so the VS Code extension can + * discover and connect to it without any port handshake. + */ +class BridgeSocket { + /** + * Connected VS Code client. + */ + #client: Socket | undefined; + + /** + * Path the server is currently listening on. + */ + #path: string | undefined; + + /** + * RPC host bound to the socket transport. + */ + #rpcHost: SocketRpcHost | undefined; + + /** + * Underlying socket server. + */ + #server: Server | undefined; + + /** + * Starts the bridge socket server on the UUID-derived path so the VS Code extension can connect. + * @param rpcHost RPC host bound to the socket transport. + * @param uuid Plugin UUID used to derive the pipe path. + */ + public async start(rpcHost: SocketRpcHost, uuid: string): Promise { + this.#rpcHost = rpcHost; + if (this.#server) { + if (this.#client) { + this.#attachRpc(this.#client); + } + + return; + } + + const path = getPipePath(uuid); + + // A Unix domain socket leaves a file behind if the previous process crashed; remove any + // stale socket before listening to avoid EADDRINUSE. Windows named pipes self-clean. + if (platform() !== "win32") { + await rm(path, { force: true }); + } + + const server = createServer((socket) => this.#onConnection(socket)); + await new Promise((resolve, reject) => { + server.once("listening", () => resolve()); + server.once("error", (err) => reject(err)); + server.listen(path); + }); + + this.#server = server; + this.#path = path; + logger.debug(`Bridge socket listening on ${path}`); + } + + /** + * Stops the bridge socket server and removes its socket file. + */ + public async stop(): Promise { + this.#client?.destroy(); + this.#client = undefined; + + const server = this.#server; + const path = this.#path; + this.#server = undefined; + this.#path = undefined; + if (server) { + await new Promise((resolve) => server.close(() => resolve())); + if (path && platform() !== "win32") { + await rm(path, { force: true }); + } + } + } + + /** + * Binds a newly connected VS Code client. + * @param socket Connected socket. + */ + #onConnection(socket: Socket): void { + this.#client?.destroy(); + this.#client = socket; + this.#attachRpc(socket); + + let buffer = ""; + let messageReceived = Promise.resolve(); + const receiveMessage = async (line: string): Promise => { + const previousMessageReceived = messageReceived; + const received = withResolvers(); + messageReceived = received.promise; + + try { + await previousMessageReceived; + await this.#onMessage(socket, line); + } finally { + received.resolve(); + } + }; + + socket.setEncoding("utf-8"); + socket.on("data", (chunk: string) => { + buffer += chunk; + let newline = buffer.indexOf("\n"); + while (newline !== -1) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + if (line.length > 0) { + receiveMessage(line); + } + + newline = buffer.indexOf("\n"); + } + }); + socket.on("close", () => { + if (this.#client === socket) { + this.#client = undefined; + } + }); + socket.on("error", () => { + if (this.#client === socket) { + this.#client = undefined; + } + }); + } + + /** + * Attaches the RPC host to a socket-specific sender. + * @param socket Socket used by the RPC sender. + */ + #attachRpc(socket: Socket): void { + this.#rpcHost?.attachRpc((message) => this.#send(socket, message)); + } + + /** + * Forwards a JSON-RPC message from the client to the RPC host. + * @param socket Socket that received the message. + * @param line Newline-delimited JSON message. + */ + async #onMessage(socket: Socket, line: string): Promise { + if (this.#client !== socket) { + return; + } + + try { + await this.#rpcHost?.receive(JSON.parse(line)); + } catch { + // Ignore malformed or stale bridge messages; reconnect will refresh state. + } + } + + /** + * Sends a JSON-RPC message to the connected client. + * @param socket Socket associated with the RPC exchange. + * @param message Message to send. + */ + async #send(socket: Socket, message: unknown): Promise { + if (this.#client === socket && !socket.destroyed) { + socket.write(`${JSON.stringify(message)}\n`); + } + } +} + +/** + * Singleton bridge socket transport. + */ +export const socket = new BridgeSocket(); + +/** + * Minimal RPC host interface required by the bridge socket transport. + */ +type SocketRpcHost = { + /** + * Attaches the socket transport to the RPC host. + * @param send Sender used for outbound JSON-RPC messages. + */ + attachRpc(send: RpcSender): void; + + /** + * Attempts to process a JSON-RPC payload received from the socket client. + * @param value Received JSON value. + * @returns `true` when the value was handled by the RPC host. + */ + receive(value: JsonValue): Promise; +}; diff --git a/packages/plugin/src/plugin/index.ts b/packages/plugin/src/plugin/index.ts index feaf1e19..09b19bd1 100644 --- a/packages/plugin/src/plugin/index.ts +++ b/packages/plugin/src/plugin/index.ts @@ -4,6 +4,7 @@ import type { Logger } from "@elgato/utils/logging"; import type { Language, RegistrationInfo } from "../api/index.js"; import { actionService, type ActionService } from "./actions/service.js"; import { connection } from "./connection.js"; +import { bridge } from "./bridge/adapter.js"; import { deviceService, type DeviceService } from "./devices/service.js"; import { fileSystemLocaleProvider } from "./i18n.js"; import { logger } from "./logging/index.js"; @@ -114,8 +115,9 @@ export const streamDeck = { * Connects the plugin to the Stream Deck. * @returns A promise resolved when a connection has been established. */ - connect(): Promise { - return connection.connect(); + async connect(): Promise { + await connection.connect(); + await bridge.start(); }, };