ts6-botflow/query-events.mjs

import ssh2 from "ssh2";
import { StringDecoder } from "node:string_decoder";
export const unescapeQuery = (value) =>
  String(value).replace(
    /\\([\\spnrtvf/|])/g,
    (_, c) =>
      ({
        "\\": "\\",
        s: " ",
        p: "|",
        n: "\n",
        r: "\r",
        t: "\t",
        v: "\v",
        f: "\f",
        "/": "/",
        "|": "|",
      })[c],
  );
export const escapeQuery = (value) =>
  String(value).replace(
    /[\\ |\n\r\t]/g,
    (c) =>
      ({
        "\\": "\\\\",
        " ": "\\s",
        "|": "\\p",
        "\n": "\\n",
        "\r": "\\r",
        "\t": "\\t",
      })[c],
  );
export function parseQuery(line) {
  return String(line)
    .split("|")
    .map((row) =>
      Object.fromEntries(
        row
          .trim()
          .split(/\s+/)
          .filter((v) => v.includes("="))
          .map((pair) => {
            const i = pair.indexOf("=");
            return [pair.slice(0, i), unescapeQuery(pair.slice(i + 1))];
          }),
      ),
    );
}
export function createEventListener(
  connection,
  onEvent,
  onState,
  { Client = ssh2.Client, channelId = 0 } = {},
) {
  let lastError = "",
    decoder = new StringDecoder("utf8"),
    client,
    stream,
    buffer = "",
    active,
    pending = [],
    closed = false,
    retry,
    ready = false,
    connecting = false,
    fingerprint = connection.sshFingerprint || "";
  const state = (status, error) => {
    if (closed) return;
    if (error) lastError = error;
    if (status === "connected") lastError = "";
    onState({
      status,
      error: error || lastError,
      channelId,
      fingerprint,
      updatedAt: new Date().toISOString(),
    });
  };
  const rejectPending = (error) => {
    if (active) {
      clearTimeout(active.timer);
      active.reject(error);
      active = null;
    }
    for (const entry of pending) entry.reject(error);
    pending = [];
  };
  function pump() {
    if (active || !stream || !pending.length) return;
    active = pending.shift();
    active.timer = setTimeout(() => {
      const error = new Error("SSH Query não respondeu em 10 segundos.");
      rejectPending(error);
      client?.destroy();
    }, 10000);
    stream.write(active.command + "\n");
  }
  function send(command) {
    return new Promise((resolve, reject) => {
      pending.push({ command, resolve, reject, rows: [] });
      pump();
    });
  }
  function data(chunk) {
    buffer += decoder.write(chunk);
    if (buffer.length > 1000000) {
      client.destroy();
      return;
    }
    let index;
    while ((index = buffer.indexOf("\n")) >= 0) {
      const line = buffer.slice(0, index).replace(/\r/g, "").trim();
      buffer = buffer.slice(index + 1);
      if (!line) continue;
      if (/^notify\w+\s/.test(line)) {
        const split = line.indexOf(" ");
        for (const event of parseQuery(line.slice(split + 1)))
          onEvent(line.slice(0, split), {
            ...event,
            __listenerChannel: String(channelId),
          });
        continue;
      }
      if (/^error id=/.test(line)) {
        const result = parseQuery(line)[0];
        if (active) {
          const entry = active;
          clearTimeout(entry.timer);
          active = null;
          if (Number(result.id) === 0)
            entry.resolve(entry.rows.flatMap(parseQuery));
          else
            entry.reject(
              new Error(
                `SSH Query ${result.id}: ${result.msg || "pedido recusado"}`,
              ),
            );
          pump();
        }
        continue;
      }
      if (active && line.includes("=") && !/^Welcome/i.test(line))
        active.rows.push(line);
    }
  }
  function reconnect() {
    ready = false;
    connecting = false;
    if (closed) return;
    state("disconnected");
    clearTimeout(retry);
    retry = setTimeout(connect, 10000);
    retry.unref?.();
  }
  function connect() {
    if (closed || connecting) return;
    connecting = true;
    state("connecting");
    client = new Client();
    client
      .on("error", (error) => {
        state("error", error.message);
        rejectPending(error);
      })
      .on("close", () => {
        stream = null;
        rejectPending(new Error("Ligação SSH fechada."));
        reconnect();
      })
      .on("ready", () =>
        client.shell(false, async (error, channel) => {
          if (error) {
            state("error", error.message);
            client.end();
            return;
          }
          stream = channel;
          buffer = "";
          decoder = new StringDecoder("utf8");
          channel.on("data", data);
          channel.on("close", () => {
            if (!closed) client.end();
          });
          channel.on("error", (error) => {
            state("error", error.message);
            client.destroy();
          });
          try {
            await send(`use sid=${connection.serverId}`);
            await send(
              `clientupdate client_nickname=XanaX-Botflow-${connection.serverId}-${channelId}`,
            );
            if (channelId) {
              const who = (await send("whoami"))[0];
              await send(
                `clientmove clid=${who.client_id || who.clid} cid=${channelId}`,
              );
              await send("servernotifyregister event=textchannel");
            } else {
              for (const event of [
                "server",
                "channel",
                "textprivate",
                "textserver",
              ])
                await send(
                  `servernotifyregister event=${event}${event === "channel" ? " id=0" : ""}`,
                );
            }
            ready = true;
            connecting = false;
            state("connected");
          } catch (error) {
            state("error", error.message);
            client.end();
          }
        }),
      );
    client.connect({
      host: connection.sshHost || new URL(connection.url).hostname,
      port: Number(connection.sshPort || 10022),
      username: connection.sshUsername,
      password: connection.sshPassword,
      readyTimeout: 10000,
      keepaliveInterval: 15000,
      keepaliveCountMax: 3,
      hostHash: "sha256",
      hostVerifier: (hash) => {
        if (fingerprint && fingerprint !== hash) {
          state(
            "error",
            "A impressão digital SSH mudou. Confirma o servidor na configuração da ligação.",
          );
          return false;
        }
        fingerprint = hash;
        return true;
      },
    });
  }
  connect();
  return {
    close() {
      closed = true;
      clearTimeout(retry);
      rejectPending(new Error("Escuta de eventos terminada."));
      client?.destroy();
    },
    get ready() {
      return ready;
    },
  };
}