1
/*
2
 *  This file is part of CoCalc: Copyright © 2020 Sagemath, Inc.
3
 *  License: MS-RSL – see LICENSE.md for details
4
 */
5

6
/*
7
NOTE: Like much of our Jupyter-related code in CoCalc,
8
the code in this file is very much run in *both* the
9
frontend web browser and backend project server.
10
*/
11

12
import { EventEmitter } from "events";
13
import { Map as iMap } from "immutable";
14
import {
15
  close,
16
  delete_null_fields,
17
  is_object,
18
  len,
19
  auxFileToOriginal,
20
  sha1,
21
} from "@cocalc/util/misc";
22
import { SyncDoc } from "./sync-doc";
23
import { SyncTable } from "@cocalc/sync/table/synctable";
24
import { Client } from "./types";
25
import { debounce } from "lodash";
26

27
type State = "init" | "ready" | "closed";
28

29
type Value = { [key: string]: any };
30

31
// When there is no activity for this much time, them we
32
// do some garbage collection.  This is only done in the
33
// backend project, and not by frontend browser clients.
34
// The garbage collection is deleting models and related
35
// data when they are not referenced in the notebook.
36
// Also, we don't implement complete object delete yet so instead we
37
// set the data field to null, which clears all state about and
38
// object and makes it easy to know to ignore it.
39
const GC_DEBOUNCE_MS = 10000;
40

41
// If for some reason GC needs to be deleted, e.g., maybe you
42
// suspect a bug, just toggle this flag.  In particular, note
43
// includeThirdPartyReferences below that has to deal with a special
44
// case schema that k3d uses for references, which they just made up,
45
// which works with official upstream, since that has no garbage
46
// collection.
47
const DISABLE_GC = false;
48

49
// ignore messages past this age.
50
const MAX_MESSAGE_TIME_MS = 10000;
51

52
interface CommMessage {
53
  header: { msg_id: string };
54
  parent_header: { msg_id: string };
55
  content: any;
56
  buffers: any[];
57
}
58

59
export interface Message {
60
  // don't know yet...
61
}
62

63
export type SerializedModelState = { [key: string]: any };
64

65
export class IpywidgetsState extends EventEmitter {
66
  private syncdoc: SyncDoc;
67
  private client: Client;
68
  private table: SyncTable;
69
  private state: State = "init";
70
  private table_options: any[] = [];
71
  private create_synctable: Function;
72
  private gc: Function;
73

74
  // TODO: garbage collect this, both on the frontend and backend.
75
  // This should be done in conjunction with the main table (with gc
76
  // on backend, and with change to null event on the frontend).
77
  private buffers: {
78
    [model_id: string]: {
79
      [path: string]: { buffer: Buffer; hash: string };
80
    };
81
  } = {};
82
  // Similar but used on frontend
83
  private arrayBuffers: {
84
    [model_id: string]: {
85
      [path: string]: { buffer: ArrayBuffer; hash: string };
86
    };
87
  } = {};
88

89
  // If capture_output[msg_id] is defined, then
90
  // all output with that msg_id is captured by the
91
  // widget with given model_id.   This data structure
92
  // is ONLY used in the project, and is not synced
93
  // between frontends and project.
94
  private capture_output: { [msg_id: string]: string[] } = {};
95

96
  // If the next output should be cleared.  Use for
97
  // clear_output with wait=true.
98
  private clear_output: { [model_id: string]: boolean } = {};
99

100
  constructor(syncdoc: SyncDoc, client: Client, create_synctable: Function) {
101
    super();
102
    this.syncdoc = syncdoc;
103
    this.client = client;
104
    this.create_synctable = create_synctable;
105
    this.table_options = [{ ephemeral: true }];
106
    this.gc =
107
      !DISABLE_GC && client.is_project() // no-op if not project or DISABLE_GC
108
        ? debounce(() => {
109
            // return; // temporarily disabled since it is still too aggressive
110
            if (this.state == "ready") {
111
              this.deleteUnused();
112
            }
113
          }, GC_DEBOUNCE_MS)
114
        : () => {};
115
  }
116

117
  init = async (): Promise<void> => {
118
    const query = {
119
      ipywidgets: [
120
        {
121
          string_id: this.syncdoc.get_string_id(),
122
          model_id: null,
123
          type: null,
124
          data: null,
125
        },
126
      ],
127
    };
128
    this.table = await this.create_synctable(query, this.table_options, 0);
129

130
    // TODO: here the project should clear the table.
131

132
    this.set_state("ready");
133

134
    this.table.on("change", (keys) => {
135
      this.emit("change", keys);
136
    });
137
  };
138

139
  keys = (): { model_id: string; type: "value" | "state" | "buffer" }[] => {
140
    // return type is arrow of s
141
    this.assert_state("ready");
142
    const x = this.table.get();
143
    if (x == null) {
144
      return [];
145
    }
146
    const keys: { model_id: string; type: "value" | "state" | "buffer" }[] = [];
147
    x.forEach((val, key) => {
148
      if (val.get("data") != null && key != null) {
149
        const [, model_id, type] = JSON.parse(key);
150
        keys.push({ model_id, type });
151
      }
152
    });
153
    return keys;
154
  };
155

156
  get = (model_id: string, type: string): iMap<string, any> | undefined => {
157
    const key: string = JSON.stringify([
158
      this.syncdoc.get_string_id(),
159
      model_id,
160
      type,
161
    ]);
162
    const record = this.table.get(key);
163
    if (record == null) {
164
      return undefined;
165
    }
166
    return record.get("data");
167
  };
168

169
  // assembles together state we know about the widget with given model_id
170
  // from info in the table, and returns it as a Javascript object.
171
  getSerializedModelState = (
172
    model_id: string,
173
  ): SerializedModelState | undefined => {
174
    this.assert_state("ready");
175
    const state = this.get(model_id, "state");
176
    if (state == null) {
177
      return undefined;
178
    }
179
    const state_js = state.toJS();
180
    let value: any = this.get(model_id, "value");
181
    if (value != null) {
182
      value = value.toJS();
183
      if (value == null) {
184
        throw Error("value must be a map");
185
      }
186
      for (const key in value) {
187
        state_js[key] = value[key];
188
      }
189
    }
190
    return state_js;
191
  };
192

193
  get_model_value = (model_id: string): Value => {
194
    this.assert_state("ready");
195
    let value: any = this.get(model_id, "value");
196
    if (value == null) {
197
      return {};
198
    }
199
    value = value.toJS();
200
    if (value == null) {
201
      return {};
202
    }
203
    return value;
204
  };
205

206
  /*
207
  Setting and getting buffers.
208

209
  - Setting the model buffers only happens on the backend project.
210
    This is done in response to a comm message from the kernel
211
    that has content.data.buffer_paths set.
212

213
  - Getting the model buffers only happens in the frontend browser.
214
    This happens when creating models that support widgets, and often
215
    happens in conjunction with deserialization.
216

217
    Getting a model buffer for a given path can happen
218
    *at any time* after the buffer is created, not just right when
219
    it is created like in JupyterLab!  The reason is because a browser
220
    can connect or get refreshed at any time, and then they need the
221
    buffer to reconstitue the model.  Moreover, a user might only
222
    scroll the widget into view in their (virtualized) notebook at any
223
    point, and it is only then that point the model gets created.
224
    This means that we have to store and garbage collect model
225
    buffers, which is a problem I don't think upstream ipywidgets
226
    has to solve.
227
  */
228
  getModelBuffers = async (
229
    model_id: string,
230
  ): Promise<{
231
    buffer_paths: string[][];
232
    buffers: ArrayBuffer[];
233
  }> => {
234
    let value: iMap<string, string> | undefined = this.get(model_id, "buffers");
235
    if (value == null) {
236
      return { buffer_paths: [], buffers: [] };
237
    }
238
    // value is an array from JSON of paths array to array buffers:
239
    const buffer_paths: string[][] = [];
240
    const buffers: ArrayBuffer[] = [];
241
    if (this.arrayBuffers[model_id] == null) {
242
      this.arrayBuffers[model_id] = {};
243
    }
244
    const f = async (path: string) => {
245
      const hash = value?.get(path);
246
      if (!hash) {
247
        // It is important to look for !hash, since we use hash='' as a sentinel (in this.clearOutputBuffers)
248
        // to indicate that we want to consider a buffer as having been deleted. This is very important
249
        // to do since large outputs are often buffers in output widgets, and clear_output
250
        // then needs to delete those buffers, or output never goes away.
251
        return;
252
      }
253
      const cur = this.arrayBuffers[model_id][path];
254
      if (cur?.hash == hash) {
255
        buffer_paths.push(JSON.parse(path));
256
        buffers.push(cur.buffer);
257
        return;
258
      }
259
      try {
260
        const buffer = await this.clientGetBuffer(model_id, path);
261
        this.arrayBuffers[model_id][path] = { buffer, hash };
262
        buffer_paths.push(JSON.parse(path));
263
        buffers.push(buffer);
264
      } catch (err) {
265
        console.log(`skipping ${model_id}, ${path} due to ${err}`);
266
      }
267
    };
268
    // Run f in parallel on all of the keys of value:
269
    await Promise.all(
270
      value
271
        .keySeq()
272
        .toJS()
273
        .filter((path) => path.startsWith("["))
274
        .map(f),
275
    );
276
    return { buffers, buffer_paths };
277
  };
278

279
  // This is used on the backend when syncing changes from project nodejs *to*
280
  // the jupyter kernel.
281
  getKnownBuffers = (model_id: string) => {
282
    let value: iMap<string, string> | undefined = this.get(model_id, "buffers");
283
    if (value == null) {
284
      return { buffer_paths: [], buffers: [] };
285
    }
286
    // value is an array from JSON of paths array to array buffers:
287
    const buffer_paths: string[][] = [];
288
    const buffers: ArrayBuffer[] = [];
289
    if (this.buffers[model_id] == null) {
290
      this.buffers[model_id] = {};
291
    }
292
    const f = (path: string) => {
293
      const hash = value?.get(path);
294
      if (!hash) {
295
        return;
296
      }
297
      const cur = this.buffers[model_id][path];
298
      if (cur?.hash == hash) {
299
        buffer_paths.push(JSON.parse(path));
300
        buffers.push(new Uint8Array(cur.buffer).buffer);
301
        return;
302
      }
303
    };
304
    value
305
      .keySeq()
306
      .toJS()
307
      .filter((path) => path.startsWith("["))
308
      .map(f);
309
    return { buffers, buffer_paths };
310
  };
311

312
  private clientGetBuffer = async (model_id: string, path: string) => {
313
    // async get of the buffer from backend
314
    if (this.client.ipywidgetsGetBuffer == null) {
315
      throw Error(
316
        "NotImplementedError: frontend client must implement ipywidgetsGetBuffer in order to support binary buffers",
317
      );
318
    }
319
    const b = await this.client.ipywidgetsGetBuffer(
320
      this.syncdoc.project_id,
321
      auxFileToOriginal(this.syncdoc.path),
322
      model_id,
323
      path,
324
    );
325
    return b;
326
  };
327

328
  // Used on the backend by the project http server
329
  getBuffer = (
330
    model_id: string,
331
    buffer_path_or_sha1: string,
332
  ): Buffer | undefined => {
333
    const dbg = this.dbg("getBuffer");
334
    dbg("getBuffer", model_id, buffer_path_or_sha1);
335
    return this.buffers[model_id]?.[buffer_path_or_sha1]?.buffer;
336
  };
337

338
  // returns the sha1 hashes of the buffers
339
  setModelBuffers = (
340
    // model that buffers are associated to:
341
    model_id: string,
342
    // if given, these are buffers with given paths; if not given, we
343
    // store buffer associated to sha1 (which is used for custom messages)
344
    buffer_paths: string[][] | undefined,
345
    // the actual buffers.
346
    buffers: Buffer[],
347
    fire_change_event: boolean = true,
348
  ): string[] => {
349
    const dbg = this.dbg("setModelBuffers");
350
    dbg("buffer_paths = ", buffer_paths);
351

352
    const data: { [path: string]: boolean } = {};
353
    if (this.buffers[model_id] == null) {
354
      this.buffers[model_id] = {};
355
    }
356
    const hashes: string[] = [];
357
    if (buffer_paths != null) {
358
      for (let i = 0; i < buffer_paths.length; i++) {
359
        const key = JSON.stringify(buffer_paths[i]);
360
        // we set to the sha1 of the buffer not just to make getting
361
        // the buffer easy, but to make it easy to KNOW if we
362
        // even need to get the buffer.
363
        const hash = sha1(buffers[i]);
364
        hashes.push(hash);
365
        data[key] = hash;
366
        this.buffers[model_id][key] = { buffer: buffers[i], hash };
367
      }
368
    } else {
369
      for (const buffer of buffers) {
370
        const hash = sha1(buffer);
371
        hashes.push(hash);
372
        this.buffers[model_id][hash] = { buffer, hash };
373
        data[hash] = hash;
374
      }
375
    }
376
    this.set(model_id, "buffers", data, fire_change_event);
377
    return hashes;
378
  };
379

380
  /*
381
  Setting model state and value
382

383
  - model state -- gets set once right when model is defined by kernel
384
  - model "value" -- should be called "update"; gets set with changes to
385
    the model state since it was created.
386
    (I think an inefficiency with this approach is the entire updated
387
    "value" gets broadcast each time anything about it is changed.
388
    Fortunately usually value is small.  However, it would be much
389
    better to broadcast only the information about what changed, though
390
    that is more difficult to implement given our current simple key:value
391
    store sync layer.  This tradeoff may be fully worth it for
392
    our applications, since large data should be in buffers, and those
393
    are efficient.)
394
  */
395

396
  set_model_value = (
397
    model_id: string,
398
    value: Value,
399
    fire_change_event: boolean = true,
400
  ): void => {
401
    this.set(model_id, "value", value, fire_change_event);
402
  };
403

404
  set_model_state = (
405
    model_id: string,
406
    state: any,
407
    fire_change_event: boolean = true,
408
  ): void => {
409
    this.set(model_id, "state", state, fire_change_event);
410
  };
411

412
  // Do any setting of the underlying table through this function.
413
  set = (
414
    model_id: string,
415
    type: "value" | "state" | "buffers" | "message",
416
    data: any,
417
    fire_change_event: boolean = true,
418
    merge?: "none" | "shallow" | "deep",
419
  ): void => {
420
    //const dbg = this.dbg("set");
421
    const string_id = this.syncdoc.get_string_id();
422
    if (typeof data != "object") {
423
      throw Error("TypeError -- data must be a map");
424
    }
425
    let defaultMerge: "none" | "shallow" | "deep";
426
    if (type == "value") {
427
      //defaultMerge = "shallow";
428
      // we manually do the shallow merge only on the data field.
429
      const current = this.get_model_value(model_id);
430
      // this can be HUGE:
431
      // dbg("value: before", { data, current });
432
      if (current != null) {
433
        for (const k in data) {
434
          if (is_object(data[k]) && is_object(current[k])) {
435
            current[k] = { ...current[k], ...data[k] };
436
          } else {
437
            current[k] = data[k];
438
          }
439
        }
440
        data = current;
441
      }
442
      // dbg("value -- after", { merged: data });
443
      defaultMerge = "none";
444
    } else if (type == "buffers") {
445
      // it's critical to not throw away existing buffers when
446
      // new ones come or current ones change.  With shallow merge,
447
      // the existing ones go away, which is very broken, e.g.,
448
      // see this with this example:
449
      /*
450
import bqplot.pyplot as plt
451
import numpy as np
452
x, y = np.random.rand(2, 10)
453
fig = plt.figure(animation_duration=3000)
454
scat = plt.scatter(x=x, y=y)
455
fig
456
---
457
scat.x, scat.y = np.random.rand(2, 50)
458

459
# now close and open it, and it breaks with shallow merge,
460
# since the second cell caused the opacity buffer to be
461
# deleted, which breaks everything.
462
*/
463
      defaultMerge = "deep";
464
    } else if (type == "message") {
465
      defaultMerge = "none";
466
    } else {
467
      defaultMerge = "deep";
468
    }
469
    if (merge == null) {
470
      merge = defaultMerge;
471
    }
472
    this.table.set(
473
      { string_id, type, model_id, data },
474
      merge,
475
      fire_change_event,
476
    );
477
  };
478

479
  save = async (): Promise<void> => {
480
    this.gc();
481
    await this.table.save();
482
  };
483

484
  close = async (): Promise<void> => {
485
    if (this.table != null) {
486
      await this.table.close();
487
    }
488
    close(this);
489
    this.set_state("closed");
490
  };
491

492
  private dbg = (_f): Function => {
493
    if (this.client.is_project()) {
494
      return this.client.dbg(`IpywidgetsState.${_f}`);
495
    } else {
496
      return (..._) => {};
497
    }
498
  };
499

500
  clear = async (): Promise<void> => {
501
    // This is used when we restart the kernel -- we reset
502
    // things so no information about any models is known
503
    // and delete all Buffers.
504
    this.assert_state("ready");
505
    const dbg = this.dbg("clear");
506
    dbg();
507

508
    this.buffers = {};
509
    // There's no implemented delete for tables yet, so instead we set the data
510
    // for everything to null.  All other code related to widgets needs to handle
511
    // such data appropriately and ignore it.  (An advantage of this over trying to
512
    // implement a genuine delete is that delete is tricky when clients reconnect
513
    // and sync...). This table is in memory only anyways, so the table will get properly
514
    // fully flushed from existence at some point.
515
    const keys = this.table?.get()?.keySeq()?.toJS();
516
    if (keys == null) return; // nothing to do.
517
    for (const key of keys) {
518
      const [string_id, model_id, type] = JSON.parse(key);
519
      this.table.set({ string_id, type, model_id, data: null }, "none", false);
520
    }
521
    await this.table.save();
522
  };
523

524
  values = () => {
525
    const x = this.table.get();
526
    if (x == null) {
527
      return [];
528
    }
529
    return Object.values(x.toJS()).filter((obj) => obj.data);
530
  };
531

532
  // Clean up all data in the table about models that are not
533
  // referenced (directly or indirectly) in any cell in the notebook.
534
  // There is also a comm:close event/message somewhere, which
535
  // could also be useful....?
536
  deleteUnused = async (): Promise<void> => {
537
    this.assert_state("ready");
538
    const dbg = this.dbg("deleteUnused");
539
    dbg();
540
    // See comment in the "clear" function above about no delete for tables,
541
    // which is why we just set the data to null.
542
    const activeIds = this.getActiveModelIds();
543
    this.table.get()?.forEach((val, key) => {
544
      if (key == null || val?.get("data") == null) {
545
        // already deleted
546
        return;
547
      }
548
      const [string_id, model_id, type] = JSON.parse(key);
549
      if (!activeIds.has(model_id)) {
550
        // Delete this model from the table (or as close to delete as we have).
551
        // This removes the last message, state, buffer info, and value,
552
        // depending on type.
553
        this.table.set(
554
          { string_id, type, model_id, data: null },
555
          "none",
556
          false,
557
        );
558

559
        // Also delete buffers for this model, which are stored in memory, and
560
        // won't be requested again.
561
        delete this.buffers[model_id];
562
      }
563
    });
564
    await this.table.save();
565
  };
566

567
  // For each model in init, we add in all the ids of models
568
  // that it explicitly references, e.g., by IPY_MODEL_[model_id] fields
569
  // and by output messages and other things we learn about (e.g., k3d
570
  // has its own custom references).
571
  getReferencedModelIds = (init: string | Set<string>): Set<string> => {
572
    const modelIds =
573
      typeof init == "string" ? new Set([init]) : new Set<string>(init);
574
    let before = 0;
575
    let after = modelIds.size;
576
    while (before < after) {
577
      before = modelIds.size;
578
      for (const model_id of modelIds) {
579
        for (const type of ["state", "value"]) {
580
          const data = this.get(model_id, type);
581
          if (data == null) continue;
582
          for (const id of getModelIds(data)) {
583
            modelIds.add(id);
584
          }
585
        }
586
      }
587
      after = modelIds.size;
588
    }
589
    // Also any custom ways of doing referencing -- e.g., k3d does this.
590
    this.includeThirdPartyReferences(modelIds);
591

592
    // Also anything that references any modelIds
593
    this.includeReferenceTo(modelIds);
594

595
    return modelIds;
596
  };
597

598
  // We find the ids of all models that are explicitly referenced
599
  // in the current version of the Jupyter notebook by iterating through
600
  // the output of all cells, then expanding the result to everything
601
  // that these models reference.  This is used as a foundation for
602
  // garbage collection.
603
  private getActiveModelIds = (): Set<string> => {
604
    const modelIds: Set<string> = new Set();
605
    this.syncdoc.get({ type: "cell" }).forEach((cell) => {
606
      const output = cell.get("output");
607
      if (output != null) {
608
        output.forEach((mesg) => {
609
          const model_id = mesg.getIn([
610
            "data",
611
            "application/vnd.jupyter.widget-view+json",
612
            "model_id",
613
          ]);
614
          if (model_id != null) {
615
            // same id could of course appear in multiple cells
616
            // if there are multiple view of the same model.
617
            modelIds.add(model_id);
618
          }
619
        });
620
      }
621
    });
622
    return this.getReferencedModelIds(modelIds);
623
  };
624

625
  private includeReferenceTo = (modelIds: Set<string>) => {
626
    // This example is extra tricky and one version of our GC broke it:
627
    // from ipywidgets import VBox, jsdlink, IntSlider, Button; s1 = IntSlider(max=200, value=100); s2 = IntSlider(value=40); jsdlink((s1, 'value'), (s2, 'max')); VBox([s1, s2])
628
    // What happens here is that this jsdlink model ends up referencing live widgets,
629
    // but is not referenced by any cell, so it would get garbage collected.
630

631
    let before = -1;
632
    let after = modelIds.size;
633
    while (before < after) {
634
      before = modelIds.size;
635
      this.table.get()?.forEach((val) => {
636
        const data = val?.get("data");
637
        if (data != null) {
638
          for (const model_id of getModelIds(data)) {
639
            if (modelIds.has(model_id)) {
640
              modelIds.add(val.get("model_id"));
641
            }
642
          }
643
        }
644
      });
645
      after = modelIds.size;
646
    }
647
  };
648

649
  private includeThirdPartyReferences = (modelIds: Set<string>) => {
650
    /*
651
    Motivation (RANT):
652
    It seems to me that third party widgets can just invent their own
653
    ways of referencing each other, and there's no way to know what they are
654
    doing.  The only possible way to do garbage collection is by reading
655
    and understanding their code or reverse engineering their data.
656
    It's not unlikely that any nontrivial third
657
    party widget has invented it's own custom way to do object references,
658
    and for every single one we may need to write custom code for garbage
659
    collection, which can randomly break if they change.
660
    <sarcasm>Yeah.</sarcasm>
661
    /*
662

663
    /* k3d:
664
    We handle k3d here, which creates models with
665
         {_model_module:'k3d', _model_name:'ObjectModel', id:number}
666
    where the id is in the object_ids attribute of some model found above:
667
       {_model_module:'k3d', object_ids:[..., id, ...]}
668
    But note that this format is something that was entirely just invented
669
    arbitrarily by the k3d dev.
670
    */
671
    // First get all object_ids of all active models:
672
    // We're not explicitly restricting to k3d here, since maybe other widgets use
673
    // this same approach, and the worst case scenario is just insufficient garbage collection.
674
    const object_ids = new Set<number>([]);
675
    for (const model_id of modelIds) {
676
      for (const type of ["state", "value"]) {
677
        this.get(model_id, type)
678
          ?.get("object_ids")
679
          ?.forEach((id) => {
680
            object_ids.add(id);
681
          });
682
      }
683
    }
684
    if (object_ids.size == 0) {
685
      // nothing to do -- no such object_ids in any current models.
686
      return;
687
    }
688
    // let's find the models with these id's as id attribute and include them.
689
    this.table.get()?.forEach((val) => {
690
      if (object_ids.has(val?.getIn(["data", "id"]))) {
691
        const model_id = val.get("model_id");
692
        modelIds.add(model_id);
693
      }
694
    });
695
  };
696

697
  // The finite state machine state, e.g., 'init' --> 'ready' --> 'close'
698
  private set_state = (state: State): void => {
699
    this.state = state;
700
    this.emit(state);
701
  };
702

703
  get_state = (): State => {
704
    return this.state;
705
  };
706

707
  private assert_state = (state: string): void => {
708
    if (this.state != state) {
709
      throw Error(`state must be "${state}" but it is "${this.state}"`);
710
    }
711
  };
712

713
  /*
714
  process_comm_message_from_kernel gets called whenever the
715
  kernel emits a comm message related to widgets.  This updates
716
  the state of the table, which results in frontends creating widgets
717
  or updating state of widgets.
718
  */
719
  process_comm_message_from_kernel = async (
720
    msg: CommMessage,
721
  ): Promise<void> => {
722
    const dbg = this.dbg("process_comm_message_from_kernel");
723
    // WARNING: serializing any msg could cause huge server load, e.g., it could contain
724
    // a 20MB buffer in it.
725
    //dbg(JSON.stringify(msg));  // EXTREME DANGER!
726
    //console.log("process_comm_message_from_kernel", msg);
727
    dbg(JSON.stringify(msg.header));
728
    this.assert_state("ready");
729

730
    const { content } = msg;
731

732
    if (content == null) {
733
      dbg("content is null -- ignoring message");
734
      return;
735
    }
736

737
    if (content.data.method == "echo_update") {
738
      // just ignore echo_update -- it's a new ipywidgets 8 mechanism
739
      // for some level of RTC sync between clients -- we don't need that
740
      // since we have our own, obviously. Setting the env var
741
      // JUPYTER_WIDGETS_ECHO to 0 will disable these messages to slightly
742
      // reduce traffic.
743
      // NOTE: this check was lower which wrecked the buffers,
744
      // which was a bug for a long time. :-(
745
      return;
746
    }
747

748
    let { comm_id } = content;
749
    if (comm_id == null) {
750
      if (msg.header != null) {
751
        comm_id = msg.header.msg_id;
752
      }
753
      if (comm_id == null) {
754
        dbg("comm_id is null -- ignoring message");
755
        return;
756
      }
757
    }
758
    const model_id: string = comm_id;
759
    dbg({ model_id, comm_id });
760

761
    const { data } = content;
762
    if (data == null) {
763
      dbg("content.data is null -- ignoring message");
764
      return;
765
    }
766

767
    const { state } = data;
768
    if (state != null) {
769
      delete_null_fields(state);
770
    }
771

772
    // It is critical to send any buffers data before
773
    // the other data; otherwise, deserialization on
774
    // the client side can't work, since it is missing
775
    // the data it needs.
776
    // This happens with method "update".  With method="custom",
777
    // there is just an array of buffers and no buffer_paths at all.
778
    if (content.data.buffer_paths?.length > 0) {
779
      // Deal with binary buffers:
780
      dbg("setting binary buffers");
781
      this.setModelBuffers(
782
        model_id,
783
        content.data.buffer_paths,
784
        msg.buffers,
785
        false,
786
      );
787
    }
788

789
    switch (content.data.method) {
790
      case "custom":
791
        const message = content.data.content;
792
        const { buffers } = msg;
793
        dbg("custom message", {
794
          message,
795
          buffers: `${buffers?.length ?? "no"} buffers`,
796
        });
797
        let buffer_hashes: string[];
798
        if (
799
          buffers != null &&
800
          buffers.length > 0 &&
801
          content.data.buffer_paths == null
802
        ) {
803
          // TODO
804
          dbg("custom message  -- there are BUFFERS -- saving them");
805
          buffer_hashes = this.setModelBuffers(
806
            model_id,
807
            undefined,
808
            buffers,
809
            false,
810
          );
811
        } else {
812
          buffer_hashes = [];
813
        }
814
        // We now send the message.
815
        this.sendCustomMessage(model_id, message, buffer_hashes, false);
816
        break;
817

818
      case "echo_update":
819
        return;
820

821
      case "update":
822
        if (state == null) {
823
          return;
824
        }
825
        dbg("method -- update");
826
        if (this.clear_output[model_id] && state.outputs != null) {
827
          // we are supposed to clear the output before inserting
828
          // the next output.
829
          dbg("clearing outputs");
830
          if (state.outputs.length > 0) {
831
            state.outputs = [state.outputs[state.outputs.length - 1]];
832
          } else {
833
            state.outputs = [];
834
          }
835
          delete this.clear_output[model_id];
836
        }
837

838
        const last_changed =
839
          (this.get(model_id, "value")?.get("last_changed") ?? 0) + 1;
840
        this.set_model_value(model_id, { ...state, last_changed }, false);
841

842
        if (state.msg_id != null) {
843
          const { msg_id } = state;
844
          if (typeof msg_id === "string" && msg_id.length > 0) {
845
            dbg("enabling capture output", msg_id, model_id);
846
            if (this.capture_output[msg_id] == null) {
847
              this.capture_output[msg_id] = [model_id];
848
            } else {
849
              // pushing onto stack
850
              this.capture_output[msg_id].push(model_id);
851
            }
852
          } else {
853
            const parent_msg_id = msg.parent_header.msg_id;
854
            dbg("disabling capture output", parent_msg_id, model_id);
855
            if (this.capture_output[parent_msg_id] != null) {
856
              const v: string[] = [];
857
              const w: string[] = this.capture_output[parent_msg_id];
858
              for (const m of w) {
859
                if (m != model_id) {
860
                  v.push(m);
861
                }
862
              }
863
              if (v.length == 0) {
864
                delete this.capture_output[parent_msg_id];
865
              } else {
866
                this.capture_output[parent_msg_id] = v;
867
              }
868
            }
869
          }
870
          delete state.msg_id;
871
        }
872
        break;
873
      case undefined:
874
        if (state == null) return;
875
        dbg("method -- undefined (=set_model_state)", { model_id, state });
876
        this.set_model_state(model_id, state, false);
877
        break;
878
      default:
879
        // TODO: Implement other methods, e.g., 'display' -- see
880
        // https://github.com/jupyter-widgets/ipywidgets/blob/master/packages/schema/messages.md
881
        dbg(`not implemented method '${content.data.method}' -- ignoring`);
882
    }
883

884
    await this.save();
885
  };
886

887
  /*
888
  process_comm_message_from_browser gets called whenever a
889
  browser client emits a comm message related to widgets.
890
  This updates the state of the table, which results in
891
  other frontends updating their widget state, *AND* the backend
892
  kernel changing the value of variables (and possibly
893
  updating other widgets).
894
  */
895
  process_comm_message_from_browser = async (
896
    msg: CommMessage,
897
  ): Promise<void> => {
898
    const dbg = this.dbg("process_comm_message_from_browser");
899
    dbg(msg);
900
    this.assert_state("ready");
901
    // TODO: not implemented!
902
  };
903

904
  // The mesg here is exactly what came over the IOPUB channel
905
  // from the kernel.
906

907
  // TODO: deal with buffers
908
  capture_output_message = (mesg: any): boolean => {
909
    const msg_id = mesg.parent_header.msg_id;
910
    if (this.capture_output[msg_id] == null) {
911
      return false;
912
    }
913
    const dbg = this.dbg("capture_output_message");
914
    dbg(JSON.stringify(mesg));
915
    const model_id =
916
      this.capture_output[msg_id][this.capture_output[msg_id].length - 1];
917
    if (model_id == null) return false; // should not happen.
918

919
    if (mesg.header.msg_type == "clear_output") {
920
      if (mesg.content?.wait) {
921
        this.clear_output[model_id] = true;
922
      } else {
923
        delete this.clear_output[model_id];
924
        this.clearOutputBuffers(model_id);
925
        this.set_model_value(model_id, { outputs: null });
926
      }
927
      return true;
928
    }
929

930
    if (mesg.content == null || len(mesg.content) == 0) {
931
      // no actual content.
932
      return false;
933
    }
934

935
    let outputs: any;
936
    if (this.clear_output[model_id]) {
937
      delete this.clear_output[model_id];
938
      this.clearOutputBuffers(model_id);
939
      outputs = [];
940
    } else {
941
      outputs = this.get_model_value(model_id).outputs;
942
      if (outputs == null) {
943
        outputs = [];
944
      }
945
    }
946
    outputs.push(mesg.content);
947
    this.set_model_value(model_id, { outputs });
948
    return true;
949
  };
950

951
  private clearOutputBuffers = (model_id: string) => {
952
    // TODO: need to clear all output buffers.
953
    /* Example where if you do not properly clear buffers, then broken output re-appears:
954

955
import ipywidgets as widgets
956
from IPython.display import YouTubeVideo
957
out = widgets.Output(layout={'border': '1px solid black'})
958
out.append_stdout('Output appended with append_stdout')
959
out.append_display_data(YouTubeVideo('eWzY2nGfkXk'))
960
out
961

962
---
963

964
out.clear_output()
965

966
---
967

968
with out:
969
   print('hi')
970
    */
971
    // TODO!!!!
972

973
    const y: any = {};
974
    let n = 0;
975
    for (const jsonPath of this.get(model_id, "buffers")?.keySeq() ?? []) {
976
      const path = JSON.parse(jsonPath);
977
      if (path[0] == "outputs") {
978
        y[jsonPath] = "";
979
        n += 1;
980
      }
981
    }
982
    if (n > 0) {
983
      this.set(model_id, "buffers", y, true, "shallow");
984
    }
985
  };
986

987
  private sendCustomMessage = async (
988
    model_id: string,
989
    message: object,
990
    buffer_hashes: string[],
991
    fire_change_event: boolean = true,
992
  ): Promise<void> => {
993
    /*
994
    Send a custom message.
995

996
    It's not at all clear what this should even mean in the context of
997
    realtime collaboration, and there will likely be clients where
998
    this is bad.  But for now, we just make the last message sent
999
    available via the table, and each successive message overwrites the previous
1000
    one.  Any clients that are connected while we do this can react,
1001
    and any that aren't just don't get the message (which is presumably fine).
1002

1003
    Some widgets like ipympl use this to initialize state, so when a new
1004
    client connects, it requests a message describing the plot, and everybody
1005
    receives it.
1006
    */
1007

1008
    this.set(
1009
      model_id,
1010
      "message",
1011
      { message, buffer_hashes, time: Date.now() },
1012
      fire_change_event,
1013
    );
1014
  };
1015

1016
  // Return the most recent message for the given model.
1017
  getMessage = async (
1018
    model_id: string,
1019
  ): Promise<{ message: object; buffers: ArrayBuffer[] } | undefined> => {
1020
    const x = this.get(model_id, "message")?.toJS();
1021
    if (x == null) {
1022
      return undefined;
1023
    }
1024
    if (Date.now() - (x.time ?? 0) >= MAX_MESSAGE_TIME_MS) {
1025
      return undefined;
1026
    }
1027
    const { message, buffer_hashes } = x;
1028
    let buffers: ArrayBuffer[] = [];
1029
    for (const hash of buffer_hashes) {
1030
      buffers.push(await this.clientGetBuffer(model_id, hash));
1031
    }
1032
    return { message, buffers };
1033
  };
1034
}
1035

1036
// Get model id's that appear either as serialized references
1037
// of the form IPY_MODEL_....
1038
// or in output messages.
1039
function getModelIds(x): Set<string> {
1040
  const ids: Set<string> = new Set();
1041
  x?.forEach((val, key) => {
1042
    if (key == "application/vnd.jupyter.widget-view+json") {
1043
      const model_id = val.get("model_id");
1044
      if (model_id) {
1045
        ids.add(model_id);
1046
      }
1047
    } else if (typeof val == "string") {
1048
      if (val.startsWith("IPY_MODEL_")) {
1049
        ids.add(val.slice("IPY_MODEL_".length));
1050
      }
1051
    } else if (val.forEach != null) {
1052
      for (const z of getModelIds(val)) {
1053
        ids.add(z);
1054
      }
1055
    }
1056
  });
1057
  return ids;
1058
}
1059

1060