Conat -- replacing NATS with socketio + sqlite3 + valkey + nodejs

.github/workflows/make-and-test.yml

@@ -94,5 +94,4 @@ jobs:
 
       - run: cd src && npm install -g pnpm
       - run: cd src && pnpm run make
-      - run: cd src && pnpm nats-server-daemon
       - run: cd src && pnpm run test

.gitignore

@@ -159,3 +159,5 @@ src/packages/frontend/i18n/trans/*.compiled.json
 **/*.db
 
 src/packages/project/nats/project-env.sh
+
+**/*.bash_history

src/compute/compute/lib/filesystem.ts

@@ -21,8 +21,8 @@ import {
 import { apiCall } from "@cocalc/api-client";
 import sendFiles from "./send-files";
 import getFiles from "./get-files";
-// ensure that the nats client is initialized so that syncfs can connect to nats properly.
-import "@cocalc/project/nats";
+// ensure that the conat client is initialized so that syncfs can connect properly.
+import "@cocalc/project/conat";
 
 const logger = getLogger("compute:filesystem");
 

src/compute/compute/lib/manager.ts

@@ -7,7 +7,7 @@ The manager does the following:
 */
 
 import debug from "debug";
-import startProjectServers from "@cocalc/project/nats";
+import startProjectServers from "@cocalc/project/conat";
 import { pingProjectUntilSuccess, waitUntilFilesystemIsOfType } from "./util";
 import { apiCall, project } from "@cocalc/api-client";
 

src/compute/compute/package.json

@@ -32,7 +32,7 @@
     "@cocalc/backend": "workspace:*",
     "@cocalc/compute": "link:",
     "@cocalc/jupyter": "workspace:*",
-    "@cocalc/nats": "workspace:*",
+    "@cocalc/conat": "workspace:*",
     "@cocalc/project": "workspace:*",
     "@cocalc/sync": "workspace:*",
     "@cocalc/sync-client": "workspace:*",

src/compute/conat

@@ -0,0 +1 @@
+../packages/conat

src/compute/pnpm-workspace.yaml

@@ -3,15 +3,15 @@ packages:
   - backend
   - comm
   - jupyter
-  - nats
+  - conat
   - project
   - sync
   - sync-fs
   - sync-client
   - util
   - terminal
   - compute
-  - nats
+  - conat
 
 onlyBuiltDependencies:
   - '@cocalc/fuse-native'

src/package.json

@@ -19,15 +19,8 @@
     "test-parallel": "unset DEBUG && pnpm run version-check && cd packages && pnpm run -r --parallel test",
     "test": "unset DEBUG && pnpm run version-check && cd packages && pnpm run -r test",
     "prettier-all": "cd packages/",
-    "nats-server": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main()\"",
-    "build-nats": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/util && pnpm install && pnpm build && cd ${COCALC_ROOT:=$INIT_CWD}/packages/nats && pnpm install && pnpm build && cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && pnpm install && pnpm build",
-    "nats-server-ci": "pnpm run build-nats && cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main()\"",
-    "nats-server-daemon": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main({daemon:true})\"",
-    "nats-server-verbose": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/install').main()\" && node -e \"require('@cocalc/backend/nats/conf').main()\" && node -e \"require('@cocalc/backend/nats/server').main({verbose:true})\"",
-    "nats-cli": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/cli').main()\"",
-    "nats-sys": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/backend && node -e \"require('@cocalc/backend/nats/cli').main({user:'sys'})\"",
-    "nats-tiered-storage": "cd ${COCALC_ROOT:=$INIT_CWD}/packages/server && DEBUG=cocalc:* DEBUG_CONSOLE=yes node -e \"require('@cocalc/server/nats/tiered-storage').init()\"",
-    "local-ci": "./scripts/ci.sh"
+    "local-ci": "./scripts/ci.sh",
+    "conat-server": "cd packages/server && pnpm conat-server"
   },
   "repository": {
     "type": "git",

src/packages/backend/auth/cookie-names.ts

@@ -39,3 +39,18 @@ export const ACCOUNT_ID_COOKIE_NAME = basePathCookieName({
   basePath,
   name: "account_id",
 });
+
+export const PROJECT_SECRET_COOKIE_NAME = basePathCookieName({
+  basePath,
+  name: "project_secret",
+});
+
+export const PROJECT_ID_COOKIE_NAME = basePathCookieName({
+  basePath,
+  name: "project_id",
+});
+
+export const HUB_PASSWORD_COOKIE_NAME = basePathCookieName({
+  basePath,
+  name: "hub_password",
+});

src/packages/backend/bin/conat-watch.cjs

@@ -0,0 +1,10 @@
+const { conat } = require('@cocalc/backend/conat')
+
+async function main() {
+    const subject = process.argv[2] ?? '>';
+    console.log("watching ", {subject})
+    const cn = await conat()
+    cn.watch(subject)
+}
+
+main();

src/packages/backend/conat/conat.ts

@@ -0,0 +1,17 @@
+import { conatPassword, conatServer } from "@cocalc/backend/data";
+import { connect, Client, type ClientOptions } from "@cocalc/conat/core/client";
+import { HUB_PASSWORD_COOKIE_NAME } from "@cocalc/backend/auth/cookie-names";
+import { inboxPrefix } from "@cocalc/conat/names";
+
+export type { Client };
+
+export function conat(options?: ClientOptions): Client {
+  return connect({
+    address: conatServer,
+    inboxPrefix: inboxPrefix({ hub_id: "hub" }),
+    extraHeaders: {
+      Cookie: `${HUB_PASSWORD_COOKIE_NAME}=${conatPassword}`,
+    },
+    ...options,
+  });
+}

src/packages/backend/conat/index.ts

@@ -0,0 +1,10 @@
+import getLogger from "@cocalc/backend/logger";
+import { setConatClient } from "@cocalc/conat/client";
+import { conat } from "./conat";
+
+export { conat };
+
+export function init() {
+  setConatClient({ conat: async (opts?) => conat(opts), getLogger });
+}
+init();

src/packages/backend/conat/persist.ts

@@ -0,0 +1,29 @@
+import "./index";
+import betterSqlite3 from "better-sqlite3";
+import { initContext } from "@cocalc/conat/persist/context";
+import { compress, decompress } from "zstd-napi";
+import { syncFiles } from "@cocalc/backend/data";
+import ensureContainingDirectoryExists from "@cocalc/backend/misc/ensure-containing-directory-exists";
+
+initContext({
+  betterSqlite3,
+  compress,
+  decompress,
+  syncFiles,
+  ensureContainingDirectoryExists,
+});
+
+export { pstream } from "@cocalc/conat/persist/storage";
+import { server } from "@cocalc/conat/persist/server";
+export { server };
+import { conat } from "./conat";
+
+let persistServer: any = undefined;
+
+export function initPersistServer() {
+  persistServer = server({ client: conat() });
+}
+export function close() {
+  persistServer?.end(); // end is a bit more graceful
+  persistServer = undefined;
+}

src/packages/backend/conat/sync.ts

@@ -0,0 +1,40 @@
+import {
+  dstream as createDstream,
+  type DStream,
+  type DStreamOptions as DstreamCreateOptions,
+} from "@cocalc/conat/sync/dstream";
+import { dkv as createDKV, type DKV, type DKVOptions } from "@cocalc/conat/sync/dkv";
+import { dko as createDKO, type DKO } from "@cocalc/conat/sync/dko";
+import { akv as createAKV, type AKV } from "@cocalc/conat/sync/akv";
+import { astream as createAStream, type AStream } from "@cocalc/conat/sync/astream";
+import { createOpenFiles, type OpenFiles } from "@cocalc/conat/sync/open-files";
+export { inventory } from "@cocalc/conat/sync/inventory";
+import "./index";
+
+export type { DStream, DKV, DKO, AKV };
+
+export async function dstream<T = any>(
+  opts: DstreamCreateOptions,
+): Promise<DStream<T>> {
+  return await createDstream<T>(opts);
+}
+
+export function astream<T = any>(opts: DstreamCreateOptions): AStream<T> {
+  return createAStream<T>(opts);
+}
+
+export async function dkv<T = any>(opts: DKVOptions): Promise<DKV<T>> {
+  return await createDKV<T>(opts);
+}
+
+export function akv<T = any>(opts: DKVOptions): AKV<T> {
+  return createAKV<T>(opts);
+}
+
+export async function dko<T = any>(opts: DKVOptions): Promise<DKO<T>> {
+  return await createDKO(opts);
+}
+
+export async function openFiles(project_id: string, opts?): Promise<OpenFiles> {
+  return await createOpenFiles({ project_id, ...opts });
+}