simplex-chat/packages/simplex-chat-nodejs/tests/core.test.ts

import {execFile, spawnSync} from "child_process";
import * as crypto from "crypto";
import * as fs from "fs";
import * as path from "path";
import * as zlib from "zlib";
import {core} from "../src/index";
import * as libs from "../src/libs";
import * as simplex from "../src/simplex";

describe("Core tests", () => {
  let libPath: string;
  let addonPath: string;
  // the first run downloads libsimplex
  beforeAll(async () => {
    const libsDir = await libs.resolveLibsDir("sqlite");
    libPath = libs.libPath(libsDir);
    addonPath = await libs.resolveAddonPath();
    await core.loadLibrary("sqlite");
  }, 300000);
  const tmpDir = "./tests/tmp";
  const dbPath = path.join(tmpDir, "simplex_v1");
  const templateDir = "./tests/template";
  const templateDbPath = path.join(templateDir, "simplex_v1");

  beforeAll(async () => {
    fs.mkdirSync(templateDir, {recursive: true});
    await stopAndClose(await core.chatMigrateInit(templateDbPath, "key", core.MigrationConfirmation.YesUp));
  }, 30000);
  afterAll(() => fs.rmSync(templateDir, {recursive: true, force: true}));
  beforeEach(() => {
    fs.mkdirSync(tmpDir, {recursive: true});
    copyTemplateDb(dbPath);
  });
  afterEach(() => fs.rmSync(tmpDir, {recursive: true, force: true}));

  async function stopAndClose(ctrl: bigint): Promise<void> {
    await expect(core.chatSendCmd(ctrl, "/_stop")).resolves.toMatchObject({type: "chatStopped"});
    await core.chatCloseStore(ctrl);
  }

  function copyTemplateDb(dbFilePrefix: string): void {
    for (const suffix of ["_chat.db", "_agent.db"]) fs.copyFileSync(templateDbPath + suffix, dbFilePrefix + suffix);
  }

  it("should initialize chat controller", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);
    expect(typeof ctrl).toBe("bigint");
    await expect(stopAndClose(ctrl)).resolves.toBe(undefined);
    
    await expect(core.chatMigrateInit(dbPath, "wrong_key", core.MigrationConfirmation.YesUp)).rejects.toMatchObject({
      message: "Database or migration error (see dbMigrationError property)",
      dbMigrationError: expect.objectContaining({type: "errorNotADatabase"})
    });
  });

  it("should initialize chat controller with queue size", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp, 65536);
    expect(typeof ctrl).toBe("bigint");
    await expect(stopAndClose(ctrl)).resolves.toBe(undefined);

    await expect(core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp, 0)).rejects.toMatchObject({
      dbMigrationError: {type: "invalidQueueSize"}
    });
    await expect(core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp, 2 ** 31)).rejects.toThrow("Expected 32-bit integer queue size");
    await expect(core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp, 1.5)).rejects.toThrow("Expected 32-bit integer queue size");
  });

  it("should send command and receive event", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);

    await expect(core.chatSendCmd(ctrl, "/v")).resolves.toMatchObject({
      type: "versionInfo"
    });
    await expect(core.chatSendCmd(ctrl, '/debug event {"type": "chatSuspended"}')).resolves.toMatchObject({
      type: "cmdOk"
    });

    const wait = 500_000;
    await expect(core.chatRecvMsgWait(ctrl, wait)).resolves.toMatchObject({
      type: "chatSuspended"
    });    
    await expect(core.chatRecvMsgWait(ctrl, wait)).resolves.toBe(undefined);
    
    await expect(core.chatSendCmd(ctrl, "/unknown")).rejects.toMatchObject({
      message: "Chat command error (see chatError property)",
      chatError: expect.objectContaining({type: "error"})
    });

    await stopAndClose(ctrl);
  });

  it("should write/read encrypted file from/to buffer", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);

    const filePath = path.join(tmpDir, "write_file.txt");
    const buffer = new Uint8Array([0, 1, 2]).buffer;
    const cryptoArgs = await core.chatWriteFile(ctrl, filePath, buffer);
    expect(typeof cryptoArgs.fileKey).toBe("string");
    expect(typeof cryptoArgs.fileNonce).toBe("string");

    const buffer2 = await core.chatReadFile(filePath, cryptoArgs);
    expect(Buffer.from(buffer2).equals(Buffer.from(buffer))).toBe(true);

    await expect(core.chatWriteFile(ctrl, path.join(tmpDir, "unknown", "unknown.txt"), buffer)).rejects.toThrow();
    await expect(core.chatReadFile(path.join(tmpDir, "unknown.txt"), cryptoArgs)).rejects.toThrow();
    
    await stopAndClose(ctrl);
  });

  it("should write the view of a Uint8Array and read an empty file", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);

    const viewPath = path.join(tmpDir, "view.txt");
    const viewArgs = await core.chatWriteFile(ctrl, viewPath, Buffer.from("xxabcxx").subarray(2, 5));
    const view = await core.chatReadFile(viewPath, viewArgs);
    expect(Buffer.isBuffer(view)).toBe(true);
    expect(view.toString()).toBe("abc");

    const emptyPath = path.join(tmpDir, "empty.txt");
    const emptyArgs = await core.chatWriteFile(ctrl, emptyPath, new Uint8Array(0));
    const empty = await core.chatReadFile(emptyPath, emptyArgs);
    expect(Buffer.isBuffer(empty)).toBe(true);
    expect(empty.length).toBe(0);

    await stopAndClose(ctrl);
  });

  it("should encrypt/decrypt file", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);

    const unencryptedPath = path.join(tmpDir, "file_unencrypted.txt");
    fs.writeFileSync(unencryptedPath, "unencrypted\n");
    const encryptedPath = path.join(tmpDir, "file_encrypted.txt");
    const cryptoArgs = await core.chatEncryptFile(ctrl, unencryptedPath, encryptedPath);
    expect(typeof cryptoArgs.fileKey).toBe("string");
    expect(typeof cryptoArgs.fileNonce).toBe("string");

    const decryptedPath: string = path.join(tmpDir, "file_decrypted.txt");
    await expect(core.chatDecryptFile(encryptedPath, cryptoArgs, decryptedPath)).resolves.toBe(undefined);

    expect(fs.readFileSync(decryptedPath, "utf8")).toBe("unencrypted\n");

    await expect(core.chatEncryptFile(ctrl, path.join(tmpDir, "unknown.txt"), encryptedPath)).rejects.toThrow();
    await expect(core.chatDecryptFile(path.join(tmpDir, "unknown.txt"), cryptoArgs, decryptedPath)).rejects.toThrow();

    await stopAndClose(ctrl);
  });

  describe("image resizing", () => {
    const notImage = Buffer.from("not an image");
    const resizeError = "image cannot be decoded or does not fit maxSize";
    const oldLibraryError = (name: string) => `loaded libsimplex does not export ${name}; image resizing needs a newer libsimplex`;
    let imageResizing = false;

    beforeAll(async () => {
      const error = await core.chatResizeImageUri(notImage, 14000).then(() => "", (e: Error) => e.message);
      imageResizing = error !== oldLibraryError("chat_resize_image_uri");
    });

    function noisePng(width: number, height: number, alpha = false): Buffer {
      const rowLength = width * (alpha ? 4 : 3) + 1;
      const rows = crypto.randomBytes(rowLength * height);
      for (let y = 0; y < height; y++) rows[y * rowLength] = 0;
      const header = Buffer.alloc(13);
      header.writeUInt32BE(width, 0);
      header.writeUInt32BE(height, 4);
      header.set([8, alpha ? 6 : 2, 0, 0, 0], 8);
      const chunk = (type: string, data: Buffer) => {
        const body = Buffer.concat([Buffer.from(type), data]);
        const length = Buffer.alloc(4);
        length.writeUInt32BE(data.length);
        const crc = Buffer.alloc(4);
        crc.writeUInt32BE(zlib.crc32(body));
        return Buffer.concat([length, body, crc]);
      };
      const signature = Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]);
      return Buffer.concat([signature, chunk("IHDR", header), chunk("IDAT", zlib.deflateSync(rows)), chunk("IEND", Buffer.alloc(0))]);
    }

    it("should resize image to data URI", async () => {
      const resize = core.chatResizeImageUri(noisePng(256, 256), 14000);
      if (!imageResizing) return expect(resize).rejects.toThrow(oldLibraryError("chat_resize_image_uri"));
      const uri = await resize;
      expect(uri.startsWith("data:image/jpg;base64,")).toBe(true);
      expect(uri.length).toBeLessThanOrEqual(14000);
    });

    it("should resize the view of a Uint8Array to JPEG and PNG", async () => {
      const image = noisePng(256, 256);
      const view = Buffer.concat([Buffer.alloc(3), image]).subarray(3);
      const resize = core.chatResizeImage(view, 20000, "jpeg");
      if (!imageResizing) return expect(resize).rejects.toThrow(oldLibraryError("chat_resize_image"));
      const jpeg = await resize;
      expect(jpeg.subarray(0, 3)).toEqual(Buffer.from([0xff, 0xd8, 0xff]));
      expect(jpeg.length).toBeLessThanOrEqual(20000);
      const png = await core.chatResizeImage(new Uint8Array(image).buffer, 20000, "png");
      expect(png.subarray(0, 8)).toEqual(image.subarray(0, 8));
      expect(png.length).toBeLessThanOrEqual(20000);
    });

    it("should resize image to each ImageSize", async () => {
      const image = noisePng(512, 512);
      const resize = core.resizeImage(image, core.ImageSize.Image);
      if (!imageResizing) return expect(resize).rejects.toThrow(oldLibraryError("chat_resize_image"));
      const jpeg = await resize;
      expect(jpeg.subarray(0, 3)).toEqual(Buffer.from([0xff, 0xd8, 0xff]));
      expect(jpeg.length).toBeLessThanOrEqual(261120);
      const profile = await core.resizeImage(image, core.ImageSize.ProfilePreview);
      expect(profile.startsWith("data:image/jpg;base64,")).toBe(true);
      expect(profile.length).toBeLessThanOrEqual(12500);
      const preview = await core.resizeImage(image, core.ImageSize.MessagePreview);
      expect(preview.startsWith("data:image/jpg;base64,")).toBe(true);
      expect(preview.length).toBeLessThanOrEqual(14000);
    });

    it("should resize image with transparency to PNG for each ImageSize", async () => {
      const image = noisePng(256, 256, true);
      const resize = core.resizeImage(image, core.ImageSize.Image);
      if (!imageResizing) return expect(resize).rejects.toThrow(oldLibraryError("chat_resize_image"));
      const png = await resize;
      expect(png.subarray(0, 8)).toEqual(image.subarray(0, 8));
      expect(png.length).toBeLessThanOrEqual(261120);
      const profile = await core.resizeImage(image, core.ImageSize.ProfilePreview);
      expect(profile.startsWith("data:image/png;base64,")).toBe(true);
      expect(profile.length).toBeLessThanOrEqual(12500);
      const preview = await core.resizeImage(image, core.ImageSize.MessagePreview);
      expect(preview.startsWith("data:image/png;base64,")).toBe(true);
      expect(preview.length).toBeLessThanOrEqual(14000);
    });

    it("should reject data that is not an image", async () => {
      await expect(core.chatResizeImage(notImage, 20000, "jpeg")).rejects.toThrow(imageResizing ? resizeError : oldLibraryError("chat_resize_image"));
    });

    it("should reject maxSize outside 32-bit integers", async () => {
      const error = imageResizing ? "Expected 32-bit integer maxSize" : oldLibraryError("chat_resize_image_uri");
      await expect(core.chatResizeImageUri(noisePng(8, 8), 2 ** 31)).rejects.toThrow(error);
    });
  });

  it("should not block the libuv pool while receiving", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);
    const receives = [1, 2, 3, 4].map(() => core.chatRecvMsgWait(ctrl, 500_000));
    const start = Date.now();
    await fs.promises.stat(tmpDir);
    expect(Date.now() - start).toBeLessThan(200);
    // one receiver thread per controller runs the 4 receives in turn
    await Promise.all(receives);
    await stopAndClose(ctrl);
  }, 30000);

  const itOnLinux = process.platform === "linux" ? it : it.skip;

  // Thread count is read from /proc/self/task, which only exists on Linux.
  itOnLinux("should keep the thread count constant while receiving", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);
    for (let i = 0; i < 10; i++) await core.chatRecvMsgWait(ctrl, 1);
    const threadCount = () => fs.readdirSync("/proc/self/task").length;
    const warmCount = threadCount();
    const counts = new Set<number>();
    for (let i = 0; i < 200; i++) {
      await core.chatRecvMsgWait(ctrl, 1);
      counts.add(threadCount());
    }
    expect({warmCount, counts: [...counts]}).toEqual({warmCount, counts: [warmCount]});
    await stopAndClose(ctrl);
  }, 30000);

  it("should receive on two controllers concurrently", async () => {
    const dbPathB = path.join(tmpDir, "simplex_b");
    copyTemplateDb(dbPathB);
    const ctrlA = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);
    const ctrlB = await core.chatMigrateInit(dbPathB, "key", core.MigrationConfirmation.YesUp);
    const start = Date.now();
    await expect(Promise.all([core.chatRecvMsgWait(ctrlA, 1_000_000), core.chatRecvMsgWait(ctrlB, 1_000_000)]))
      .resolves.toEqual([undefined, undefined]);
    const elapsed = Date.now() - start;
    expect(elapsed).toBeGreaterThanOrEqual(900);
    expect(elapsed).toBeLessThan(1800);
    await stopAndClose(ctrlA);
    await stopAndClose(ctrlB);
  });

  it("should receive events of one controller in order", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);
    for (const action of ["first", "second"]) {
      await expect(core.chatSendCmd(ctrl, `/debug event {"type": "timedAction", "action": "${action}", "durationMilliseconds": 1}`))
        .resolves.toMatchObject({type: "cmdOk"});
    }
    const events = await Promise.all([core.chatRecvMsgWait(ctrl, 500_000), core.chatRecvMsgWait(ctrl, 500_000)]);
    expect(events).toMatchObject([
      {type: "timedAction", action: "first"},
      {type: "timedAction", action: "second"}
    ]);
    await stopAndClose(ctrl);
  });

  it("should close the store while a receive is in flight", async () => {
    const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);
    // starts the receiver thread, so the next request only has to wake it
    await core.chatRecvMsgWait(ctrl, 1);
    const settle = (p: Promise<unknown>) => p.then((event) => ({event}), (e: Error) => ({error: e.message}));
    const receives = Promise.all([settle(core.chatRecvMsgWait(ctrl, 3_000_000)), settle(core.chatRecvMsgWait(ctrl, 3_000_000))]);
    // the thread enters the first receive within this margin even under load, so the second one is still queued at close
    await new Promise((resolve) => setTimeout(resolve, 500));
    await expect(core.chatSendCmd(ctrl, "/_stop")).resolves.toMatchObject({type: "chatStopped"});
    let closed = false;
    const close = core.chatCloseStore(ctrl).then(() => { closed = true; });
    const timerStart = Date.now();
    const timerDelay = await new Promise<number>((resolve) => setTimeout(() => resolve(Date.now() - timerStart), 10));
    expect({closed, timerDelayBelow100ms: timerDelay < 100}).toEqual({closed: false, timerDelayBelow100ms: true});
    await close;
    expect(await receives).toEqual([{event: undefined}, {error: "chat receiver stopped"}]);
  });

  it("should accept loading libsimplex again from the same path", () => {
    expect(() => simplex.load(addonPath, libPath)).not.toThrow();
  });

  it("should refuse to load libsimplex from another path", () => {
    expect(() => simplex.load(addonPath, path.resolve("other", "libsimplex.so"))).toThrow(`libsimplex already loaded from ${libPath}`);
  });

  it("should load libsimplex after a failed load and in several workers", () => {
    const addon = path.resolve(__dirname, "..", "build", "Release", "simplex.node");
    const childDbPath = path.resolve(tmpDir, "simplex_workers");
    const script = `
      const {Worker} = require("worker_threads");
      const simplex = require(${JSON.stringify(addon)});
      try { simplex.load(${JSON.stringify(addon)}) } catch (e) { console.log(JSON.stringify(e.message.endsWith("does not export hs_init_with_rtsopts"))) }
      const load = ${JSON.stringify(`require(${JSON.stringify(addon)}).load(${JSON.stringify(libPath)})`)};
      const workers = Array.from({length: 4}, () => new Worker(load, {eval: true}));
      Promise.all(workers.map((w) => new Promise((resolve, reject) => w.on("exit", resolve).on("error", reject))))
        .then(async (codes) => {
          simplex.load(${JSON.stringify(libPath)});
          const [ctrl] = await simplex.chat_migrate_init(${JSON.stringify(childDbPath)}, "key", "yesUp");
          await simplex.chat_send_cmd(ctrl, "/_stop");
          console.log(JSON.stringify({codes, closed: await simplex.chat_close_store(ctrl) === ""}));
        });
    `;
    const child = spawnSync(process.execPath, ["-e", script], {timeout: 30000, encoding: "utf8"});
    if (child.status !== 0 || child.signal !== null) console.log("child stderr:", child.stderr);
    expect({status: child.status, signal: child.signal, stdout: child.stdout.trim()})
      .toEqual({status: 0, signal: null, stdout: 'true\n{"codes":[0,0,0,0],"closed":true}'});
  }, 35000);

  it("should let the process exit while a receiver is idle", () => {
    const childDbPath = path.resolve(dbPath);
    const script = `
      const simplex = require("./build/Release/simplex.node");
      simplex.load(${JSON.stringify(libPath)});
      simplex.chat_migrate_init(${JSON.stringify(childDbPath)}, "key", "yesUp")
        .then(([ctrl]) => simplex.chat_recv_msg_wait(ctrl, 1))
        .then((res) => console.log("received " + JSON.stringify(res)));
    `;
    const child = spawnSync(process.execPath, ["-e", script], {cwd: path.join(__dirname, ".."), timeout: 25000, encoding: "utf8"});
    if (child.status !== 0 || child.signal !== null) console.log("child stderr:", child.stderr);
    expect({status: child.status, signal: child.signal, stdout: child.stdout.trim()})
      .toEqual({status: 0, signal: null, stdout: 'received ""'});
  });

  it("should not crash when closing stopped controllers repeatedly", async () => {
    const script = `
      const fs = require("fs"), path = require("path");
      const simplex = require("./build/Release/simplex.node");
      simplex.load(${JSON.stringify(libPath)});
      (async () => {
        // a new database runs every migration, each committed with a disk flush that takes up to 30 ms on CI runners
        const dir = fs.mkdtempSync(path.join(${JSON.stringify(path.resolve(tmpDir))}, "close-"));
        for (let i = 0; i < 40; i++) {
          const [ctrl] = await simplex.chat_migrate_init(path.join(dir, "simplex"), "key", "yesUp");
          await simplex.chat_send_cmd(ctrl, "/v");
          await simplex.chat_send_cmd(ctrl, "/_stop");
          const res = await simplex.chat_close_store(ctrl);
          if (res !== "") throw new Error("close failed: " + res);
        }
        fs.rmSync(dir, {recursive: true, force: true});
      })();
    `;
    const runChild = () => new Promise<{code: number | null, signal: NodeJS.Signals | null, stderr: string}>((resolve) => {
      const child = execFile(process.execPath, ["-e", script], {cwd: path.join(__dirname, ".."), timeout: 150000}, (_error, _stdout, stderr) =>
        resolve({code: child.exitCode, signal: child.signalCode, stderr}));
    });
    const childCount = 3;
    const results = await Promise.all(Array.from({length: childCount}, runChild));
    for (const r of results) if (r.code !== 0 || r.signal !== null) console.log("child stderr:", r.stderr);
    expect(results.map(({code, signal}) => ({code, signal}))).toEqual(Array(childCount).fill({code: 0, signal: null}));
  }, 180000);
});