refactor(server): plugin modules (#5630)

- [x] separates modules into `fundamental`, `core`, `plugins` - [x] optional modules with `@OptionalModule` decorator to install modules with requirements met(`requires`, `if`) - [x] `module.contributesTo` defines optional features that will be enabled if module registered - [x] `AFFiNE.plugins.use('payment', {})` to enable a optional/plugin module - [x] `PaymentModule` is the first plugin module - [x] GraphQLSchema will not be generated for non-included modules - [x] Frontend can use `ServerConfigType` query to detect which features are enabled - [x] override existing provider globally
2026-07-19 02:56:23 +08:00 · 2024-01-22 07:40:28 +00:00
parent ae8401b6f4
commit e516e0db23
130 changed files with 1297 additions and 974 deletions
@@ -0,0 +1,194 @@
+import { Redis } from 'ioredis';
+
+import type { Cache, CacheSetOptions } from '../../fundamentals/cache/def';
+
+export class RedisCache implements Cache {
+  constructor(private readonly redis: Redis) {}
+
+  // standard operation
+  async get<T = unknown>(key: string): Promise<T> {
+    return this.redis
+      .get(key)
+      .then(v => {
+        if (v) {
+          return JSON.parse(v);
+        }
+        return undefined;
+      })
+      .catch(() => undefined);
+  }
+
+  async set<T = unknown>(
+    key: string,
+    value: T,
+    opts: CacheSetOptions = {}
+  ): Promise<boolean> {
+    if (opts.ttl) {
+      return this.redis
+        .set(key, JSON.stringify(value), 'PX', opts.ttl)
+        .then(() => true)
+        .catch(() => false);
+    }
+
+    return this.redis
+      .set(key, JSON.stringify(value))
+      .then(() => true)
+      .catch(() => false);
+  }
+
+  async increase(key: string, count: number = 1): Promise<number> {
+    return this.redis.incrby(key, count).catch(() => 0);
+  }
+
+  async decrease(key: string, count: number = 1): Promise<number> {
+    return this.redis.decrby(key, count).catch(() => 0);
+  }
+
+  async setnx<T = unknown>(
+    key: string,
+    value: T,
+    opts: CacheSetOptions = {}
+  ): Promise<boolean> {
+    if (opts.ttl) {
+      return this.redis
+        .set(key, JSON.stringify(value), 'PX', opts.ttl, 'NX')
+        .then(v => !!v)
+        .catch(() => false);
+    }
+
+    return this.redis
+      .set(key, JSON.stringify(value), 'NX')
+      .then(v => !!v)
+      .catch(() => false);
+  }
+
+  async delete(key: string): Promise<boolean> {
+    return this.redis
+      .del(key)
+      .then(v => v > 0)
+      .catch(() => false);
+  }
+
+  async has(key: string): Promise<boolean> {
+    return this.redis
+      .exists(key)
+      .then(v => v > 0)
+      .catch(() => false);
+  }
+
+  async ttl(key: string): Promise<number> {
+    return this.redis.ttl(key).catch(() => 0);
+  }
+
+  async expire(key: string, ttl: number): Promise<boolean> {
+    return this.redis
+      .pexpire(key, ttl)
+      .then(v => v > 0)
+      .catch(() => false);
+  }
+
+  // list operations
+  async pushBack<T = unknown>(key: string, ...values: T[]): Promise<number> {
+    return this.redis
+      .rpush(key, ...values.map(v => JSON.stringify(v)))
+      .catch(() => 0);
+  }
+
+  async pushFront<T = unknown>(key: string, ...values: T[]): Promise<number> {
+    return this.redis
+      .lpush(key, ...values.map(v => JSON.stringify(v)))
+      .catch(() => 0);
+  }
+
+  async len(key: string): Promise<number> {
+    return this.redis.llen(key).catch(() => 0);
+  }
+
+  async list<T = unknown>(
+    key: string,
+    start: number,
+    end: number
+  ): Promise<T[]> {
+    return this.redis
+      .lrange(key, start, end)
+      .then(data => data.map(v => JSON.parse(v)))
+      .catch(() => []);
+  }
+
+  async popFront<T = unknown>(key: string, count: number = 1): Promise<T[]> {
+    return this.redis
+      .lpop(key, count)
+      .then(data => (data ?? []).map(v => JSON.parse(v)))
+      .catch(() => []);
+  }
+
+  async popBack<T = unknown>(key: string, count: number = 1): Promise<T[]> {
+    return this.redis
+      .rpop(key, count)
+      .then(data => (data ?? []).map(v => JSON.parse(v)))
+      .catch(() => []);
+  }
+
+  // map operations
+  async mapSet<T = unknown>(
+    map: string,
+    key: string,
+    value: T
+  ): Promise<boolean> {
+    return this.redis
+      .hset(map, key, JSON.stringify(value))
+      .then(v => v > 0)
+      .catch(() => false);
+  }
+
+  async mapIncrease(
+    map: string,
+    key: string,
+    count: number = 1
+  ): Promise<number> {
+    return this.redis.hincrby(map, key, count);
+  }
+
+  async mapDecrease(
+    map: string,
+    key: string,
+    count: number = 1
+  ): Promise<number> {
+    return this.redis.hincrby(map, key, -count);
+  }
+
+  async mapGet<T = unknown>(map: string, key: string): Promise<T | undefined> {
+    return this.redis
+      .hget(map, key)
+      .then(v => (v ? JSON.parse(v) : undefined))
+      .catch(() => undefined);
+  }
+
+  async mapDelete(map: string, key: string): Promise<boolean> {
+    return this.redis
+      .hdel(map, key)
+      .then(v => v > 0)
+      .catch(() => false);
+  }
+
+  async mapKeys(map: string): Promise<string[]> {
+    return this.redis.hkeys(map).catch(() => []);
+  }
+
+  async mapRandomKey(map: string): Promise<string | undefined> {
+    return this.redis
+      .hrandfield(map, 1)
+      .then(v =>
+        typeof v === 'string'
+          ? v
+          : Array.isArray(v)
+            ? (v[0] as string)
+            : undefined
+      )
+      .catch(() => undefined);
+  }
+
+  async mapLen(map: string): Promise<number> {
+    return this.redis.hlen(map).catch(() => 0);
+  }
+}
@@ -0,0 +1,62 @@
+import { Global, Provider, Type } from '@nestjs/common';
+import { Redis, type RedisOptions } from 'ioredis';
+import { ThrottlerStorageRedisService } from 'nestjs-throttler-storage-redis';
+
+import { Cache, OptionalModule, SessionCache } from '../../fundamentals';
+import { ThrottlerStorage } from '../../fundamentals/throttler';
+import { SocketIoAdapterImpl } from '../../fundamentals/websocket';
+import { RedisCache } from './cache';
+import {
+  CacheRedis,
+  SessionRedis,
+  SocketIoRedis,
+  ThrottlerRedis,
+} from './instances';
+import { createSockerIoAdapterImpl } from './ws-adapter';
+
+function makeProvider(token: Type, impl: Type<Redis>): Provider {
+  return {
+    provide: token,
+    useFactory: (redis: Redis) => {
+      return new RedisCache(redis);
+    },
+    inject: [impl],
+  };
+}
+
+// cache
+const cacheProvider = makeProvider(Cache, CacheRedis);
+const sessionCacheProvider = makeProvider(SessionCache, SessionRedis);
+
+// throttler
+const throttlerStorageProvider: Provider = {
+  provide: ThrottlerStorage,
+  useFactory: (redis: Redis) => {
+    return new ThrottlerStorageRedisService(redis);
+  },
+  inject: [ThrottlerRedis],
+};
+
+// socket io
+const socketIoRedisAdapterProvider: Provider = {
+  provide: SocketIoAdapterImpl,
+  useFactory: (redis: Redis) => {
+    return createSockerIoAdapterImpl(redis);
+  },
+  inject: [SocketIoRedis],
+};
+
+@Global()
+@OptionalModule({
+  providers: [CacheRedis, SessionRedis, ThrottlerRedis, SocketIoRedis],
+  overrides: [
+    cacheProvider,
+    sessionCacheProvider,
+    socketIoRedisAdapterProvider,
+    throttlerStorageProvider,
+  ],
+  requires: ['plugins.redis.host'],
+})
+export class RedisModule {}
+
+export { RedisOptions };
@@ -0,0 +1,56 @@
+import {
+  Injectable,
+  Logger,
+  OnModuleDestroy,
+  OnModuleInit,
+} from '@nestjs/common';
+import { Redis as IORedis, RedisOptions } from 'ioredis';
+
+import { Config } from '../../fundamentals/config';
+
+class Redis extends IORedis implements OnModuleDestroy, OnModuleInit {
+  logger = new Logger(Redis.name);
+  constructor(opts: RedisOptions) {
+    super({
+      ...opts,
+      lazyConnect: true,
+    });
+  }
+
+  async onModuleInit() {
+    await this.connect().catch(() => {
+      this.logger.error('Failed to connect to Redis server.');
+    });
+  }
+  onModuleDestroy() {
+    this.disconnect();
+  }
+}
+
+@Injectable()
+export class CacheRedis extends Redis {
+  constructor(config: Config) {
+    super(config.plugins.redis ?? {});
+  }
+}
+
+@Injectable()
+export class ThrottlerRedis extends Redis {
+  constructor(config: Config) {
+    super({ ...config.plugins.redis, db: (config.plugins.redis?.db ?? 0) + 1 });
+  }
+}
+
+@Injectable()
+export class SessionRedis extends Redis {
+  constructor(config: Config) {
+    super({ ...config.plugins.redis, db: (config.plugins.redis?.db ?? 0) + 2 });
+  }
+}
+
+@Injectable()
+export class SocketIoRedis extends Redis {
+  constructor(config: Config) {
+    super({ ...config.plugins.redis, db: (config.plugins.redis?.db ?? 0) + 3 });
+  }
+}
@@ -0,0 +1,3 @@
+import { RedisOptions } from 'ioredis';
+
+export type { RedisOptions };
@@ -0,0 +1,28 @@
+import { createAdapter } from '@socket.io/redis-adapter';
+import { Redis } from 'ioredis';
+import { Server, ServerOptions } from 'socket.io';
+
+import { SocketIoAdapter } from '../../fundamentals';
+
+export function createSockerIoAdapterImpl(
+  redis: Redis
+): typeof SocketIoAdapter {
+  class RedisIoAdapter extends SocketIoAdapter {
+    override createIOServer(port: number, options?: ServerOptions): Server {
+      const pubClient = redis;
+      pubClient.on('error', err => {
+        console.error(err);
+      });
+      const subClient = pubClient.duplicate();
+      subClient.on('error', err => {
+        console.error(err);
+      });
+
+      const server = super.createIOServer(port, options) as Server;
+      server.adapter(createAdapter(pubClient, subClient));
+      return server;
+    }
+  }
+
+  return RedisIoAdapter;
+}