fix: event missing after redis reconnected (#674)

* chore: add automationGap config

* fix: event missing after redis reconnected

* fix: pnpm lock

apps/nestjs-backend/package.json

@@ -150,6 +150,7 @@
  "fs-extra": "11.2.0",
  "handlebars": "4.7.8",
  "helmet": "7.1.0",
+ "ioredis": "5.4.1",
  "is-port-reachable": "3.1.0",
  "joi": "17.12.2",
  "keyv": "4.5.4",
@@ -180,7 +181,6 @@
  "reflect-metadata": "0.2.1",
  "rxjs": "7.8.1",
  "sharedb": "4.1.2",
- "sharedb-redis-pubsub": "5.0.0",
  "sharp": "0.33.3",
  "transliteration": "2.3.5",
  "ts-pattern": "5.0.8",

apps/nestjs-backend/src/configs/threshold.config.ts

@@ -17,6 +17,7 @@ export const thresholdConfig = registerAs('threshold', () => ({
  bigTransactionTimeout: Number(
  process.env.BIG_TRANSACTION_TIMEOUT ?? 10 * 60 * 1000 /* 10 mins */
  ),
+ automationGap: Number(process.env.AUTOMATION_GAP ?? 200),
 }));
 
 export const ThresholdConfig = () => Inject(thresholdConfig.KEY);

apps/nestjs-backend/src/features/calculation/batch.service.ts

@@ -330,7 +330,7 @@ export class BatchService {
  },
  };
 
- this.logger.log(`saveOp: ${baseRaw.src}-${collection}`);
+ this.logger.verbose(`saveOp: ${baseRaw.src}-${collection}`);
 
  const rawOps = dataList.map(({ docId: docId, version, data }) => {
  let rawOp: IRawOp;

apps/nestjs-backend/src/share-db/share-db.service.ts

@@ -1,5 +1,5 @@
+import url from 'url';
 import { Injectable, Logger } from '@nestjs/common';
-import { loadPackage } from '@nestjs/common/utils/load-package.util';
 import { context as otelContext, trace as otelTrace } from '@opentelemetry/api';
 import { FieldOpBuilder, IdPrefix, ViewOpBuilder } from '@teable/core';
 import { PrismaService } from '@teable/db-main-prisma';
@@ -14,6 +14,7 @@ import { Timing } from '../utils/timing';
 import { authMiddleware } from './auth.middleware';
 import type { IRawOpMap } from './interface';
 import { ShareDbAdapter } from './share-db.adapter';
+import { RedisPubSub } from './sharedb-redis.pubsub';
 
 @Injectable()
 export class ShareDbService extends ShareDBClass {
@@ -33,11 +34,13 @@ export class ShareDbService extends ShareDBClass {
  });
 
  const { provider, redis } = this.cacheConfig;
-
  if (provider === 'redis') {
- const redisPubsub = loadPackage('sharedb-redis-pubsub', ShareDbService.name, () =>
- require('sharedb-redis-pubsub')
- )({ url: redis.uri });
+ const parsedURL = new url.URL(redis.uri as string);
+ const redisPubsub = new RedisPubSub({
+ host: parsedURL.hostname,
+ port: Number(parsedURL.port),
+ password: parsedURL.password,
+ });
 
  this.logger.log(`> Detected Redis cache; enabled the Redis pub/sub adapter for ShareDB.`);
  this.pubsub = redisPubsub;

apps/nestjs-backend/src/share-db/sharedb-redis.pubsub.ts

@@ -0,0 +1,76 @@
+/* eslint-disable @typescript-eslint/naming-convention */
+import Redis from 'ioredis';
+import type { RedisOptions } from 'ioredis';
+import type { Error } from 'sharedb';
+import { PubSub } from 'sharedb';
+
+const PUBLISH_SCRIPT = 'for i = 2, #ARGV do ' + 'redis.call("publish", ARGV[i], ARGV[1]) ' + 'end';
+
+// Redis pubsub driver for ShareDB.
+//
+// The redis driver requires two redis clients (a single redis client can't do
+// both pubsub and normal messaging). These clients will be created
+// automatically if you don't provide them.
+export class RedisPubSub extends PubSub {
+ client: Redis;
+ observer: Redis;
+ _closing?: boolean;
+
+ constructor(options: RedisOptions & { prefix?: string } = {}) {
+ super(options);
+
+ this.client = new Redis(options);
+
+ // Redis doesn't allow the same connection to both listen to channels and do
+ // operations. Make an extra redis connection for subscribing with the same
+ // options if not provided
+ this.observer = new Redis(options);
+ this.observer.on('message', this.handleMessage.bind(this));
+ }
+
+ close(
+ callback = function (err: Error | null) {
+ if (err) throw err;
+ }
+ ): void {
+ PubSub.prototype.close.call(this, (err) => {
+ if (err) return callback(err);
+ this._close().then(function () {
+ callback(null);
+ }, callback);
+ });
+ }
+
+ async _close() {
+ if (this._closing) {
+ return;
+ }
+ this._closing = true;
+ this.observer.removeAllListeners();
+ await Promise.all([this.client.quit(), this.observer.quit()]);
+ }
+
+ _subscribe(channel: string, callback: (err: Error | null) => void): void {
+ this.observer.subscribe(channel).then(function () {
+ callback(null);
+ }, callback);
+ }
+
+ handleMessage(channel: string, message: string) {
+ this._emit(channel, JSON.parse(message));
+ }
+
+ _unsubscribe(channel: string, callback: (err: Error | null) => void): void {
+ this.observer.unsubscribe(channel).then(function () {
+ callback(null);
+ }, callback);
+ }
+
+ async _publish(channels: string[], data: unknown, callback: (err: Error | null) => void) {
+ const message = JSON.stringify(data);
+ const args = [message].concat(channels);
+ this.client.eval(PUBLISH_SCRIPT, 0, ...args).then(function () {
+ callback(null);
+ }, callback);
+ }
+}

apps/nestjs-backend/src/utils/timing.ts

@@ -22,7 +22,7 @@ export function Timing(customLoggerKey?: string): MethodDecorator {
 
  const printLog = () => {
  const end = process.hrtime.bigint();
- logger.log(
+ logger.verbose(
  `${className} - ${String(customLoggerKey || propertyKey)} Execution Time: ${
  (end - start) / BigInt(1000000)
  } ms; Heap Usage: ${