feat: finish TASK-27.4 mpv-service protocol transport split

src/core/services/mpv-service.test.ts

@@ -187,6 +187,86 @@ test("MpvIpcClient scheduleReconnect clears existing reconnect timer", () => {
   assert.equal(connectCalled, true);
 });
 
+test("MpvIpcClient onClose resolves outstanding requests and schedules reconnect", () => {
+  const timers: Array<ReturnType<typeof setTimeout> | null> = [];
+  const client = new MpvIpcClient(
+    "/tmp/mpv.sock",
+    makeDeps({
+      getReconnectTimer: () => null,
+      setReconnectTimer: (timer) => {
+        timers.push(timer);
+      },
+    }),
+  );
+
+  const resolved: Array<unknown> = [];
+  (client as any).pendingRequests.set(1, (message: unknown) => {
+    resolved.push(message);
+  });
+
+  let reconnectConnectCount = 0;
+  (client as any).connect = () => {
+    reconnectConnectCount += 1;
+  };
+
+  const originalSetTimeout = globalThis.setTimeout;
+  (globalThis as any).setTimeout = (handler: () => void, _delay: number) => {
+    handler();
+    return 1 as unknown as ReturnType<typeof setTimeout>;
+  };
+
+  try {
+    (client as any).transport.callbacks.onClose();
+  } finally {
+    (globalThis as any).setTimeout = originalSetTimeout;
+  }
+
+  assert.equal(resolved.length, 1);
+  assert.deepEqual(resolved[0], { request_id: 1, error: "disconnected" });
+  assert.equal(reconnectConnectCount, 1);
+  assert.equal(timers.length, 1);
+});
+
+test("MpvIpcClient reconnect replays property subscriptions and initial state requests", () => {
+  const commands: unknown[] = [];
+  const client = new MpvIpcClient("/tmp/mpv.sock", makeDeps());
+  (client as any).send = (command: unknown) => {
+    commands.push(command);
+    return true;
+  };
+
+  const callbacks = (client as any).transport.callbacks;
+  callbacks.onConnect();
+
+  commands.length = 0;
+  callbacks.onConnect();
+
+  const hasSecondaryVisibilityReset = commands.some(
+    (command) =>
+      Array.isArray((command as { command: unknown[] }).command) &&
+      (command as { command: unknown[] }).command[0] === "set_property" &&
+      (command as { command: unknown[] }).command[1] === "secondary-sub-visibility" &&
+      (command as { command: unknown[] }).command[2] === "no",
+  );
+  const hasTrackSubscription = commands.some(
+    (command) =>
+      Array.isArray((command as { command: unknown[] }).command) &&
+      (command as { command: unknown[] }).command[0] === "observe_property" &&
+      (command as { command: unknown[] }).command[1] === 1 &&
+      (command as { command: unknown[] }).command[2] === "sub-text",
+  );
+  const hasPathRequest = commands.some(
+    (command) =>
+      Array.isArray((command as { command: unknown[] }).command) &&
+      (command as { command: unknown[] }).command[0] === "get_property" &&
+      (command as { command: unknown[] }).command[1] === "path",
+  );
+
+  assert.equal(hasSecondaryVisibilityReset, true);
+  assert.equal(hasTrackSubscription, true);
+  assert.equal(hasPathRequest, true);
+});
+
 test("MpvIpcClient captures and disables secondary subtitle visibility on request", async () => {
   const commands: unknown[] = [];
   const client = new MpvIpcClient("/tmp/mpv.sock", makeDeps());

src/core/services/mpv-service.ts

@@ -1,4 +1,3 @@
-import * as net from "net";
 import { EventEmitter } from "events";
 import {
   Config,
@@ -16,6 +15,7 @@ import {
 import { requestMpvInitialState, subscribeToMpvProperties } from "./mpv-properties";
 import {
   scheduleMpvReconnect,
+  MpvSocketTransport,
 } from "./mpv-transport";
 import { resolveCurrentAudioStreamIndex } from "./mpv-state";
 
@@ -49,9 +49,9 @@ export interface MpvIpcClientEventMap {
 type MpvIpcClientEventName = keyof MpvIpcClientEventMap;
 
 export class MpvIpcClient implements MpvClient {
-  private socketPath: string;
   private deps: MpvIpcClientProtocolDeps;
-  public socket: net.Socket | null = null;
+  private transport: MpvSocketTransport;
+  public socket: ReturnType<MpvSocketTransport["getSocket"]> = null;
   private eventBus = new EventEmitter();
   private buffer = "";
   public connected = false;
@@ -95,8 +95,52 @@ export class MpvIpcClient implements MpvClient {
     socketPath: string,
     deps: MpvIpcClientDeps,
   ) {
-    this.socketPath = socketPath;
     this.deps = deps;
+
+    this.transport = new MpvSocketTransport({
+      socketPath,
+      onConnect: () => {
+        console.log("Connected to MPV socket");
+        this.connected = true;
+        this.connecting = false;
+        this.socket = this.transport.getSocket();
+        this.reconnectAttempt = 0;
+        this.hasConnectedOnce = true;
+        this.setSecondarySubVisibility(false);
+        subscribeToMpvProperties(this.send.bind(this));
+        requestMpvInitialState(this.send.bind(this));
+
+        const shouldAutoStart =
+          this.deps.autoStartOverlay ||
+          this.deps.getResolvedConfig().auto_start_overlay === true;
+        if (this.firstConnection && shouldAutoStart) {
+          console.log("Auto-starting overlay, hiding mpv subtitles");
+          setTimeout(() => {
+            this.deps.setOverlayVisible(true);
+          }, 100);
+        } else if (this.deps.shouldBindVisibleOverlayToMpvSubVisibility()) {
+          this.setSubVisibility(!this.deps.isVisibleOverlayVisible());
+        }
+
+        this.firstConnection = false;
+      },
+      onData: (data) => {
+        this.buffer += data.toString();
+        this.processBuffer();
+      },
+      onError: (err: Error) => {
+        console.error("MPV socket error:", err.message);
+        this.failPendingRequests();
+      },
+      onClose: () => {
+        console.log("MPV socket closed");
+        this.connected = false;
+        this.connecting = false;
+        this.socket = null;
+        this.failPendingRequests();
+        this.scheduleReconnect();
+      },
+    });
   }
 
   on<EventName extends MpvIpcClientEventName>(
@@ -131,7 +175,7 @@ export class MpvIpcClient implements MpvClient {
   }
 
   setSocketPath(socketPath: string): void {
-    this.socketPath = socketPath;
+    this.transport.setSocketPath(socketPath);
   }
 
   connect(): void {
@@ -139,59 +183,8 @@ export class MpvIpcClient implements MpvClient {
       return;
     }
 
-    if (this.socket) {
-      this.socket.destroy();
-    }
-
     this.connecting = true;
-    this.socket = new net.Socket();
-
-    this.socket.on("connect", () => {
-      console.log("Connected to MPV socket");
-      this.connected = true;
-      this.connecting = false;
-      this.reconnectAttempt = 0;
-      this.hasConnectedOnce = true;
-      this.setSecondarySubVisibility(false);
-      subscribeToMpvProperties(this.send.bind(this));
-      requestMpvInitialState(this.send.bind(this));
-
-      const shouldAutoStart =
-        this.deps.autoStartOverlay ||
-        this.deps.getResolvedConfig().auto_start_overlay === true;
-      if (this.firstConnection && shouldAutoStart) {
-        console.log("Auto-starting overlay, hiding mpv subtitles");
-        setTimeout(() => {
-          this.deps.setOverlayVisible(true);
-        }, 100);
-      } else if (this.deps.shouldBindVisibleOverlayToMpvSubVisibility()) {
-        this.setSubVisibility(!this.deps.isVisibleOverlayVisible());
-      }
-
-      this.firstConnection = false;
-    });
-
-    this.socket.on("data", (data: Buffer) => {
-      this.buffer += data.toString();
-      this.processBuffer();
-    });
-
-    this.socket.on("error", (err: Error) => {
-      console.error("MPV socket error:", err.message);
-      this.connected = false;
-      this.connecting = false;
-      this.failPendingRequests();
-    });
-
-    this.socket.on("close", () => {
-      console.log("MPV socket closed");
-      this.connected = false;
-      this.connecting = false;
-      this.failPendingRequests();
-      this.scheduleReconnect();
-    });
-
-    this.socket.connect(this.socketPath);
+    this.transport.connect();
   }
 
   private scheduleReconnect(): void {
@@ -349,9 +342,7 @@ export class MpvIpcClient implements MpvClient {
     if (!this.connected || !this.socket) {
       return false;
     }
-    const msg = JSON.stringify(command) + "\n";
-    this.socket.write(msg);
-    return true;
+    return this.transport.send(command);
   }
 
   request(command: unknown[]): Promise<MpvMessage> {

src/core/services/mpv-transport.test.ts

@@ -1,10 +1,49 @@
 import test from "node:test";
 import assert from "node:assert/strict";
+import * as net from "node:net";
+import { EventEmitter } from "node:events";
 import {
   getMpvReconnectDelay,
+  MpvSocketMessagePayload,
+  MpvSocketTransport,
   scheduleMpvReconnect,
 } from "./mpv-transport";
 
+class FakeSocket extends EventEmitter {
+  public connectedPaths: string[] = [];
+  public writePayloads: string[] = [];
+  public destroyed = false;
+
+  connect(path: string): void {
+    this.connectedPaths.push(path);
+    setTimeout(() => {
+      this.emit("connect");
+    }, 0);
+  }
+
+  write(payload: string): boolean {
+    this.writePayloads.push(payload);
+    return true;
+  }
+
+  destroy(): void {
+    this.destroyed = true;
+    this.emit("close");
+  }
+}
+
+function withSocketMock<T>(fn: () => T): T {
+  const OriginalSocket = net.Socket;
+  (net as any).Socket = FakeSocket as any;
+  try {
+    return fn();
+  } finally {
+    (net as any).Socket = OriginalSocket;
+  }
+}
+
+const wait = () => new Promise((resolve) => setTimeout(resolve, 0));
+
 test("getMpvReconnectDelay follows existing reconnect ramp", () => {
   assert.equal(getMpvReconnectDelay(0, true), 1000);
   assert.equal(getMpvReconnectDelay(1, true), 1000);
@@ -62,3 +101,143 @@ test("scheduleMpvReconnect clears existing timer and increments attempt", () =>
   assert.equal(calls[0].delay, getMpvReconnectDelay(3, true));
   assert.equal(connected, 1);
 });
+
+test("MpvSocketTransport connects and sends payloads over a live socket", async () => {
+  const events: string[] = [];
+  await withSocketMock(async () => {
+    const transport = new MpvSocketTransport({
+      socketPath: "/tmp/mpv.sock",
+      onConnect: () => {
+        events.push("connect");
+      },
+      onData: () => {
+        events.push("data");
+      },
+      onError: () => {
+        events.push("error");
+      },
+      onClose: () => {
+        events.push("close");
+      },
+    });
+
+    const payload: MpvSocketMessagePayload = {
+      command: ["sub-seek", 1],
+      request_id: 1,
+    };
+
+    assert.equal(transport.send(payload), false);
+
+    transport.connect();
+    await wait();
+
+    assert.equal(events.includes("connect"), true);
+    assert.equal(transport.send(payload), true);
+
+    const fakeSocket = transport.getSocket() as unknown as FakeSocket;
+    assert.equal(fakeSocket.connectedPaths.at(0), "/tmp/mpv.sock");
+    assert.equal(fakeSocket.writePayloads.length, 1);
+    assert.equal(fakeSocket.writePayloads.at(0), `${JSON.stringify(payload)}\n`);
+  });
+});
+
+test("MpvSocketTransport reports lifecycle transitions and callback order", async () => {
+  const events: string[] = [];
+  const fakeError = new Error("boom");
+
+  await withSocketMock(async () => {
+    const transport = new MpvSocketTransport({
+      socketPath: "/tmp/mpv.sock",
+      onConnect: () => {
+        events.push("connect");
+      },
+      onData: () => {
+        events.push("data");
+      },
+      onError: () => {
+        events.push("error");
+      },
+      onClose: () => {
+        events.push("close");
+      },
+    });
+
+    transport.connect();
+    await wait();
+
+    const socket = transport.getSocket() as unknown as FakeSocket;
+    socket.emit("error", fakeError);
+    socket.emit("data", Buffer.from("{}"));
+    socket.destroy();
+    await wait();
+
+    assert.equal(events.includes("connect"), true);
+    assert.equal(events.includes("data"), true);
+    assert.equal(events.includes("error"), true);
+    assert.equal(events.includes("close"), true);
+    assert.equal(transport.isConnected, false);
+    assert.equal(transport.isConnecting, false);
+    assert.equal(socket.destroyed, true);
+  });
+});
+
+test("MpvSocketTransport ignores connect requests while already connecting or connected", async () => {
+  const events: string[] = [];
+
+  await withSocketMock(async () => {
+    const transport = new MpvSocketTransport({
+      socketPath: "/tmp/mpv.sock",
+      onConnect: () => {
+        events.push("connect");
+      },
+      onData: () => {
+        events.push("data");
+      },
+      onError: () => {
+        events.push("error");
+      },
+      onClose: () => {
+        events.push("close");
+      },
+    });
+
+    transport.connect();
+    transport.connect();
+    await wait();
+
+    assert.equal(events.includes("connect"), true);
+    const socket = transport.getSocket() as unknown as FakeSocket;
+    socket.emit("close");
+    await wait();
+
+    transport.connect();
+    await wait();
+
+    assert.equal(events.filter((entry) => entry === "connect").length, 2);
+  });
+});
+
+test("MpvSocketTransport.shutdown clears socket and lifecycle flags", async () => {
+  await withSocketMock(async () => {
+    const transport = new MpvSocketTransport({
+      socketPath: "/tmp/mpv.sock",
+      onConnect: () => {
+      },
+      onData: () => {
+      },
+      onError: () => {
+      },
+      onClose: () => {
+      },
+    });
+
+    transport.connect();
+    await wait();
+    assert.equal(transport.isConnected, true);
+
+    transport.shutdown();
+    assert.equal(transport.isConnected, false);
+    assert.equal(transport.isConnecting, false);
+    assert.equal(transport.getSocket(), null);
+  });
+});

src/core/services/mpv-transport.ts

@@ -1,3 +1,5 @@
+import * as net from "net";
+
 export function getMpvReconnectDelay(
   attempt: number,
   hasConnectedOnce: boolean,
@@ -52,3 +54,116 @@ export function scheduleMpvReconnect(
   );
   return deps.attempt + 1;
 }
+
+export interface MpvSocketMessagePayload {
+  command: unknown[];
+  request_id?: number;
+}
+
+interface MpvSocketTransportEvents {
+  onConnect: () => void;
+  onData: (data: Buffer) => void;
+  onError: (error: Error) => void;
+  onClose: () => void;
+}
+
+export interface MpvSocketTransportOptions {
+  socketPath: string;
+  onConnect: () => void;
+  onData: (data: Buffer) => void;
+  onError: (error: Error) => void;
+  onClose: () => void;
+}
+
+export class MpvSocketTransport {
+  private socketPath: string;
+  private readonly callbacks: MpvSocketTransportEvents;
+  private socketRef: net.Socket | null = null;
+  public socket: net.Socket | null = null;
+  public connected = false;
+  public connecting = false;
+
+  constructor(options: MpvSocketTransportOptions) {
+    this.socketPath = options.socketPath;
+    this.callbacks = {
+      onConnect: options.onConnect,
+      onData: options.onData,
+      onError: options.onError,
+      onClose: options.onClose,
+    };
+  }
+
+  setSocketPath(socketPath: string): void {
+    this.socketPath = socketPath;
+  }
+
+  connect(): void {
+    if (this.connected || this.connecting) {
+      return;
+    }
+
+    if (this.socketRef) {
+      this.socketRef.destroy();
+    }
+
+    this.connecting = true;
+    this.socketRef = new net.Socket();
+    this.socket = this.socketRef;
+
+    this.socketRef.on("connect", () => {
+      this.connected = true;
+      this.connecting = false;
+      this.callbacks.onConnect();
+    });
+
+    this.socketRef.on("data", (data: Buffer) => {
+      this.callbacks.onData(data);
+    });
+
+    this.socketRef.on("error", (error: Error) => {
+      this.connected = false;
+      this.connecting = false;
+      this.callbacks.onError(error);
+    });
+
+    this.socketRef.on("close", () => {
+      this.connected = false;
+      this.connecting = false;
+      this.callbacks.onClose();
+    });
+
+    this.socketRef.connect(this.socketPath);
+  }
+
+  send(payload: MpvSocketMessagePayload): boolean {
+    if (!this.connected || !this.socketRef) {
+      return false;
+    }
+
+    const message = JSON.stringify(payload) + "\n";
+    this.socketRef.write(message);
+    return true;
+  }
+
+  shutdown(): void {
+    if (this.socketRef) {
+      this.socketRef.destroy();
+    }
+    this.socketRef = null;
+    this.socket = null;
+    this.connected = false;
+    this.connecting = false;
+  }
+
+  getSocket(): net.Socket | null {
+    return this.socketRef;
+  }
+
+  get isConnected(): boolean {
+    return this.connected;
+  }
+
+  get isConnecting(): boolean {
+    return this.connecting;
+  }
+}