From 5cf1fa581e724be8777643bec7e498984fcc954e Mon Sep 17 00:00:00 2001 From: Eric Doughty-Papassideris Date: Sun, 15 Sep 2024 01:25:59 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20ooconnector:=20callback=20command?= =?UTF-8?q?=20retreival=20system=20(#525)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/controllers/onlyoffice.controller.ts | 18 ++- .../src/lib/pending-request-matcher.ts | 106 ++++++++++++++++++ .../src/services/onlyoffice.service.ts | 89 ++++++++++++++- 3 files changed, 209 insertions(+), 4 deletions(-) create mode 100644 tdrive/connectors/onlyoffice-connector/src/lib/pending-request-matcher.ts diff --git a/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts b/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts index fcd16c84..1605dcfc 100644 --- a/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts +++ b/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts @@ -6,6 +6,7 @@ import logger from '@/lib/logger'; import * as OnlyOffice from '@/services/onlyoffice.service'; import { Request, Response, NextFunction } from 'express'; import jwt from 'jsonwebtoken'; +import * as Utils from '@/utils'; interface RequestQuery { company_id: string; @@ -63,11 +64,24 @@ class OnlyOfficeController { try { const { url, key } = req.body; const { token } = req.query; - logger.info('OO callback', req.body); - + logger.info( + `OO callback status: ${Utils.getKeyForValueSafe( + req.body.status, + OnlyOffice.Callback.Status, + 'OnlyOffice.Callback.Status', + )} - ${JSON.stringify(req.body.userdata)}`, + req.body, + ); const officeTokenPayload = jwt.verify(token, CREDENTIALS_SECRET) as OfficeToken; const { preview, company_id, file_id, /* user_id, */ drive_file_id, in_page_token, editing_session_key } = officeTokenPayload; + // Ignore errors generated by pending request + // try-catch not needed because it is async + // there may be later reasons to wait for callbacks + // to process and eventually respond accordingly to + // OO an error for certain statuses + void OnlyOffice.default.ooCallbackCalled(req.body); + // check token is an in_page_token and allow save if (!in_page_token) throw new Error('Invalid token, must be a in_page_token'); if (preview) throw new Error('Invalid token, must not be a preview token for save operation'); diff --git a/tdrive/connectors/onlyoffice-connector/src/lib/pending-request-matcher.ts b/tdrive/connectors/onlyoffice-connector/src/lib/pending-request-matcher.ts new file mode 100644 index 00000000..c739d47d --- /dev/null +++ b/tdrive/connectors/onlyoffice-connector/src/lib/pending-request-matcher.ts @@ -0,0 +1,106 @@ +import logger from './logger'; + +export type PendingRequestCallback = (timeout: boolean, result?: TResult) => Promise; + +/** Represents a single pending query that can be resolved with success or failure */ +class PendingRequest { + private readonly startedAt = new Date(); + private callbacks?: PendingRequestCallback[] = []; + + constructor(public readonly key: string, public readonly userdata: string, callback: PendingRequestCallback) { + this.addCallback(callback); + } + + public addCallback(callback: PendingRequestCallback) { + if (!this.callbacks) throw new Error("Cannot add callback to PendingRequest after it's been resolved"); + this.callbacks.push(callback); + return this; + } + + public getAge() { + return new Date().getTime() - this.startedAt.getTime(); + } + + public matches(key: string, userdata: string) { + return this.key === key && this.userdata === userdata; + } + + public async resolve(...cbArgs: Parameters>) { + if (!this.callbacks) throw new Error("Cannot resolve PendingRequest after it's already been resolved"); + const callbacks = this.callbacks; + this.callbacks = null; + return Promise.all(callbacks.map(fn => fn(...cbArgs))); + } +} + +/** + * Tracks a list of pending requests that are started with a pair + * of `key` and `userdata` strings. (Note. `userdata` is not the + * typical use of user data as a `void *`, but instead used to + * identify the specific request asynchroneously). + * Both must match. + * + * There is no timing garantee as to when expired requests' callbacks + * are executed. + */ +export class PendingRequestQueue { + private queue: PendingRequest[] = []; + + constructor(private readonly timeoutMs: number, niquystSamplingRatio: 4) { + setInterval(() => { + this.flush(); + }, timeoutMs / niquystSamplingRatio); + } + + /** + * Add a callback to be called when `gotResult` or `cancelPending` are + * called with identical `key` and `userdata` + * @param key First half of identifying string + * @param userdata Second half of identifying string + * @param callback When the request is resolved, succesfully or in error + */ + public enqueue(key: string, userdata: string, callback: PendingRequestCallback) { + const existing = this.queue.find(pending => pending.key === key && pending.userdata === userdata); + if (existing) return void existing.addCallback(callback); + this.queue.push(new PendingRequest(key, userdata, callback)); + } + + private remove(predicate: (request: PendingRequest) => boolean): PendingRequest[] { + // Warning: This operation must be synchroneous + const foundRequests = []; + this.queue = this.queue.filter(pending => { + if (!predicate(pending)) return true; + foundRequests.push(pending); + return false; + }); + return foundRequests; + } + + private async resolve(predicate: (request: PendingRequest) => boolean, ...cbArgs: Parameters>) { + const foundRequests = this.remove(predicate); + if (!foundRequests.length) return null; + return Promise.all(foundRequests.map(request => request.resolve(...cbArgs))); + } + + protected async flush() { + const expiredRequests = this.remove(request => request.getAge() > this.timeoutMs); + return Promise.all(expiredRequests.map(request => request.resolve(true))); + } + + public async gotResult(key: string, userdata: string, result?: TResult) { + const resolutions = this.resolve(request => request.matches(key, userdata), false, result); + if (!resolutions) + logger.error(`Got resolution on pending request that was unknown`, { + key, + userdata, + result, + }); + await this.flush(); + return resolutions; + } + + /** Equivalent to `gotResult` with an `undefined` result. (Callbacks are called) */ + public async cancelPending(key: string, userdata: string) { + return this.gotResult(key, userdata, undefined); + } +} diff --git a/tdrive/connectors/onlyoffice-connector/src/services/onlyoffice.service.ts b/tdrive/connectors/onlyoffice-connector/src/services/onlyoffice.service.ts index 887b6363..9b67f2fd 100644 --- a/tdrive/connectors/onlyoffice-connector/src/services/onlyoffice.service.ts +++ b/tdrive/connectors/onlyoffice-connector/src/services/onlyoffice.service.ts @@ -1,6 +1,8 @@ +import { randomUUID } from 'crypto'; import axios from 'axios'; -import { ONLY_OFFICE_SERVER, onlyOfficeConnectivityCheckPeriodMS } from '@config'; +import { ONLY_OFFICE_SERVER, onlyOfficeConnectivityCheckPeriodMS, onlyOfficeCallbackTimeoutMS } from '@config'; import { PolledThingieValue } from '@/lib/polled-thingie-value'; +import { PendingRequestQueue, PendingRequestCallback } from '@/lib/pending-request-matcher'; import logger from '@/lib/logger'; import * as Utils from '@/utils'; @@ -14,6 +16,12 @@ export enum ErrorCode { COMMAND_NOT_CORRECT = 5, INVALID_TOKEN = 6, } + +/** Error generated by this connector. Should be string to share field with {@link ErrorCode} */ +export enum ConnectorErrorCode { + CALLBACK_TIMEOUT = 'callback_timeout', +} + /** Return the name of the error code in the `ErrorCode` enum if recognised, or a descript string */ export const ErrorCodeFromValue = (value: number) => Utils.getKeyForValueSafe(value, ErrorCode, 'OnlyOffice.ErrorCode'); @@ -59,6 +67,7 @@ export namespace Callback { url?: string; actions?: Action[]; users?: string[]; + userdata?: string; } } @@ -93,7 +102,7 @@ namespace CommandService { async postUnsafe(): Promise { logger.silly(`OnlyOffice command ${this.c} sent: ${JSON.stringify(this)}`); const result = await axios.post(Utils.joinURL([ONLY_OFFICE_SERVER, 'coauthoring/CommandService.ashx']), this); - logger.info(`OnlyOffice command ${this.c} response: ${result.status}: ${JSON.stringify(result.data)}`); + logger.info(`OnlyOffice command ${this.c} response ${result.status}: ${ErrorCodeFromValue(result.data.error)}: ${JSON.stringify(result.data)}`); return result.data as ErrorResponse | TSuccessResponse; } @@ -174,6 +183,32 @@ namespace CommandService { } } } + + export namespace Info { + export type Response = SuccessResponse; + export class Request extends BaseRequest { + constructor(public readonly key: string, public readonly userdata: string = '') { + super('info'); + } + } + } +} + +/** + * This object holds possible outcomes for commands like `info` who's result + * is sent to the callback instead of replied to the request + */ +class CallbackResponseFromCommand { + /** + * If the `info` command returned an error code, this is it, or it can + * be an internal error to this connector (eg.: timeout waiting for callback). + * (If the `info` command returned {@link ErrorCode.SUCCESS} then this field will be `undefined`) + */ + public readonly error: Exclude | ConnectorErrorCode | undefined; + + public constructor(error: ErrorCode | ConnectorErrorCode, public readonly result?: Callback.Parameters) { + this.error = error === ErrorCode.SUCCESS ? undefined : error; + } } /** @@ -182,6 +217,8 @@ namespace CommandService { */ class OnlyOfficeService { private readonly poller: PolledThingieValue; + // Technically the timeout field is from the PendingRequestQueue but avoid 2 classes + private readonly pendingRequests = new PendingRequestQueue(onlyOfficeCallbackTimeoutMS); constructor() { this.poller = new PolledThingieValue( @@ -227,6 +264,20 @@ class OnlyOfficeService { return deleted; } + /** Generates and returns a random UUID that has a matching pending task enqueued for */ + private enqueuePendingCallback(key: string, callback: PendingRequestCallback): string { + const userdata = randomUUID(); + this.pendingRequests.enqueue(key, userdata, callback); + return userdata; + } + + /** Called by the OnlyOffice controller when the OO document editing services uses our callback */ + async ooCallbackCalled(result: Callback.Parameters) { + if (!result.userdata) return; + logger.info('OO Callback pending request response received', result); + return this.pendingRequests.gotResult(result.key, result.userdata, new CallbackResponseFromCommand(ErrorCode.SUCCESS, result)); + } + // Note that `async` is important in the functions below. While they avoid the overhead // of `await`, the `async` is still required to catch the throw in `.post()` @@ -239,6 +290,40 @@ class OnlyOfficeService { //TODO: When typing the response more fully, don't return the response object itself as here return new CommandService.License.Request().post(); } + + /** + * Requests a document status and the list of the identifiers of the users who opened the document for editing. + * The response will be sent to the callback handler. + * This method just sends the command. The response from the callback will be ignored if + * this called by itself + * + * *Warning*: returns non succesful error codes instead of throwing errors like the other + * methods. This is because the immediate response is likely to be itself usefull + * for detecting errors. + */ + async getInfoUnsafe(key: string, userdata: string): Promise { + return await new CommandService.Info.Request(key, userdata).postUnsafe().then(({ error }) => error); + } + + async getInfoAndWaitForCallbackUnsafe(key: string): Promise { + // const userdata = randomUUID(); + return new Promise((resolve, reject) => { + const userdata = this.enqueuePendingCallback(key, async (timeout, result) => { + // The callback has called, unless timeout = true, or result is undefined (cancelled request) + return resolve(timeout ? new CallbackResponseFromCommand(ConnectorErrorCode.CALLBACK_TIMEOUT) : result); + }); + void this.getInfoUnsafe(key, userdata).then( + response => { + // The command service responded to the `info` command request + if (response !== ErrorCode.SUCCESS) return this.pendingRequests.gotResult(key, userdata, new CallbackResponseFromCommand(response)); + // If it succeded, just wait for timeout or resolution by the callback + }, + error => { + this.pendingRequests.cancelPending(key, userdata).then(() => reject(error)); + }, + ); + }); + } /** Force a save in the editing session key provided. `userdata` will be forwarded to the callback */ async forceSave(key: string, userdata = ''): Promise { return new CommandService.ForceSave.Request(key, userdata).post().then(response => response.key);