Merge remote-tracking branch 'origin/master' into conat-optimize1

Author: haraldschilly

src/packages/conat/core/client.ts

@@ -1450,14 +1450,13 @@ export class Client extends EventEmitter {
   sync = {
     dkv: async <T,>(opts: DKVOptions): Promise<DKV<T>> =>
       await dkv<T>({ ...opts, client: this }),
-    akv: async <T,>(opts: DKVOptions): Promise<AKV<T>> =>
-      await akv<T>({ ...opts, client: this }),
+    akv: <T,>(opts: DKVOptions): AKV<T> => akv<T>({ ...opts, client: this }),
     dko: async <T,>(opts: DKVOptions): Promise<DKO<T>> =>
       await dko<T>({ ...opts, client: this }),
     dstream: async <T,>(opts: DStreamOptions): Promise<DStream<T>> =>
       await dstream<T>({ ...opts, client: this }),
-    astream: async <T,>(opts: DStreamOptions): Promise<AStream<T>> =>
-      await astream<T>({ ...opts, client: this }),
+    astream: <T,>(opts: DStreamOptions): AStream<T> =>
+      astream<T>({ ...opts, client: this }),
     synctable: async (opts: SyncTableOptions): Promise<ConatSyncTable> =>
       await createSyncTable({ ...opts, client: this }),
   };

src/packages/conat/hub/api/sync.ts

@@ -1,11 +1,43 @@
 import { authFirst } from "./util";
 
+export interface Patch {
+  seq: number;
+  time: number;
+  mesg: {
+    time: number;
+    wall: number;
+    patch: string;
+    user_id: number;
+    is_snapshot?: boolean;
+    parents: number[];
+    version?: number;
+  };
+}
+
+export interface HistoryInfo {
+  doctype: string;
+  init: { time: Date; size: number; error: string };
+  last_active: Date;
+  path: string;
+  project_id: string;
+  read_only: boolean;
+  save: {
+    state: string;
+    error: string;
+    hash: number;
+    time: number;
+    expected_hash: number;
+  };
+  string_id: string;
+  users: string[];
+}
+
 export interface Sync {
   history: (opts: {
     account_id?: string;
     project_id: string;
     path: string;
-  }) => Promise<any[]>;
+  }) => Promise<{ patches: Patch[] }>;
 }
 
 export const sync = {

src/packages/conat/hub/changefeeds/util.ts

@@ -6,18 +6,18 @@ export const SUBJECT = "changefeeds.*";
 // If the user refreshes their browser, it is still about a minute
 // before all the changefeeds they had open are free (due to the
 // SERVER_KEEPALIVE time below).
-export const MAX_PER_ACCOUNT = 500;
-export const MAX_GLOBAL = 10000;
+export const MAX_PER_ACCOUNT = 1_000;
+export const MAX_GLOBAL = 50_000;
 
 const DEBUG_DEVEL_MODE = false;
 
-export let CLIENT_KEEPALIVE = 90000;
-export let SERVER_KEEPALIVE = 45000;
-export let KEEPALIVE_TIMEOUT = 10000;
+export let CLIENT_KEEPALIVE = 90_000;
+export let SERVER_KEEPALIVE = 45_000;
+export let KEEPALIVE_TIMEOUT = 15_000;
 
 if (DEBUG_DEVEL_MODE) {
   console.log(
-    "*** WARNING: Using DEBUB_DEVEL_MODE changefeed parameters!! ***",
+    "*** WARNING: Using DEBUG_DEVEL_MODE changefeed parameters!! ***",
   );
   CLIENT_KEEPALIVE = 6000;
   SERVER_KEEPALIVE = 3000;

src/packages/conat/sync/akv.ts

@@ -141,6 +141,15 @@ export class AKV<T = any> {
     });
   };
 
+  // WARNING/TODO: this getAll implementation is not at all clever!
+  //   getAll = async () => {
+  //     const v: { [key: string]: T } = {};
+  //     for (const key of await this.keys()) {
+  //       v[key] = await this.get(key);
+  //     }
+  //     return v;
+  //   };
+
   sqlite = async (
     statement: string,
     params?: any[],

src/packages/conat/sync/astream.ts

@@ -99,7 +99,7 @@ export class AStream<T = any> {
   // or network is flaky, but will return all data properly in order
   // then throw an exception with code 503 rather than returning data
   // with something skipped.
-  async *getAll(opts): AsyncGenerator<
+  async *getAll(opts?): AsyncGenerator<
     {
       mesg: T;
       headers?: Headers;

src/packages/server/conat/api/sync.ts

@@ -1,20 +1,52 @@
-//import { conat } from "@cocalc/backend/conat";
+import { conat } from "@cocalc/backend/conat";
 import isCollaborator from "@cocalc/server/projects/is-collaborator";
+import { patchesStreamName } from "@cocalc/conat/sync/synctable-stream";
+import { type Patch, type HistoryInfo } from "@cocalc/conat/hub/api/sync";
+import { client_db } from "@cocalc/util/db-schema/client-db";
 
 export async function history({
   account_id,
   project_id,
   path,
+  start_seq = 0,
+  end_seq,
 }: {
   account_id?: string;
   project_id: string;
   path: string;
-}): Promise<any[]> {
+  start_seq?: number;
+  end_seq?: number;
+}): Promise<{ patches: Patch[]; info: HistoryInfo }> {
   if (!account_id || !(await isCollaborator({ account_id, project_id }))) {
     throw Error("user must be collaborator on source project");
   }
 
-  // const client = conat();
-  // this will be much easier once the fs2 branch is merged
-  throw Error(`not implemented yet -- can't get history of ${path} yet`);
+  const client = conat();
+  const name = patchesStreamName({ project_id, path });
+  const astream = client.sync.astream({
+    name,
+    project_id,
+    noInventory: true,
+  });
+  const patches: Patch[] = [];
+  for await (const patch of await astream.getAll({
+    start_seq,
+    end_seq,
+  })) {
+    patches.push(patch as any);
+  }
+
+  const akv = client.sync.akv({
+    name: `__dko__syncstrings:${client_db.sha1(project_id, path)}`,
+    project_id,
+    noInventory: true,
+  });
+  const keys = await akv.keys();
+  const info: Partial<HistoryInfo> = {};
+  for (const key of keys) {
+    if (key[0] != "[") continue;
+    info[JSON.parse(key)[1]] = await akv.get(key);
+  }
+
+  return { patches, info: info as HistoryInfo };
 }