nats listings: packaged client

src/packages/backend/path-watcher.ts

@@ -54,7 +54,7 @@ const logger = getLogger("backend:path-watcher");
 const POLLING = true;
 
 const DEFAULT_POLL_MS = parseInt(
-  process.env.COCALC_FS_WATCHER_POLL_INTERVAL_MS ?? "3000",
+  process.env.COCALC_FS_WATCHER_POLL_INTERVAL_MS ?? "2000",
 );
 
 const ChokidarOpts: WatchOptions = {

src/packages/frontend/nats/client.ts

@@ -40,11 +40,7 @@ import type {
   CallNatsServiceFunction,
   CreateNatsServiceFunction,
 } from "@cocalc/nats/service";
-import {
-  createListingsClient,
-  getListingsKV,
-  getListingsTimesKV,
-} from "@cocalc/nats/service/listings";
+import { listingsClient } from "@cocalc/nats/service/listings";
 
 export class NatsClient {
   client: WebappClient;
@@ -440,28 +436,17 @@ export class NatsClient {
     return await dko<T>({ env: await this.getEnv(), ...opts });
   };
 
-  microservicesClient = async () => {
+  microservices = async () => {
     const nc = await this.getConnection();
     // @ts-ignore
     const svcm = new Svcm(nc);
     return svcm.client();
   };
 
-  listings = (opts: { project_id: string; compute_server_id?: number }) => {
-    return createListingsClient(opts);
-  };
-
-  getListingsKV = async (opts: {
-    project_id: string;
-    compute_server_id?: number;
-  }) => {
-    return await getListingsKV(opts);
-  };
-
-  getListingsTimesKV = async (opts: {
+  listings = async (opts: {
     project_id: string;
     compute_server_id?: number;
   }) => {
-    return await getListingsTimesKV(opts);
+    return await listingsClient(opts);
   };
 }

src/packages/nats/service/listings.ts

@@ -1,10 +1,11 @@
 /*
-Service for expressing interest in directory listings in a project or compute server.
+Service for watching directory listings in a project or compute server.
 */
 
 import { createServiceClient, createServiceHandler } from "./typed";
 import type { DirectoryListingEntry } from "@cocalc/util/types";
 import { dkv, type DKV } from "@cocalc/nats/sync/dkv";
+import { EventEmitter } from "events";
 
 // record info about at most this many files in a given directory
 export const MAX_FILES_PER_DIRECTORY = 10;
@@ -19,9 +20,9 @@ export const INTEREST_CUTOFF_MS = 1000 * 30;
 
 //export const INTEREST_CUTOFF_MS = 1000 * 60 * 10;
 
-interface ListingsApi {
+export interface ListingsApi {
   // cause the directory listing key:value store to watch path
-  interest: (path: string) => Promise<void>;
+  watch: (path: string) => Promise<void>;
 
   // just directly get the listing info now for this path
   getListing: (opts: {
@@ -88,7 +89,7 @@ export async function getListingsKV(
 export interface Times {
   // time last files for a given directory were attempted to be updated
   updated?: number;
-  // time user last expressed interest in a given directory
+  // time user requested to watch a given directory
   interest?: number;
 }
 
@@ -101,3 +102,55 @@ export async function getListingsTimesKV(
     ...opts,
   });
 }
+
+/* Unified interface to the above components for clients */
+
+export class ListingsClient extends EventEmitter {
+  options: { project_id: string; compute_server_id: number };
+  api: ListingsApi;
+  times: DKV<Times>;
+  listings: DKV<Listing>;
+
+  constructor({
+    project_id,
+    compute_server_id = 0,
+  }: {
+    project_id: string;
+    compute_server_id?: number;
+  }) {
+    super();
+    this.options = { project_id, compute_server_id };
+  }
+
+  init = async () => {
+    this.api = createListingsClient(this.options);
+    this.times = await getListingsTimesKV(this.options);
+    this.listings = await getListingsKV(this.options);
+    this.listings.on("change", (path) => this.emit("change", path));
+  };
+
+  get = (path: string): Listing | undefined => {
+    return this.listings.get(path);
+  };
+
+  getAll = () => this.listings.getAll();
+
+  close = () => {
+    this.times.close();
+    this.listings.close();
+  };
+
+  watch = async (path) => {
+    await this.api.watch(path);
+  };
+
+  getListing = async (opts) => {
+    return await this.api.getListing(opts);
+  };
+}
+
+export async function listingsClient(options) {
+  const C = new ListingsClient(options);
+  await C.init();
+  return C;
+}

src/packages/nats/service/service.ts

@@ -262,7 +262,16 @@ export async function waitForNatsService({
   let d = 100;
   let m = 100;
   const start = Date.now();
-  let ping = await pingNatsService({ options, maxWait: m });
+  const getPing = async (m: number) => {
+    try {
+      return await pingNatsService({ options, maxWait: m });
+    } catch {
+      // ping can fail, e.g, if not connected to nats at all or the ping
+      // service isn't up yet.
+      return [] as ServiceIdentity[];
+    }
+  };
+  let ping = await getPing(m);
   while (ping.length == 0) {
     d = Math.min(10000, d * 1.3);
     m = Math.min(1500, m * 1.3);
@@ -271,7 +280,7 @@ export async function waitForNatsService({
       throw Error("timeout");
     }
     await delay(d);
-    ping = await pingNatsService({ options, maxWait: m });
+    ping = await getPing(m);
   }
   return ping;
 }

src/packages/nats/sync/dkv.ts

@@ -47,7 +47,7 @@ export class DKV<T = any> extends EventEmitter {
     const kvname = jsName({ account_id, project_id });
     this.name = name + localLocationName(options);
     this.sha1 = env.sha1 ?? sha1;
-    this.prefix = this.sha1(name);
+    this.prefix = this.sha1(this.name);
     this.opts = {
       name: kvname,
       filter: `${this.prefix}.>`,

src/packages/nats/sync/kv.ts

@@ -45,7 +45,7 @@ export class KV<T = any> extends EventEmitter {
       throw Error("env must be defined");
     }
     this.sha1 = env.sha1 ?? sha1;
-    this.prefix = this.sha1(name);
+    this.prefix = this.sha1(this.name);
     this.generalKV = new GeneralKV({
       name: kvname,
       filter: `${this.prefix}.>`,

src/packages/project/nats/listings.ts

@@ -8,6 +8,21 @@
   few hundred (ordered by recent) files in that directory, all relative
   to the home directory.
 
+
+DEVELOPMENT:
+
+1. Setup project environment variables as usual
+
+2. Stop listings service running in the project:
+
+     define COCALC_PROJECT_ID, COCALC_NATS_JWT, etc., as usual.
+
+3. Start your own server
+
+.../src/packages/project/nats$ node
+
+> await require('@cocalc/project/nats/listings').init()
+
 */
 
 import getListing from "@cocalc/backend/get-listing";
@@ -34,11 +49,11 @@ let listings: Listings | null;
 
 const impl = {
   // cause the directory listing key:value store to watch path
-  interest: async (path: string) => {
+  watch: async (path: string) => {
     while (listings == null) {
       await delay(3000);
     }
-    listings.interest(path);
+    listings.watch(path);
   },
 
   getListing: async ({ path, hidden }) => {
@@ -164,8 +179,8 @@ class Listings {
     }
   };
 
-  interest = async (path: string) => {
-    logger.debug("interest", { path });
+  watch = async (path: string) => {
+    logger.debug("watch", { path });
     path = canonicalPath(path);
     this.times.set(path, { ...this.times.get(path), interest: Date.now() });
     this.updateListing(path);