DSHWatch_Server/server.js

// DSHWatch relay broker
// Bridges DeepSeek Harness instances (DSH plugins) with phone/watch clients over WebSocket.
//
// Roles:
//   dsh    - the DeepSeek Harness plugin (one per userId)
//   client - phone app (and through the phone app, the watch)
//
// Protocol is a tiny JSON envelope. See README.md.

import { createServer } from 'node:http';
import { WebSocketServer } from 'ws';
import { readFileSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import path from 'node:path';
import process from 'node:process';

const __dirname = path.dirname(fileURLToPath(import.meta.url));

function loadConfig() {
  const configPath = process.env.DSHWATCH_SERVER_CONFIG || path.join(__dirname, 'config.json');
  let config = {};
  try {
    config = JSON.parse(readFileSync(configPath, 'utf8'));
  } catch {
    // use defaults
  }
  return {
    port: Number(config.port || process.env.PORT || 8787),
    host: config.host || process.env.HOST || '0.0.0.0',
    token: config.token || process.env.DSHWATCH_TOKEN || 'change-me',
    heartbeatMs: Number(config.heartbeatMs || 30000)
  };
}

const config = loadConfig();

const harnesses = new Map();
const clients = new Map();
const pending = new Map();
const lastStatus = new Map();
const wsUserId = new Map();

function bindUserId(ws, userId) { wsUserId.set(ws, userId); }
function userIdOf(ws) { return wsUserId.get(ws); }

function clientsFor(userId) {
  let set = clients.get(userId);
  if (!set) {
    set = new Set();
    clients.set(userId, set);
  }
  return set;
}

function send(ws, obj) {
  if (ws.readyState === ws.OPEN) {
    try { ws.send(JSON.stringify(obj)); } catch { /* ignore */ }
  }
}

function broadcastToClients(userId, obj) {
  for (const ws of clientsFor(userId)) send(ws, obj);
}

function replyError(clientWs, requestId, message, code) {
  send(clientWs, { type: 'response', requestId, ok: false, error: { code: code || 'unreachable', message } });
}

function registerHarness(userId, ws) {
  const prev = harnesses.get(userId);
  if (prev && prev !== ws) {
    send(prev, { type: 'notice', code: 'replaced', message: 'A newer harness connection took over this userId.' });
    try { prev.close(); } catch {}
  }
  harnesses.set(userId, ws);
  const status = lastStatus.get(userId) || null;
  send(ws, { type: 'hello', role: 'dsh', online: true, status: status ? { userId: status.userId } : {} });
  broadcastToClients(userId, { type: 'harness', online: true });
}

function unregisterHarness(userId, ws) {
  if (harnesses.get(userId) === ws) {
    harnesses.delete(userId);
    broadcastToClients(userId, { type: 'harness', online: false });
  }
}

function handleDshHello(userId, ws, msg) {
  bindUserId(ws, userId);
  if (msg.token !== config.token) {
    send(ws, { type: 'auth-error', message: 'invalid token' });
    ws.close();
    return;
  }
  ws.role = 'dsh';
  registerHarness(userId, ws);
}

function handleClientHello(userId, ws, msg) {
  bindUserId(ws, userId);
  if (msg.token !== config.token) {
    send(ws, { type: 'auth-error', message: 'invalid token' });
    ws.close();
    return;
  }
  ws.role = 'client';
  clientsFor(userId).add(ws);
  const harness = harnesses.get(userId);
  const status = lastStatus.get(userId) || null;
  send(ws, {
    type: 'hello',
    role: 'client',
    online: Boolean(harness),
    harness: Boolean(harness),
    status
  });
}

function handleHarnessMessage(userId, ws, msg) {
  if (msg.type === 'response') {
    const req = pending.get(msg.requestId);
    if (req) {
      clearTimeout(req.timer);
      pending.delete(msg.requestId);
      send(req.clientWs, {
        type: 'response',
        requestId: req.clientRequestId,
        ok: msg.ok !== false,
        result: msg.result,
        error: msg.error
      });
    }
  } else if (msg.type === 'event') {
    broadcastToClients(userId, { type: 'event', event: msg.event });
  } else if (msg.type === 'status') {
    lastStatus.set(userId, msg.status || null);
    broadcastToClients(userId, { type: 'status', status: msg.status });
  } else if (msg.type === 'ping') {
    send(ws, { type: 'pong' });
  }
}

function handleClientMessage(userId, ws, msg) {
  if (msg.type === 'request') {
    const harness = harnesses.get(userId);
    if (!harness) {
      replyError(ws, msg.requestId, 'harness offline');
      return;
    }
    const requestId = msg.requestId || ('req-' + Date.now() + '-' + Math.random().toString(16).slice(2));
    if (pending.has(requestId)) {
      replyError(ws, msg.requestId, 'duplicate requestId', 'bad-request');
      return;
    }
    const timer = setTimeout(() => {
      pending.delete(requestId);
      replyError(ws, msg.requestId, 'harness timeout');
    }, Number(msg.timeoutMs) || 30000);
    pending.set(requestId, { userId, clientWs: ws, clientRequestId: msg.requestId, timer });
    send(harness, { type: 'request', requestId, method: msg.method, params: msg.params || {} });
  } else if (msg.type === 'ping') {
    send(ws, { type: 'pong' });
  }
}

function handleMessage(ws, raw) {
  let msg;
  try { msg = JSON.parse(String(raw)); } catch {
    send(ws, { type: 'error', message: 'invalid JSON' });
    return;
  }
  if (!msg || typeof msg !== 'object') return;
  const userId = msg.userId;
  if (typeof userId !== 'string' || userId.length === 0) {
    send(ws, { type: 'auth-error', message: 'missing userId' });
    ws.close();
    return;
  }
  if (msg.type === 'hello' && !ws.role) {
    if (msg.role === 'dsh') handleDshHello(userId, ws, msg);
    else handleClientHello(userId, ws, msg);
    return;
  }
  if (ws.role === 'dsh') {
    handleHarnessMessage(userId, ws, msg);
  } else if (ws.role === 'client') {
    handleClientMessage(userId, ws, msg);
  } else {
    send(ws, { type: 'auth-error', message: 'hello required' });
    ws.close();
  }
}

function handleConnection(ws, req) {
  const url = new URL(req.url || '/', 'http://localhost');
  if (url.pathname !== '/ws' && url.pathname !== '/') {
    ws.close(4001, 'unknown path');
    return;
  }
  ws.isAlive = true;
  ws.on('pong', () => { ws.isAlive = true; });
  ws.on('message', (raw) => {
    try { handleMessage(ws, raw); } catch (err) {
      send(ws, { type: 'error', message: String(err && err.message || err) });
      ws.close();
    }
  });
  ws.on('close', () => {
    const userId = userIdOf(ws);
    if (userId) {
      if (ws.role === 'dsh') unregisterHarness(userId, ws);
      else {
        const set = clients.get(userId);
        if (set) { set.delete(ws); if (set.size === 0) clients.delete(userId); }
      }
      for (const [id, req] of pending) {
        if (req.clientWs === ws) {
          clearTimeout(req.timer);
          pending.delete(id);
        }
      }
    }
    wsUserId.delete(ws);
  });
  ws.on('error', () => {});
}

const server = createServer((req, res) => {
  const url = new URL(req.url || '/', 'http://localhost');
  if (url.pathname === '/health') {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({
      ok: true,
      harnesses: Array.from(harnesses.entries()).map(([userId]) => ({ userId })),
      clients: Array.from(clients.entries()).map(([userId, set]) => ({ userId, count: set.size }))
    }));
    return;
  }
  res.writeHead(200, { 'Content-Type': 'text/plain; charset=utf-8' });
  res.end('DSHWatch relay broker running. Connect clients to /ws\n');
});

const wss = new WebSocketServer({ server });
wss.on('connection', handleConnection);

setInterval(() => {
  for (const ws of wss.clients) {
    if (ws.isAlive === false) { ws.terminate(); continue; }
    ws.isAlive = false;
    ws.ping();
  }
}, config.heartbeatMs);

server.listen(config.port, config.host, () => {
  console.log(String('[dshwatch-server] listening on ws://' + config.host + ':' + config.port + '/ws'));
  console.log(String('[dshwatch-server] health: http://' + config.host + ':' + config.port + '/health'));
});