Jayden/AFFiNE-Mirror
0
0
Fork 0
mirror of https://github.com/toeverything/AFFiNE.git synced 2026-07-12 23:56:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(server): make redis required module (#9121)

Browse Source
This commit is contained in:
forehalo
2024-12-13 06:27:15 +00:00
parent 81c68032e1
commit 0e73737407
40 changed files with 292 additions and 728 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/backend/server/src/plugins/copilot/resolver.ts
+4 -4
View File
@@ -359,7 +359,7 @@ export class CopilotResolver {
user.id
);
const lockFlag = `${COPILOT_LOCKER}:session:${user.id}:${options.workspaceId}`;
await using lock = await this.mutex.lock(lockFlag);
await using lock = await this.mutex.acquire(lockFlag);
if (!lock) {
return new TooManyRequest('Server is busy');
}
@@ -387,7 +387,7 @@ export class CopilotResolver {
user.id
);
const lockFlag = `${COPILOT_LOCKER}:session:${user.id}:${options.workspaceId}`;
await using lock = await this.mutex.lock(lockFlag);
await using lock = await this.mutex.acquire(lockFlag);
if (!lock) {
return new TooManyRequest('Server is busy');
}
@@ -418,7 +418,7 @@ export class CopilotResolver {
return new NotFoundException('Session not found');
}
const lockFlag = `${COPILOT_LOCKER}:session:${user.id}:${options.workspaceId}`;
await using lock = await this.mutex.lock(lockFlag);
await using lock = await this.mutex.acquire(lockFlag);
if (!lock) {
return new TooManyRequest('Server is busy');
}
@@ -439,7 +439,7 @@ export class CopilotResolver {
options: CreateChatMessageInput
) {
const lockFlag = `${COPILOT_LOCKER}:message:${user?.id}:${options.sessionId}`;
await using lock = await this.mutex.lock(lockFlag);
await using lock = await this.mutex.acquire(lockFlag);
if (!lock) {
return new TooManyRequest('Server is busy');
}
packages/backend/server/src/plugins/index.ts
-1
View File
@@ -3,7 +3,6 @@ import './copilot';
import './gcloud';
import './oauth';
import './payment';
import './redis';
import './storage';
export {
packages/backend/server/src/plugins/payment/service.ts
+1 -1
View File
@@ -555,7 +555,7 @@ export class SubscriptionService implements OnApplicationBootstrap {
return;
}
await using lock = await this.mutex.lock('init stripe prices');
await using lock = await this.mutex.acquire('init stripe prices');
if (!lock) {
return;
packages/backend/server/src/plugins/redis/cache.ts
-194
View File
@@ -1,194 +0,0 @@
import { Redis } from 'ioredis';
import type { Cache, CacheSetOptions } from '../../base/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);
}
}
packages/backend/server/src/plugins/redis/config.ts
-11
View File
@@ -1,11 +0,0 @@
import { RedisOptions } from 'ioredis';
import { defineStartupConfig, ModuleConfig } from '../../base/config';
declare module '../config' {
interface PluginsConfig {
redis: ModuleConfig<RedisOptions>;
}
}
defineStartupConfig('plugins.redis', {});
packages/backend/server/src/plugins/redis/index.ts
-67
View File
@@ -1,67 +0,0 @@
import './config';
import { Global, Provider, Type } from '@nestjs/common';
import { Redis } from 'ioredis';
import { ThrottlerStorageRedisService } from 'nestjs-throttler-storage-redis';
import { Cache, Locker, SessionCache } from '../../base';
import { ThrottlerStorage } from '../../base/throttler';
import { SocketIoAdapterImpl } from '../../base/websocket';
import { Plugin } from '../registry';
import { RedisCache } from './cache';
import { CacheRedis, SessionRedis, SocketIoRedis } from './instances';
import { RedisMutexLocker } from './mutex';
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: [SessionRedis],
};
// socket io
const socketIoRedisAdapterProvider: Provider = {
provide: SocketIoAdapterImpl,
useFactory: (redis: Redis) => {
return createSockerIoAdapterImpl(redis);
},
inject: [SocketIoRedis],
};
// mutex
const mutexRedisAdapterProvider: Provider = {
provide: Locker,
useClass: RedisMutexLocker,
};
@Global()
@Plugin({
name: 'redis',
providers: [CacheRedis, SessionRedis, SocketIoRedis],
overrides: [
cacheProvider,
sessionCacheProvider,
socketIoRedisAdapterProvider,
throttlerStorageProvider,
mutexRedisAdapterProvider,
],
requires: ['plugins.redis.host'],
})
export class RedisModule {}
packages/backend/server/src/plugins/redis/instances.ts
-49
View File
@@ -1,49 +0,0 @@
import {
Injectable,
Logger,
OnModuleDestroy,
OnModuleInit,
} from '@nestjs/common';
import { Redis as IORedis, RedisOptions } from 'ioredis';
import { Config } from '../../base/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 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 });
}
}
packages/backend/server/src/plugins/redis/mutex.ts
-65
View File
@@ -1,65 +0,0 @@
import { Injectable, Logger } from '@nestjs/common';
import { Command } from 'ioredis';
import { ILocker, Lock } from '../../base';
import { SessionRedis } from './instances';
// === atomic mutex lock ===
// acquire lock
// return 1 if lock is acquired
// return 0 if lock is not acquired
const lockScript = `local key = KEYS[1]
local owner = ARGV[1]
-- if lock is not exists or lock is owned by the owner
-- then set lock to the owner and return 1, otherwise return 0
-- if the lock is not released correctly due to unexpected reasons
-- lock will be released after 60 seconds
if redis.call("get", key) == owner or redis.call("set", key, owner, "NX", "EX", 60) then
return 1
else
return 0
end`;
// release lock
// return 1 if lock is released or lock is not exists
// return 0 if lock is not owned by the owner
const unlockScript = `local key = KEYS[1]
local owner = ARGV[1]
local value = redis.call("get", key)
if value == owner then
return redis.call("del", key)
elseif value == nil then
return 1
else
return 0
end`;
@Injectable()
export class RedisMutexLocker implements ILocker {
private readonly logger = new Logger(RedisMutexLocker.name);
constructor(private readonly redis: SessionRedis) {}
async lock(owner: string, key: string): Promise<Lock> {
const lockKey = `MutexLock:${key}`;
this.logger.verbose(`Client ${owner} is trying to lock resource ${key}`);
const success = await this.redis.sendCommand(
new Command('EVAL', [lockScript, '1', lockKey, owner])
);
if (success === 1) {
return new Lock(async () => {
const result = await this.redis.sendCommand(
new Command('EVAL', [unlockScript, '1', lockKey, owner])
);
if (result === 0) {
throw new Error(`Failed to release lock ${key}`);
}
});
}
throw new Error(`Failed to acquire lock for resource [${key}]`);
}
}
packages/backend/server/src/plugins/redis/ws-adapter.ts
-28
View File
@@ -1,28 +0,0 @@
import { createAdapter } from '@socket.io/redis-adapter';
import { Redis } from 'ioredis';
import { Server, ServerOptions } from 'socket.io';
import { SocketIoAdapter } from '../../base/websocket';
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);
server.adapter(createAdapter(pubClient, subClient));
return server;
}
}
return RedisIoAdapter;
}