251 lines
6.2 KiB
TypeScript
251 lines
6.2 KiB
TypeScript
// Copyright 2021 Signal Messenger, LLC
|
|
// SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
import { join } from 'path';
|
|
import { Worker } from 'worker_threads';
|
|
import { format } from 'util';
|
|
import { app } from 'electron';
|
|
|
|
import { strictAssert } from '../util/assert';
|
|
import { explodePromise } from '../util/explodePromise';
|
|
import type { LoggerType } from '../types/Logging';
|
|
import { SqliteErrorKind } from './errors';
|
|
import type DB from './Server';
|
|
|
|
const MIN_TRACE_DURATION = 40;
|
|
|
|
export type InitializeOptions = Readonly<{
|
|
appVersion: string;
|
|
configDir: string;
|
|
key: string;
|
|
logger: LoggerType;
|
|
}>;
|
|
|
|
export type WorkerRequest = Readonly<
|
|
| {
|
|
type: 'init';
|
|
options: Omit<InitializeOptions, 'logger'>;
|
|
}
|
|
| {
|
|
type: 'close';
|
|
}
|
|
| {
|
|
type: 'removeDB';
|
|
}
|
|
| {
|
|
type: 'sqlCall';
|
|
method: keyof typeof DB;
|
|
args: ReadonlyArray<unknown>;
|
|
}
|
|
>;
|
|
|
|
export type WrappedWorkerRequest = Readonly<{
|
|
seq: number;
|
|
request: WorkerRequest;
|
|
}>;
|
|
|
|
export type WrappedWorkerLogEntry = Readonly<{
|
|
type: 'log';
|
|
level: 'fatal' | 'error' | 'warn' | 'info' | 'debug' | 'trace';
|
|
args: ReadonlyArray<unknown>;
|
|
}>;
|
|
|
|
export type WrappedWorkerResponse =
|
|
| Readonly<{
|
|
type: 'response';
|
|
seq: number;
|
|
error: string | undefined;
|
|
errorKind: SqliteErrorKind | undefined;
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
response: any;
|
|
}>
|
|
| WrappedWorkerLogEntry;
|
|
|
|
type PromisePair<T> = {
|
|
resolve: (response: T) => void;
|
|
reject: (error: Error) => void;
|
|
};
|
|
|
|
type KnownErrorResolverType = Readonly<{
|
|
kind: SqliteErrorKind;
|
|
resolve: (err: Error) => void;
|
|
}>;
|
|
|
|
export class MainSQL {
|
|
private readonly worker: Worker;
|
|
|
|
private isReady = false;
|
|
|
|
private onReady: Promise<void> | undefined;
|
|
|
|
private readonly onExit: Promise<void>;
|
|
|
|
// Promise resolve callbacks for corruption and readonly errors.
|
|
private errorResolvers = new Array<KnownErrorResolverType>();
|
|
|
|
private seq = 0;
|
|
|
|
private logger?: LoggerType;
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
private onResponse = new Map<number, PromisePair<any>>();
|
|
|
|
private shouldTimeQueries = false;
|
|
|
|
constructor() {
|
|
const scriptDir = join(app.getAppPath(), 'ts', 'sql', 'mainWorker.js');
|
|
this.worker = new Worker(scriptDir);
|
|
|
|
this.worker.on('message', (wrappedResponse: WrappedWorkerResponse) => {
|
|
if (wrappedResponse.type === 'log') {
|
|
const { level, args } = wrappedResponse;
|
|
strictAssert(this.logger !== undefined, 'Logger not initialized');
|
|
this.logger[level](`MainSQL: ${format(...args)}`);
|
|
return;
|
|
}
|
|
|
|
const { seq, error, errorKind, response } = wrappedResponse;
|
|
|
|
const pair = this.onResponse.get(seq);
|
|
this.onResponse.delete(seq);
|
|
if (!pair) {
|
|
throw new Error(`Unexpected worker response with seq: ${seq}`);
|
|
}
|
|
|
|
if (error) {
|
|
const errorObj = new Error(error);
|
|
this.onError(errorKind ?? SqliteErrorKind.Unknown, errorObj);
|
|
|
|
pair.reject(errorObj);
|
|
} else {
|
|
pair.resolve(response);
|
|
}
|
|
});
|
|
|
|
const { promise: onExit, resolve: resolveOnExit } = explodePromise<void>();
|
|
this.onExit = onExit;
|
|
this.worker.once('exit', resolveOnExit);
|
|
}
|
|
|
|
public async initialize({
|
|
appVersion,
|
|
configDir,
|
|
key,
|
|
logger,
|
|
}: InitializeOptions): Promise<void> {
|
|
if (this.isReady || this.onReady) {
|
|
throw new Error('Already initialized');
|
|
}
|
|
|
|
this.shouldTimeQueries = Boolean(process.env.TIME_QUERIES);
|
|
|
|
this.logger = logger;
|
|
|
|
this.onReady = this.send({
|
|
type: 'init',
|
|
options: { appVersion, configDir, key },
|
|
});
|
|
|
|
await this.onReady;
|
|
|
|
this.onReady = undefined;
|
|
this.isReady = true;
|
|
}
|
|
|
|
public whenCorrupted(): Promise<Error> {
|
|
const { promise, resolve } = explodePromise<Error>();
|
|
this.errorResolvers.push({ kind: SqliteErrorKind.Corrupted, resolve });
|
|
return promise;
|
|
}
|
|
|
|
public whenReadonly(): Promise<Error> {
|
|
const { promise, resolve } = explodePromise<Error>();
|
|
this.errorResolvers.push({ kind: SqliteErrorKind.Readonly, resolve });
|
|
return promise;
|
|
}
|
|
|
|
public async close(): Promise<void> {
|
|
if (!this.isReady) {
|
|
throw new Error('Not initialized');
|
|
}
|
|
|
|
await this.send({ type: 'close' });
|
|
await this.onExit;
|
|
}
|
|
|
|
public async removeDB(): Promise<void> {
|
|
await this.send({ type: 'removeDB' });
|
|
}
|
|
|
|
public async sqlCall<Method extends keyof typeof DB>(
|
|
method: Method,
|
|
...args: Parameters<typeof DB[Method]>
|
|
): Promise<ReturnType<typeof DB[Method]>> {
|
|
if (this.onReady) {
|
|
await this.onReady;
|
|
}
|
|
|
|
if (!this.isReady) {
|
|
throw new Error('Not initialized');
|
|
}
|
|
|
|
type SqlCallResult = Readonly<{
|
|
result: ReturnType<typeof DB[Method]>;
|
|
duration: number;
|
|
}>;
|
|
|
|
const { result, duration } = await this.send<SqlCallResult>({
|
|
type: 'sqlCall',
|
|
method,
|
|
args,
|
|
});
|
|
|
|
if (this.shouldTimeQueries && !app.isPackaged) {
|
|
const twoDecimals = Math.round(100 * duration) / 100;
|
|
this.logger?.info(`MainSQL query: ${method}, duration=${twoDecimals}ms`);
|
|
}
|
|
if (duration > MIN_TRACE_DURATION) {
|
|
strictAssert(this.logger !== undefined, 'Logger not initialized');
|
|
this.logger.info(
|
|
`MainSQL: slow query ${method} duration=${Math.round(duration)}ms`
|
|
);
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
private async send<Response>(request: WorkerRequest): Promise<Response> {
|
|
const { seq } = this;
|
|
this.seq += 1;
|
|
|
|
const { promise: result, resolve, reject } = explodePromise<Response>();
|
|
this.onResponse.set(seq, { resolve, reject });
|
|
|
|
const wrappedRequest: WrappedWorkerRequest = {
|
|
seq,
|
|
request,
|
|
};
|
|
this.worker.postMessage(wrappedRequest);
|
|
|
|
return result;
|
|
}
|
|
|
|
private onError(errorKind: SqliteErrorKind, error: Error): void {
|
|
if (errorKind === SqliteErrorKind.Unknown) {
|
|
return;
|
|
}
|
|
|
|
const resolvers = new Array<(error: Error) => void>();
|
|
this.errorResolvers = this.errorResolvers.filter(entry => {
|
|
if (entry.kind === errorKind) {
|
|
resolvers.push(entry.resolve);
|
|
return false;
|
|
}
|
|
return true;
|
|
});
|
|
|
|
for (const resolve of resolvers) {
|
|
resolve(error);
|
|
}
|
|
}
|
|
}
|