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
Start the Sage server and also get a new socket connection to it.
8
*/
9

10
import { reuseInFlight } from "@cocalc/util/reuse-in-flight";
11
import { getLogger } from "@cocalc/backend/logger";
12
import processKill from "@cocalc/backend/misc/process-kill";
13
import { abspath } from "@cocalc/backend/misc_node";
14
import type {
15
  Type as TCPMesgType,
16
  Message as TCPMessage,
17
} from "@cocalc/backend/tcp/enable-messaging-protocol";
18
import { CoCalcSocket } from "@cocalc/backend/tcp/enable-messaging-protocol";
19
import * as message from "@cocalc/util/message";
20
import {
21
  path_split,
22
  to_json,
23
  trunc,
24
  trunc_middle,
25
  uuid,
26
} from "@cocalc/util/misc";
27
import { CB } from "@cocalc/util/types/callback";
28
import { ISageSession, SageCallOpts } from "@cocalc/util/types/sage";
29
import { Client } from "./client";
30
import { getSageSocket } from "./sage_socket";
31

32
const logger = getLogger("sage-session");
33

34
//##############################################
35
// Direct Sage socket session -- used internally in local hub, e.g., to assist CodeMirror editors...
36
//##############################################
37

38
// we have to make sure to only export the type to avoid error TS4094
39
export type SageSessionType = InstanceType<typeof SageSession>;
40

41
interface SageSessionOpts {
42
  client: Client;
43
  path: string; // the path to the *worksheet* file
44
}
45

46
const cache: { [path: string]: SageSessionType } = {};
47

48
export function sage_session(opts: Readonly<SageSessionOpts>): SageSessionType {
49
  const { path } = opts;
50
  // compute and cache if not cached; otherwise, get from cache:
51
  return (cache[path] = cache[path] ?? new SageSession(opts));
52
}
53

54
/*
55
Sage Session object
56

57
Until you actually try to call it no socket need
58
*/
59
class SageSession implements ISageSession {
60
  private _path: string;
61
  private _client: Client;
62
  private _output_cb: {
63
    [key: string]: CB<{ done: boolean; error: string }, any>;
64
  } = {};
65
  private _socket: CoCalcSocket | undefined;
66

67
  constructor(opts: Readonly<SageSessionOpts>) {
68
    this.dbg = this.dbg.bind(this);
69
    this.dbg("constructor")();
70
    this._path = opts.path;
71
    this._client = opts.client;
72
    this._output_cb = {};
73
  }
74

75
  private dbg = (f: string) => {
76
    return (m?: string) =>
77
      logger.debug(`SageSession(path='${this._path}').${f}: ${m}`);
78
  };
79

80
  close = (): void => {
81
    if (this._socket != null) {
82
      const pid = this._socket.pid;
83
      if (pid != null) {
84
        processKill(pid, 9);
85
      }
86
      this._socket.end();
87
      delete this._socket;
88
    }
89
    for (let id in this._output_cb) {
90
      const cb = this._output_cb[id];
91
      cb({ done: true, error: "killed" });
92
    }
93
    this._output_cb = {};
94
    delete cache[this._path];
95
  };
96

97
  // return true if there is a socket connection to a sage server process
98
  is_running = (): boolean => {
99
    return this._socket != null;
100
  };
101

102
  // NOTE: There can be many simultaneous init_socket calls at the same time,
103
  // if e.g., the socket doesn't exist and there are a bunch of calls to @call
104
  // at the same time.
105
  // See https://github.com/sagemathinc/cocalc/issues/3506
106
  // wrapped in reuseInFlight !
107
  init_socket = reuseInFlight(async (): Promise<void> => {
108
    const dbg = this.dbg("init_socket()");
109
    dbg();
110
    try {
111
      const socket: CoCalcSocket = await getSageSocket();
112

113
      dbg("successfully opened a sage session");
114
      this._socket = socket;
115

116
      socket.on("end", () => {
117
        delete this._socket;
118
        return dbg("codemirror session terminated");
119
      });
120

121
      // CRITICAL: we must define this handler before @_init_path below,
122
      // or @_init_path can't possibly work... since it would wait for
123
      // this handler to get the response message!
124
      socket.on("mesg", (type: TCPMesgType, mesg: TCPMessage) => {
125
        dbg(`sage session: received message ${type}`);
126
        switch (type) {
127
          case "json":
128
            this._handle_mesg_json(mesg);
129
            break;
130
          case "blob":
131
            this._handle_mesg_blob(mesg);
132
            break;
133
        }
134
      });
135

136
      await this._init_path();
137
    } catch (err) {
138
      if (err) {
139
        dbg(`fail -- ${err}.`);
140
        throw err;
141
      }
142
    }
143
  });
144

145
  private _init_path = async (): Promise<void> => {
146
    const dbg = this.dbg("_init_path()");
147
    dbg();
148
    return new Promise<void>((resolve, reject) => {
149
      this.call({
150
        input: {
151
          event: "execute_code",
152
          code: "os.chdir(salvus.data['path']);__file__=salvus.data['file']",
153
          data: {
154
            path: abspath(path_split(this._path).head),
155
            file: abspath(this._path),
156
          },
157
          preparse: false,
158
        },
159
        cb: (resp) => {
160
          let err: string | undefined = undefined;
161
          if (resp.stderr) {
162
            err = resp.stderr;
163
            dbg(`error '${err}'`);
164
          }
165
          if (resp.done) {
166
            if (err) {
167
              reject(err);
168
            } else {
169
              resolve();
170
            }
171
          }
172
        },
173
      });
174
    });
175
  };
176

177
  public call = async ({
178
    input,
179
    cb,
180
  }: Readonly<SageCallOpts>): Promise<void> => {
181
    const dbg = this.dbg("call");
182
    dbg(`input='${trunc(to_json(input), 300)}'`);
183
    switch (input.event) {
184
      case "ping":
185
        cb({ pong: true });
186
        return;
187

188
      case "status":
189
        cb({ running: this.is_running() });
190
        return;
191

192
      case "signal":
193
        if (this._socket != null) {
194
          dbg(`sending signal ${input.signal} to process ${this._socket.pid}`);
195
          const pid = this._socket.pid;
196
          if (pid != null) processKill(pid, input.signal);
197
        }
198
        cb({});
199
        return;
200

201
      case "restart":
202
        dbg("restarting sage session");
203
        if (this._socket != null) {
204
          this.close();
205
        }
206
        try {
207
          await this.init_socket();
208
          cb({});
209
        } catch (err) {
210
          cb({ error: err });
211
        }
212
        return;
213

214
      case "raw_input":
215
        dbg("sending sage_raw_input event");
216
        this._socket?.write_mesg("json", {
217
          event: "sage_raw_input",
218
          value: input.value,
219
        });
220
        return;
221

222
      default:
223
        // send message over socket and get responses
224
        try {
225
          if (this._socket == null) {
226
            await this.init_socket();
227
          }
228

229
          if (input.id == null) {
230
            input.id = uuid();
231
            dbg(`generated new random uuid for input: '${input.id}' `);
232
          }
233

234
          if (this._socket == null) {
235
            throw new Error("no socket");
236
          }
237

238
          this._socket.write_mesg("json", input);
239

240
          this._output_cb[input.id] = cb; // this is when opts.cb will get called...
241
        } catch (err) {
242
          cb({ done: true, error: err });
243
        }
244
    }
245
  };
246
  private _handle_mesg_blob = (mesg: TCPMessage) => {
247
    const { uuid } = mesg;
248
    let { blob } = mesg;
249
    const dbg = this.dbg(`_handle_mesg_blob(uuid='${uuid}')`);
250
    dbg();
251

252
    if (blob == null) {
253
      dbg("no blob -- dropping message");
254
      return;
255
    }
256

257
    // This should never happen, typing enforces this to be a Buffer
258
    if (typeof blob === "string") {
259
      dbg("blob is string -- converting to buffer");
260
      blob = Buffer.from(blob, "utf8");
261
    }
262

263
    this._client.save_blob({
264
      blob,
265
      uuid,
266
      cb: (err, resp) => {
267
        if (err) {
268
          resp = message.save_blob({
269
            error: err,
270
            sha1: uuid, // dumb - that sha1 should be called uuid...
271
          });
272
        }
273
        this._socket?.write_mesg("json", resp);
274
      },
275
    });
276
  };
277

278
  private _handle_mesg_json = (mesg: TCPMessage) => {
279
    const dbg = this.dbg("_handle_mesg_json");
280
    dbg(`mesg='${trunc_middle(to_json(mesg), 400)}'`);
281
    if (mesg == null) return; // should not happen
282
    const { id } = mesg;
283
    if (id == null) return; // should not happen
284
    const cb = this._output_cb[id];
285
    if (cb != null) {
286
      // Must do this check first since it uses done:false.
287
      if (mesg.done || mesg.done == null) {
288
        delete this._output_cb[id];
289
        mesg.done = true;
290
      }
291
      if (mesg.done != null && !mesg.done) {
292
        // waste of space to include done part of mesg if just false for everything else...
293
        delete mesg.done;
294
      }
295
      cb(mesg);
296
    }
297
  };
298
}
299

300