✨ ooconnector: callback command retreival system (#525)

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');

tdrive/connectors/onlyoffice-connector/src/lib/pending-request-matcher.ts

@@ -0,0 +1,106 @@
+import logger from './logger';
+
+export type PendingRequestCallback<TResult> = (timeout: boolean, result?: TResult) => Promise<void>;
+
+/** Represents a single pending query that can be resolved with success or failure */
+class PendingRequest<TResult> {
+  private readonly startedAt = new Date();
+  private callbacks?: PendingRequestCallback<TResult>[] = [];
+
+  constructor(public readonly key: string, public readonly userdata: string, callback: PendingRequestCallback<TResult>) {
+    this.addCallback(callback);
+  }
+
+  public addCallback(callback: PendingRequestCallback<TResult>) {
+    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<PendingRequestCallback<TResult>>) {
+    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<TResult> {
+  private queue: PendingRequest<TResult>[] = [];
+
+  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<TResult>) {
+    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<TResult>) => boolean): PendingRequest<TResult>[] {
+    // 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<TResult>) => boolean, ...cbArgs: Parameters<PendingRequestCallback<TResult>>) {
+    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);
+  }
+}

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<ErrorResponse | TSuccessResponse> {
       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<Response> {
+      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<ErrorCode, ErrorCode.SUCCESS> | 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<CommandService.License.Response>;
+  // Technically the timeout field is from the PendingRequestQueue but avoid 2 classes
+  private readonly pendingRequests = new PendingRequestQueue<CallbackResponseFromCommand>(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<CallbackResponseFromCommand>): 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<ErrorCode> {
+    return await new CommandService.Info.Request(key, userdata).postUnsafe().then(({ error }) => error);
+  }
+
+  async getInfoAndWaitForCallbackUnsafe(key: string): Promise<CallbackResponseFromCommand> {
+    // 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<string> {
     return new CommandService.ForceSave.Request(key, userdata).post().then(response => response.key);