comm/chat/protocols/matrix/lib/matrix-sdk/ToDeviceMessageQueue.js


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

"use strict";

Object.defineProperty(exports, "__esModule", {
  value: true
});
exports.ToDeviceMessageQueue = void 0;
var _event = require("./@types/event");
var _logger = require("./logger");
var _client = require("./client");
var _scheduler = require("./scheduler");
var _sync = require("./sync");
var _utils = require("./utils");
function _defineProperty(obj, key, value) { key = _toPropertyKey(key); if (key in obj) { Object.defineProperty(obj, key, { value: value, enumerable: true, configurable: true, writable: true }); } else { obj[key] = value; } return obj; }
function _toPropertyKey(arg) { var key = _toPrimitive(arg, "string"); return typeof key === "symbol" ? key : String(key); }
function _toPrimitive(input, hint) { if (typeof input !== "object" || input === null) return input; var prim = input[Symbol.toPrimitive]; if (prim !== undefined) { var res = prim.call(input, hint || "default"); if (typeof res !== "object") return res; throw new TypeError("@@toPrimitive must return a primitive value."); } return (hint === "string" ? String : Number)(input); } /*
                                                                                                                                                                                                                                                                                                                                                                                          Copyright 2022 The Matrix.org Foundation C.I.C.
                                                                                                                                                                                                                                                                                                                                                                                          
                                                                                                                                                                                                                                                                                                                                                                                          Licensed under the Apache License, Version 2.0 (the "License");
                                                                                                                                                                                                                                                                                                                                                                                          you may not use this file except in compliance with the License.
                                                                                                                                                                                                                                                                                                                                                                                          You may obtain a copy of the License at
                                                                                                                                                                                                                                                                                                                                                                                          
                                                                                                                                                                                                                                                                                                                                                                                              http://www.apache.org/licenses/LICENSE-2.0
                                                                                                                                                                                                                                                                                                                                                                                          
                                                                                                                                                                                                                                                                                                                                                                                          Unless required by applicable law or agreed to in writing, software
                                                                                                                                                                                                                                                                                                                                                                                          distributed under the License is distributed on an "AS IS" BASIS,
                                                                                                                                                                                                                                                                                                                                                                                          WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
                                                                                                                                                                                                                                                                                                                                                                                          See the License for the specific language governing permissions and
                                                                                                                                                                                                                                                                                                                                                                                          limitations under the License.
                                                                                                                                                                                                                                                                                                                                                                                          */
const MAX_BATCH_SIZE = 20;

/**
 * Maintains a queue of outgoing to-device messages, sending them
 * as soon as the homeserver is reachable.
 */
class ToDeviceMessageQueue {
  constructor(client) {
    this.client = client;
    _defineProperty(this, "sending", false);
    _defineProperty(this, "running", true);
    _defineProperty(this, "retryTimeout", null);
    _defineProperty(this, "retryAttempts", 0);
    _defineProperty(this, "sendQueue", async () => {
      if (this.retryTimeout !== null) clearTimeout(this.retryTimeout);
      this.retryTimeout = null;
      if (this.sending || !this.running) return;
      _logger.logger.debug("Attempting to send queued to-device messages");
      this.sending = true;
      let headBatch;
      try {
        while (this.running) {
          headBatch = await this.client.store.getOldestToDeviceBatch();
          if (headBatch === null) break;
          await this.sendBatch(headBatch);
          await this.client.store.removeToDeviceBatch(headBatch.id);
          this.retryAttempts = 0;
        }

        // Make sure we're still running after the async tasks: if not, stop.
        if (!this.running) return;
        _logger.logger.debug("All queued to-device messages sent");
      } catch (e) {
        ++this.retryAttempts;
        // eslint-disable-next-line @typescript-eslint/naming-convention
        // eslint-disable-next-line new-cap
        const retryDelay = _scheduler.MatrixScheduler.RETRY_BACKOFF_RATELIMIT(null, this.retryAttempts, e);
        if (retryDelay === -1) {
          // the scheduler function doesn't differentiate between fatal errors and just getting
          // bored and giving up for now
          if (Math.floor(e.httpStatus / 100) === 4) {
            _logger.logger.error("Fatal error when sending to-device message - dropping to-device batch!", e);
            await this.client.store.removeToDeviceBatch(headBatch.id);
          } else {
            _logger.logger.info("Automatic retry limit reached for to-device messages.");
          }
          return;
        }
        _logger.logger.info(`Failed to send batch of to-device messages. Will retry in ${retryDelay}ms`, e);
        this.retryTimeout = setTimeout(this.sendQueue, retryDelay);
      } finally {
        this.sending = false;
      }
    });
    /**
     * Listen to sync state changes and automatically resend any pending events
     * once syncing is resumed
     */
    _defineProperty(this, "onResumedSync", (state, oldState) => {
      if (state === _sync.SyncState.Syncing && oldState !== _sync.SyncState.Syncing) {
        _logger.logger.info(`Resuming queue after resumed sync`);
        this.sendQueue();
      }
    });
  }
  start() {
    this.running = true;
    this.sendQueue();
    this.client.on(_client.ClientEvent.Sync, this.onResumedSync);
  }
  stop() {
    this.running = false;
    if (this.retryTimeout !== null) clearTimeout(this.retryTimeout);
    this.retryTimeout = null;
    this.client.removeListener(_client.ClientEvent.Sync, this.onResumedSync);
  }
  async queueBatch(batch) {
    const batches = [];
    for (let i = 0; i < batch.batch.length; i += MAX_BATCH_SIZE) {
      const batchWithTxnId = {
        eventType: batch.eventType,
        batch: batch.batch.slice(i, i + MAX_BATCH_SIZE),
        txnId: this.client.makeTxnId()
      };
      batches.push(batchWithTxnId);
      const msgmap = batchWithTxnId.batch.map(msg => `${msg.userId}/${msg.deviceId} (msgid ${msg.payload[_event.ToDeviceMessageId]})`);
      _logger.logger.info(`Enqueuing batch of to-device messages. type=${batch.eventType} txnid=${batchWithTxnId.txnId}`, msgmap);
    }
    await this.client.store.saveToDeviceBatches(batches);
    this.sendQueue();
  }
  /**
   * Attempts to send a batch of to-device messages.
   */
  async sendBatch(batch) {
    const contentMap = new _utils.MapWithDefault(() => new Map());
    for (const item of batch.batch) {
      contentMap.getOrCreate(item.userId).set(item.deviceId, item.payload);
    }
    _logger.logger.info(`Sending batch of ${batch.batch.length} to-device messages with ID ${batch.id} and txnId ${batch.txnId}`);
    await this.client.sendToDevice(batch.eventType, contentMap, batch.txnId);
  }
}
exports.ToDeviceMessageQueue = ToDeviceMessageQueue;