nats api: a little bit of refactoring

docs/nats/devlog.md

@@ -254,17 +254,17 @@ Plan.
 
 This is critical to solve.  This sucks now.   This is key to eliminating "hub\-websocket".  This might be very easy.  Here's the plan:
 
-- make a request/response listener that listens on hub.account.{account\_id} and hub.db.project.{project\_id} for a db query.
-- if changes is false, just responds with the result of the query.
-- if changes is true, get kv store k named  `account-{account_id}` or `project-{project_id}` \(which might be used by project or compute server\).
+- [x] make a request/response listener that listens on hub.account.{account\_id} and hub.db.project.{project\_id} for a db query.
+- [x] if changes is false, just responds with the result of the query.
+- [ ] if changes is true, get kv store k named  `account-{account_id}` or `project-{project_id}` \(which can be used by project or compute server\).
   - let id be the sha1 hash of the query \(and options\)
   - k.id.update is less than X seconds ago, do nothing... it's already being updated by another server.
   - do the query to the database \(with changes true\)
   - write the results into k under k.id.data.key = value.
   - keep watching for changes so long as k.id.interest is at most n\*X seconds ago.
   - Also set k.id.update to now.
   - return id
-- another message to `hub.db.{account_id}` which contains a list of id's.
+- [ ] another message to `hub.db.{account_id}` which contains a list of id's.
   - When get this one, update k.id.interest to now for each of the id's.
 
 With the above algorithm, it should be very easy to reimplement the client side of SyncTable.  Moreover, there are many advantages:

src/packages/database/nats/index.ts

@@ -7,9 +7,9 @@
 
 import getLogger from "@cocalc/backend/logger";
 import { JSONCodec } from "nats";
-import { isValidUUID } from "@cocalc/util/misc";
 import userQuery from "@cocalc/database/user-query";
 import { getConnection } from "@cocalc/backend/nats";
+import { getUserId } from "@cocalc/nats/api";
 
 const logger = getLogger("database:nats");
 
@@ -31,22 +31,7 @@ async function handleRequest(mesg) {
   console.log({ subject: mesg.subject });
   let resp;
   try {
-    const segments = mesg.subject.split(".");
-    const uuid = segments[2];
-    if (!isValidUUID(uuid)) {
-      throw Error(`invalid uuid '${uuid}'`);
-    }
-    const type = segments[1]; // 'project' or 'account'
-    let account_id, project_id;
-    if (type == "project") {
-      project_id = uuid;
-      account_id = undefined;
-    } else if (type == "account") {
-      project_id = undefined;
-      account_id = uuid;
-    } else {
-      throw Error("must be project or account");
-    }
+    const { account_id, project_id } = getUserId(mesg.subject);
     const { name, args } = jc.decode(mesg.data) ?? ({} as any);
     if (!name) {
       throw Error("api endpoint name must be given in message");

src/packages/database/package.json

@@ -21,6 +21,7 @@
     "@cocalc/backend": "workspace:*",
     "@cocalc/database": "workspace:*",
     "@cocalc/util": "workspace:*",
+    "@cocalc/nats": "workspace:*",
     "@types/lodash": "^4.14.202",
     "@types/pg": "^8.6.1",
     "@types/uuid": "^8.3.1",

src/packages/database/tsconfig.json

@@ -7,7 +7,8 @@
   },
   "exclude": ["node_modules", "dist"],
   "references": [
-    { "path": "../util" },
     { "path": "../backend" },
+    { "path": "../nats" },
+    { "path": "../util" }
   ]
 }

src/packages/frontend/client/nats.ts

@@ -63,6 +63,7 @@ export class NatsClient {
         name,
         args,
       }),
+      { timeout: 5000 },
     );
     return this.jc.decode(resp.data);
   };

src/packages/nats/api/index.ts

@@ -1,4 +1,5 @@
 import type { Customize } from "@cocalc/util/db-schema/server-settings";
+import { isValidUUID } from "@cocalc/util/misc";
 
 export interface HubApi {
   getCustomize: (fields?: string[]) => Promise<Customize>;
@@ -16,3 +17,28 @@ export function initHubApi(callHubApi): HubApi {
   }
   return hubApi as HubApi;
 }
+
+type UserId =
+  | {
+      account_id: string;
+      project_id: undefined;
+    }
+  | {
+      account_id: undefined;
+      project_id: string;
+    };
+export function getUserId(subject: string): UserId {
+  const segments = subject.split(".");
+  const uuid = segments[2];
+  if (!isValidUUID(uuid)) {
+    throw Error(`invalid uuid '${uuid}'`);
+  }
+  const type = segments[1]; // 'project' or 'account'
+  if (type == "project") {
+    return { project_id: uuid } as UserId;
+  } else if (type == "account") {
+    return { account_id: uuid } as UserId;
+  } else {
+    throw Error("must be project or account");
+  }
+}

src/packages/nats/package.json

@@ -4,6 +4,7 @@
   "description": "CoCalc NATS integration code.  Usable by both nodejs and browser.",
   "exports": {
     "./sync/*": "./dist/sync/*.js",
+    "./api": "./dist/api/index.js",
     "./api/*": "./dist/api/*.js"
   },
   "scripts": {
@@ -13,17 +14,9 @@
     "test": "pnpm exec jest",
     "prepublishOnly": "pnpm test"
   },
-  "files": [
-    "dist/**",
-    "README.md",
-    "package.json"
-  ],
+  "files": ["dist/**", "README.md", "package.json"],
   "author": "SageMath, Inc.",
-  "keywords": [
-    "utilities",
-    "nats",
-    "cocalc"
-  ],
+  "keywords": ["utilities", "nats", "cocalc"],
   "license": "SEE LICENSE.md",
   "dependencies": {
     "@cocalc/nats": "workspace:*",

src/packages/server/nats/api.ts

@@ -26,8 +26,7 @@ To view all requests (and replies) in realtime:
 
 import { JSONCodec } from "nats";
 import getLogger from "@cocalc/backend/logger";
-import { isValidUUID } from "@cocalc/util/misc";
-import { type HubApi } from "@cocalc/nats/api/index";
+import { type HubApi, getUserId } from "@cocalc/nats/api/index";
 
 const logger = getLogger("server:nats:api");
 
@@ -48,19 +47,16 @@ async function handleApiRequest(mesg) {
   console.log({ subject: mesg.subject });
   let resp;
   try {
-    const segments = mesg.subject.split(".");
-    const type = segments[1];
-    if (type != "account") {
-      throw Error("only type='account' is supported now");
-    }
-    const account_id = segments[2];
-    if (!isValidUUID(account_id)) {
-      throw Error(`invalid account_id '${account_id}'`);
-    }
+    const { account_id, project_id } = getUserId(mesg.subject);
     const request = jc.decode(mesg.data) ?? {};
     const { name, args } = request as any;
-    logger.debug("handling hub.api request:", { account_id, name, args });
-    resp = await getResponse({ name, args, account_id });
+    logger.debug("handling hub.api request:", {
+      account_id,
+      project_id,
+      name,
+      args,
+    });
+    resp = await getResponse({ name, args, account_id, project_id });
   } catch (err) {
     resp = { error: `${err}` };
   }
@@ -75,13 +71,17 @@ const hubApi: HubApi = {
   userQuery,
 };
 
-async function getResponse({ name, args, account_id }) {
+async function getResponse({ name, args, account_id, project_id }) {
   const f = hubApi[name];
   if (f == null) {
     throw Error(`unknown function '${name}'`);
   }
   if (name == "userQuery" && args[0] != null) {
-    args[0].account_id = account_id;
+    if (account_id) {
+      args[0].account_id = account_id;
+    } else if (project_id) {
+      args[0].project_id = project_id;
+    }
   }
   return await f(...args);
 }

src/packages/server/nats/auth.ts

@@ -120,6 +120,9 @@ export async function configureNatsUser(cocalcUser: CoCalcUser) {
   const goalSub = new Set(["_INBOX.>"]);
 
   if (userType == "account") {
+
+    goalSub.add(`$KV.account-${userId}.>`);
+
     const pool = getPool();
     // all RUNNING projects with the user's group
     const query = `SELECT project_id, users#>>'{${userId},group}' AS group FROM projects WHERE state#>>'{state}'='running' AND users ? '${userId}' ORDER BY project_id`;