import { createServer } from 'node:http';
import { WebSocket, WebSocketServer } from 'ws';
import z from '@deepseek-ai/schemastery';

const name = 'dsh-watch';

const Config = z.object({
  mode: z.union([z.const('direct'), z.const('relay')]).default('relay'),
  // Relay mode
  serverUrl: z.string().default(''),
  userId: z.string().default(''),
  token: z.string().default(''),
  reconnectMs: z.number().min(500).default(5000),
  timeoutMs: z.number().min(1000).default(30000),
  // Direct mode
  host: z.union([z.const('0.0.0.0'), z.const('127.0.0.1')]).default('0.0.0.0'),
  port: z.natural().max(65535).default(18321)
});

function apply(ctx, config) {
  let api;
  ctx.inject(['apiProxy'], (c) => { api = c.apiProxy; });

  const directClients = new Set();
  let relayWs = null;
  let httpServer = null;
  let wss = null;
  let closed = false;
  let reconnectTimer = null;
  let statusTimer = null;

  function log(msg) { ctx.logger.info('[dsh-watch] ' + msg); }
  function warn(msg) { ctx.logger.warn('[dsh-watch] ' + msg); }

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

  function pushEvent(event) {
    const frame = { type: 'event', event };
    if (config.mode === 'direct') {
      for (const client of directClients) safeSend(client, frame);
    } else {
      safeSend(relayWs, frame);
    }
  }

  function pushStatus(status) {
    const frame = { type: 'status', status };
    if (config.mode === 'direct') {
      for (const client of directClients) safeSend(client, frame);
    } else {
      safeSend(relayWs, frame);
    }
  }

  function sendStatus() {
    computeStatus().then((status) => {
      pushStatus(status);
    }).catch((err) => warn('status error ' + String(err && err.message || err)));
  }

  async function computeStatus() {
    let host = {};
    let items = [];
    try {
      if (api && api.host) host = await api.host.describe({});
    } catch (err) { warn('host.describe ' + String(err && err.message || err)); }
    try {
      if (api && api.sessions) {
        const list = await api.sessions.list({});
        items = list && list.items ? list.items : [];
      }
    } catch (err) { warn('session.list ' + String(err && err.message || err)); }
    let active = null;
    if (items.length) {
      active = items.find((s) => s.running) || items.find((s) => !s.blank) || items.slice().sort((a, b) => b.updatedAt - a.updatedAt)[0] || null;
    }
    return {
      online: true,
      ts: Date.now(),
      version: host.version,
      cwd: host.cwd,
      provider: host.provider,
      model: host.model,
      attachedSessions: host.attachedSessions,
      activeSessionId: active ? active.sessionId : null,
      running: active ? Boolean(active.running) : false,
      sessions: items.slice(0, 50)
    };
  }

  async function invoke(method, params) {
    params = params || {};
    if (method === 'harness.status') return computeStatus();
    if (method === 'harness.prompt') return harnessPrompt(params);
    if (method === 'harness.cancel') return harnessCancel();
    const dot = method.indexOf('.');
    if (dot <= 0) throw new Error('unknown method ' + method);
    const domain = method.slice(0, dot);
    const fnName = method.slice(dot + 1);
    const domainObj = api && api[domain];
    if (!domainObj || typeof domainObj[fnName] !== 'function') {
      throw new Error('no handler for ' + method);
    }
    return domainObj[fnName](params);
  }

  async function harnessPrompt(params) {
    let sessionId = params.sessionId;
    if (!sessionId) {
      const status = await computeStatus();
      sessionId = status.activeSessionId;
    }
    if (!sessionId) {
      const created = await api.sessions.create({});
      sessionId = created.sessionId;
    }
    let content = params.content;
    if (typeof content === 'string') content = [{ type: 'text', text: content }];
    if (!Array.isArray(content)) throw new Error('content must be a string or an array of content blocks');
    const mode = params.mode === 'queue' ? 'queue' : 'steer';
    return api.sessions.prompt({
      sessionId,
      mode,
      content,
      clientTimeZone: params.clientTimeZone
    });
  }

  async function harnessCancel() {
    const status = await computeStatus();
    const sessionId = status.activeSessionId;
    if (!sessionId) throw new Error('no active session to cancel');
    return api.sessions.cancel({ sessionId });
  }

  function handleRequest(msg, respond) {
    invoke(msg.method || '', msg.params || {}).then((result) => {
      respond({ type: 'response', requestId: msg.requestId, ok: true, result });
    }).catch((err) => {
      respond({
        type: 'response',
        requestId: msg.requestId,
        ok: false,
        error: { code: 'invoke-error', message: String(err && err.message || err) }
      });
    });
  }

  function isAuthed(msg) {
    return config.token && msg.token === config.token;
  }

  // ---------- Direct mode ----------
  function handleDirectConnection(ws) {
    ws.isAlive = true;
    ws.on('pong', () => { ws.isAlive = true; });
    ws.on('message', (raw) => {
      let msg;
      try { msg = JSON.parse(String(raw)); } catch { return; }
      if (!msg || typeof msg !== 'object') return;
      if (msg.type === 'hello') {
        if (msg.role !== 'client' || !isAuthed(msg)) {
          safeSend(ws, { type: 'auth-error', message: 'invalid token or role' });
          ws.close();
          return;
        }
        directClients.add(ws);
        computeStatus().then((status) => {
          safeSend(ws, { type: 'hello', role: 'client', online: true, harness: true, status });
        }).catch(() => {
          safeSend(ws, { type: 'hello', role: 'client', online: true, harness: true, status: null });
        });
        return;
      }
      if (!directClients.has(ws)) { ws.close(); return; }
      if (msg.type === 'request') {
        handleRequest(msg, (resp) => safeSend(ws, resp));
      } else if (msg.type === 'ping') {
        safeSend(ws, { type: 'pong' });
      }
    });
    ws.on('close', () => directClients.delete(ws));
    ws.on('error', () => {});
  }

  function startDirect() {
    httpServer = 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, mode: 'direct', clients: directClients.size, port: config.port }));
        return;
      }
      res.writeHead(200, { 'Content-Type': 'text/plain; charset=utf-8' });
      res.end('dsh-watch direct mode. Connect clients to /ws\n');
    });
    wss = new WebSocketServer({ server: httpServer });
    wss.on('connection', handleDirectConnection);
    httpServer.listen(config.port, config.host, () => {
      log('direct mode listening on ws://' + config.host + ':' + config.port + '/ws');
      sendStatus();
    });
  }

  // ---------- Relay mode ----------
  function connectRelay() {
    if (closed || (relayWs && relayWs.readyState === WebSocket.OPEN)) return;
    const w = new WebSocket(config.serverUrl);
    relayWs = w;
    w.on('open', () => {
      safeSend(w, { type: 'hello', role: 'dsh', userId: config.userId, token: config.token });
      sendStatus();
      log('relay connected to ' + config.serverUrl + ' as ' + config.userId);
    });
    w.on('message', (raw) => {
      let msg;
      try { msg = JSON.parse(String(raw)); } catch { return; }
      if (!msg || typeof msg !== 'object') return;
      if (msg.type === 'request') {
        handleRequest(msg, (resp) => safeSend(w, resp));
      } else if (msg.type === 'ping') {
        safeSend(w, { type: 'pong' });
      } else if (msg.type === 'hello') {
        log('registered');
      }
    });
    w.on('close', () => {
      if (relayWs === w) relayWs = null;
      scheduleReconnect();
    });
    w.on('error', (err) => warn('relay socket error ' + String(err && err.message || err)));
  }

  function scheduleReconnect() {
    if (closed || reconnectTimer) return;
    reconnectTimer = setTimeout(() => {
      reconnectTimer = null;
      connectRelay();
    }, config.reconnectMs);
  }

  // ---------- Shared event hooks ----------
  ctx.on('session/event', (session, event) => {
    pushEvent({ type: 'session/event', sessionId: session.id, event });
    scheduleStatus();
  });
  ctx.inject(['sessionProjections'], (c) => {
    c.sessionProjections.onChanged((session, key, value, seq) => {
      pushEvent({ type: 'session/projection', sessionId: session.id, key, value, seq });
    });
  });

  let statusDebounce = null;
  function scheduleStatus() {
    if (statusDebounce) return;
    statusDebounce = setTimeout(() => {
      statusDebounce = null;
      sendStatus();
    }, 500);
  }

  statusTimer = setInterval(() => sendStatus(), 15000);

  ctx.effect(() => {
    closed = true;
    if (reconnectTimer) clearTimeout(reconnectTimer);
    if (statusTimer) clearInterval(statusTimer);
    if (statusDebounce) clearTimeout(statusDebounce);
    for (const ws of directClients) { try { ws.close(); } catch {} }
    directClients.clear();
    if (relayWs) { try { relayWs.close(); } catch {} relayWs = null; }
    if (wss) { try { wss.close(); } catch {} wss = null; }
    if (httpServer) { try { httpServer.close(); } catch {} httpServer = null; }
  });

  if (config.mode === 'direct') {
    if (!config.port || !config.token) {
      warn('direct mode requires port and token; bridge stays idle.');
      return;
    }
    startDirect();
  } else {
    if (!config.serverUrl || !config.userId || !config.token) {
      warn('relay mode requires serverUrl/userId/token; bridge stays idle.');
      return;
    }
    connectRelay();
  }
}

export { Config, apply, name };