1
/*
2

3
Used mainly from a browser client frontend to manage what compute server
4
is used to edit a given file.
5

6
Access this in the browser for the project you have open:
7

8
> m = await cc.client.conat_client.computeServerManager({project_id:cc.current().project_id})
9

10
*/
11

12
import { dkv, type DKV } from "@cocalc/conat/sync/dkv";
13
import { EventEmitter } from "events";
14
import { once, until } from "@cocalc/util/async-utils";
15

16
type State = "init" | "connected" | "closed";
17

18
export interface Info {
19
  // id = compute server where this path should be opened
20
  id: number;
21
}
22

23
export interface Options {
24
  project_id: string;
25
  noAutosave?: boolean;
26
  noCache?: boolean;
27
}
28

29
export function computeServerManager(options: Options) {
30
  return new ComputeServerManager(options);
31
}
32

33
export class ComputeServerManager extends EventEmitter {
34
  private dkv?: DKV<Info>;
35
  private options: Options;
36
  public state: State = "init";
37

38
  constructor(options: Options) {
39
    super();
40
    this.options = options;
41
    // It's reasonable to have many clients, e.g., one for each open file
42
    this.setMaxListeners(100);
43
    this.init();
44
  }
45

46
  waitUntilReady = async () => {
47
    if (this.state == "closed") {
48
      throw Error("closed");
49
    } else if (this.state == "connected") {
50
      return;
51
    }
52
    await once(this, "connected");
53
  };
54

55
  save = async () => {
56
    await this.dkv?.save();
57
  };
58

59
  private initialized = false;
60
  init = async () => {
61
    if (this.initialized) {
62
      throw Error("init can only be called once");
63
    }
64
    this.initialized = true;
65
    await until(
66
      async () => {
67
        if (this.state != "init") {
68
          return true;
69
        }
70
        const d = await dkv<Info>({
71
          name: "compute-server-manager",
72
          ...this.options,
73
        });
74
        if (this.state == ("closed" as any)) {
75
          d.close();
76
          return true;
77
        }
78
        this.dkv = d;
79
        d.on("change", this.handleChange);
80
        this.setState("connected");
81
        return true;
82
      },
83
      {
84
        start: 3000,
85
        decay: 1.3,
86
        max: 15000,
87
        log: (...args) =>
88
          console.log(
89
            "WARNING: issue creating compute server manager",
90
            ...args,
91
          ),
92
      },
93
    );
94
  };
95

96
  private handleChange = ({ key: path, value, prev }) => {
97
    this.emit("change", {
98
      path,
99
      id: value?.id,
100
      prev_id: prev?.id,
101
    });
102
  };
103

104
  close = () => {
105
    // console.log("close compute server manager", this.options);
106
    if (this.dkv != null) {
107
      this.dkv.removeListener("change", this.handleChange);
108
      this.dkv.close();
109
      delete this.dkv;
110
    }
111
    this.setState("closed");
112
    this.removeAllListeners();
113
  };
114

115
  private setState = (state: State) => {
116
    this.state = state;
117
    this.emit(state);
118
  };
119

120
  private getDkv = () => {
121
    if (this.dkv == null) {
122
      const message = `compute server manager not initialized -- in state '${this.state}'`;
123
      console.warn(message);
124
      throw Error(message);
125
    }
126
    return this.dkv;
127
  };
128

129
  set = (path, id) => {
130
    const kv = this.getDkv();
131
    if (!id) {
132
      kv.delete(path);
133
      return;
134
    }
135
    kv.set(path, { id });
136
  };
137

138
  delete = (path) => {
139
    this.getDkv().delete(path);
140
  };
141

142
  get = (path) => this.getDkv().get(path)?.id;
143

144
  getAll = () => {
145
    return this.getDkv().getAll();
146
  };
147

148
  // Async API that doesn't assume manager has been initialized, with
149
  // very long names.  Used in the frontend.
150

151
  // Call this if you want the compute server with given id to
152
  // connect and handle being the server for the given path.
153
  connectComputeServerToPath = async ({
154
    path,
155
    id,
156
  }: {
157
    path: string;
158
    id: number;
159
  }) => {
160
    try {
161
      await this.waitUntilReady();
162
    } catch {
163
      return;
164
    }
165
    this.set(path, id);
166
  };
167

168
  // Call this if you want no compute servers to provide the backend server
169
  // for given path.
170
  disconnectComputeServer = async ({ path }: { path: string }) => {
171
    try {
172
      await this.waitUntilReady();
173
    } catch {
174
      return;
175
    }
176
    this.delete(path);
177
  };
178

179
  // Returns the explicitly set server id for the given
180
  // path, if one is set. Otherwise, return undefined
181
  // if nothing is explicitly set for this path (i.e., usually means home base).
182
  getServerIdForPath = async (path: string): Promise<number | undefined> => {
183
    try {
184
      await this.waitUntilReady();
185
    } catch {
186
      return;
187
    }
188
    return this.get(path);
189
  };
190

191
  // Get the server ids (as a map) for every file and every directory contained in path.
192
  // NOTE/TODO: this just does a linear search through all paths with a server id; nothing clever.
193
  getServerIdForSubtree = async (
194
    path: string,
195
  ): Promise<{ [path: string]: number }> => {
196
    await this.waitUntilReady();
197
    if (this.state == "closed") {
198
      throw Error("closed");
199
    }
200
    const kv = this.getDkv();
201
    const v: { [path: string]: number } = {};
202
    const slash = path.endsWith("/") ? path : path + "/";
203
    const x = kv.getAll();
204
    for (const p in x) {
205
      if (p == path || p.startsWith(slash)) {
206
        v[p] = x[p].id;
207
      }
208
    }
209
    return v;
210
  };
211
}
212

213