// 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 { isCorruptionError } from './errors'; import type DB from './Server'; const MIN_TRACE_DURATION = 40; export type InitializeOptions = Readonly<{ configDir: string; key: string; logger: LoggerType; }>; export type WorkerRequest = Readonly< | { type: 'init'; options: Omit; } | { type: 'close'; } | { type: 'removeDB'; } | { type: 'sqlCall'; method: keyof typeof DB; args: ReadonlyArray; } >; export type WrappedWorkerRequest = Readonly<{ seq: number; request: WorkerRequest; }>; export type WrappedWorkerLogEntry = Readonly<{ type: 'log'; level: 'fatal' | 'error' | 'warn' | 'info' | 'debug' | 'trace'; args: ReadonlyArray; }>; export type WrappedWorkerResponse = | Readonly<{ type: 'response'; seq: number; error: string | undefined; // eslint-disable-next-line @typescript-eslint/no-explicit-any response: any; }> | WrappedWorkerLogEntry; type PromisePair = { resolve: (response: T) => void; reject: (error: Error) => void; }; export class MainSQL { private readonly worker: Worker; private isReady = false; private onReady: Promise | undefined; private readonly onExit: Promise; // This promise is resolved when any of the queries that we run against the // database reject with a corruption error (see `isCorruptionError`) private readonly onCorruption: Promise; private seq = 0; private logger?: LoggerType; // eslint-disable-next-line @typescript-eslint/no-explicit-any private onResponse = new Map>(); constructor() { const scriptDir = join(app.getAppPath(), 'ts', 'sql', 'mainWorker.js'); this.worker = new Worker(scriptDir); const { promise: onCorruption, resolve: resolveCorruption } = explodePromise(); this.onCorruption = onCorruption; 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, 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); if (isCorruptionError(errorObj)) { resolveCorruption(errorObj); } pair.reject(errorObj); } else { pair.resolve(response); } }); this.onExit = new Promise(resolve => { this.worker.once('exit', resolve); }); } public async initialize({ configDir, key, logger, }: InitializeOptions): Promise { if (this.isReady || this.onReady) { throw new Error('Already initialized'); } this.logger = logger; this.onReady = this.send({ type: 'init', options: { configDir, key }, }); await this.onReady; this.onReady = undefined; this.isReady = true; } public whenCorrupted(): Promise { return this.onCorruption; } public async close(): Promise { if (!this.isReady) { throw new Error('Not initialized'); } await this.send({ type: 'close' }); await this.onExit; } public async removeDB(): Promise { await this.send({ type: 'removeDB' }); } public async sqlCall( method: Method, ...args: Parameters ): Promise> { if (this.onReady) { await this.onReady; } if (!this.isReady) { throw new Error('Not initialized'); } type SqlCallResult = Readonly<{ result: ReturnType; duration: number; }>; const { result, duration } = await this.send({ type: 'sqlCall', method, args, }); if (duration > MIN_TRACE_DURATION) { strictAssert(this.logger !== undefined, 'Logger not initialized'); this.logger.info(`MainSQL: slow query ${method} duration=${duration}ms`); } return result; } private async send(request: WorkerRequest): Promise { const { seq } = this; this.seq += 1; const result = new Promise((resolve, reject) => { this.onResponse.set(seq, { resolve, reject }); }); const wrappedRequest: WrappedWorkerRequest = { seq, request, }; this.worker.postMessage(wrappedRequest); return result; } }