update message buffer

README.md

@@ -38,7 +38,8 @@ routes:
     heartbeat: 3000 # 心跳包的间隔。0 或不填为禁用心跳包。
     readonly: false # 该路由是否为只读。只读路由的连接无法对机器人进行写操作，只会得到模拟响应，但是可以进行 get 操作以及接收事件。
     rateLimitInterval: 500 # 限速调用间隔，默认 500ms。
-    bufferMessage: false # 是否在 app 断线期间缓存消息，并在 app 恢复连接时发送。
+    bufferAppMessage: false # 是否在 app 断线期间缓存消息，并在 app 恢复连接时发送。
+    bufferBotMessage: false # 是否在机器人断线期间缓存消息，并在机器人恢复连接时发送。
     wsReverse: # 该路由的反向 WebSocket 配置。可以配置多个。
       - endpoint: 'ws://localhost:8080'
         token: 'token'

config.example.yaml

@@ -25,7 +25,8 @@ routes:
     heartbeat: 3000 # 心跳包的间隔。0 或不填为禁用心跳包。
     readonly: false # 该路由是否为只读。只读路由的连接无法对机器人进行写操作，只会得到模拟响应，但是可以进行 get 操作以及接收事件。
     rateLimitInterval: 500 # 限速调用间隔，默认 500ms。
-    bufferMessage: false # 是否在 app 断线期间缓存消息，并在 app 恢复连接时发送。
+    bufferAppMessage: false # 是否在 app 断线期间缓存消息，并在 app 恢复连接时发送。
+    bufferBotMessage: false # 是否在机器人断线期间缓存消息，并在机器人恢复连接时发送。
     wsReverse: # 该路由的反向 WebSocket 配置。可以配置多个。
       - endpoint: 'ws://localhost:8080'
         token: 'token'

src/app.module.ts

@@ -10,6 +10,7 @@ import { ReverseWsService } from './reverse-ws/reverse-ws.service';
 import { WaitBotService } from './wait-bot/wait-bot.service';
 import { HealthService } from './health/health.service';
 import { HealthController } from './health/health.controller';
+import { BotRegistryService } from './bot-registry/bot-registry.service';
 
 @Module({
   imports: [
@@ -31,6 +32,7 @@ import { HealthController } from './health/health.controller';
     ReverseWsService,
     WaitBotService,
     HealthService,
+    BotRegistryService,
   ],
   controllers: [HealthController],
 })

src/bot-registry/bot-registry.service.spec.ts

+import { Test, TestingModule } from '@nestjs/testing';
+import { BotRegistryService } from './bot-registry.service';
+
+describe('BotRegistryService', () => {
+  let service: BotRegistryService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [BotRegistryService],
+    }).compile();
+
+    service = module.get<BotRegistryService>(BotRegistryService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+});

src/bot-registry/bot-registry.service.ts

+import { Injectable } from '@nestjs/common';
+import { WireContextService } from 'koishi-nestjs';
+import { OneBotBot } from '@koishijs/plugin-adapter-onebot/lib/bot';
+
+@Injectable()
+export class BotRegistryService {
+  @WireContextService('bots')
+  private bots: OneBotBot[];
+
+  private botMap = new Map<string, OneBotBot>();
+
+  getBotWithId(selfId: string) {
+    if (!this.botMap.has(selfId)) {
+      const bot = this.bots.find((bot) => bot.selfId === selfId);
+      if (bot) {
+        this.botMap.set(selfId, bot);
+      }
+    }
+    return this.botMap.get(selfId);
+  }
+
+  getAllBots() {
+    return this.bots;
+  }
+}

src/health/health.service.ts

@@ -3,12 +3,13 @@ import { RouteService } from '../route/route.service';
 import { InjectContext } from 'koishi-nestjs';
 import { Context } from 'koishi';
 import { HealthInfoDto } from '../dto/HealthInfo.dto';
+import { BotRegistryService } from '../bot-registry/bot-registry.service';
 
 @Injectable()
 export class HealthService {
   constructor(
     private readonly routeService: RouteService,
-    @InjectContext() private readonly ctx: Context,
+    private readonly botRegistry: BotRegistryService,
   ) {}
 
   healthOfAllRoutes() {
@@ -20,13 +21,13 @@ export class HealthService {
   }
 
   healthOfAllBots() {
-    return this.ctx.bots.map(
-      (b) => new HealthInfoDto(b.selfId, b.status === 'online'),
-    );
+    return this.botRegistry
+      .getAllBots()
+      .map((b) => new HealthInfoDto(b.selfId, b.status === 'online'));
   }
 
   healthOfBot(selfId: string) {
-    const bot = this.ctx.bots.find((b) => b.selfId === selfId);
+    const bot = this.botRegistry.getBotWithId(selfId);
     if (!bot) {
       return;
     }

src/message/message.service.ts

@@ -10,6 +10,7 @@ import {
 } from '../utility/onebot-protocol';
 import { OneBotBot } from '@koishijs/plugin-adapter-onebot/lib/bot';
 import { WaitBotService } from '../wait-bot/wait-bot.service';
+import { BotRegistryService } from '../bot-registry/bot-registry.service';
 
 export interface SendTask {
   bot: OneBotBot;
@@ -20,8 +21,8 @@ export interface SendTask {
 @Injectable()
 export class MessageService extends ConsoleLogger {
   constructor(
-    @InjectContext() private ctx: Context,
-    private waitBot: WaitBotService,
+    private readonly botRegistry: BotRegistryService,
+    private readonly waitBot: WaitBotService,
   ) {
     super('message');
   }
@@ -66,8 +67,28 @@ export class MessageService extends ConsoleLogger {
     client.send(JSON.stringify(genMetaEvent(route.selfId, 'enable')));
   }
 
+  private isRouteBotHealthy(route: Route): boolean {
+    const bot = this.botRegistry.getBotWithId(route.selfId);
+    return bot && bot.status === 'online';
+  }
+
   private async sendToBot(task: SendTask) {
-    await this.waitBot.waitForBotOnline(task.bot);
+    if (!this.isRouteBotHealthy(task.route)) {
+      if (task.route.bufferBotMessage) {
+        await this.waitBot.waitForBotOnline(task.bot);
+      } else {
+        return {
+          retcode: 1404,
+          status: 'failed',
+          data: null,
+          error: {
+            code: 1404,
+            message: `Bot ${task.route.selfId} from ${task.route.name} not online.`,
+          },
+          echo: task.data?.echo,
+        };
+      }
+    }
     try {
       const result = await task.bot.internal._request(
         task.data.action,
@@ -98,9 +119,7 @@ export class MessageService extends ConsoleLogger {
   }
 
   private async onWsEvent(route: Route, data: OnebotProtocol) {
-    const bot = this.ctx.bots.find(
-      (b) => b.selfId === route.selfId && b.platform === 'onebot',
-    ) as OneBotBot;
+    const bot = this.botRegistry.getBotWithId(route.selfId);
     if (!bot) {
       this.error(`Bot ${route.selfId} from ${route.name} not found.`);
       return {
@@ -153,6 +172,10 @@ export class MessageService extends ConsoleLogger {
   }
 
   private async resolveSendTaskOfRoute(route: Route) {
+    const bot = this.botRegistry.getBotWithId(route.selfId);
+    if (!bot || bot.status !== 'online') {
+      return;
+    }
     const task = route.fetchSendTask();
     if (!task) {
       return;

src/route/Route.ts

@@ -24,7 +24,8 @@ export interface RouteConfig {
   readonly?: boolean;
   rateLimitInterval?: number;
   reverseWs?: ReverseWsConfig[];
-  bufferMessage?: boolean;
+  bufferAppMessage?: boolean;
+  bufferBotMessage?: boolean;
 }
 export class Route implements RouteConfig {
   private connections: WebSocket[] = [];
@@ -40,7 +41,8 @@ export class Route implements RouteConfig {
   reverseWs?: ReverseWsConfig[];
   readonly?: boolean;
   rateLimitInterval: number;
-  bufferMessage?: boolean;
+  bufferAppMessage?: boolean;
+  bufferBotMessage?: boolean;
   preMessages: { data: any; session: Session }[] = [];
   constructor(routeConfig: RouteConfig, ctx: Context) {
     Object.assign(this, routeConfig);
@@ -68,7 +70,7 @@ export class Route implements RouteConfig {
   }
   send(data: any, session: Session, allConns = this.connections) {
     if (!allConns.length) {
-      if (this.bufferMessage) {
+      if (this.bufferAppMessage) {
         this.preMessages.push({ data, session });
       }
       return;
@@ -160,7 +162,7 @@ export class Route implements RouteConfig {
   }
   addConnection(conn: WebSocket) {
     this.connections.push(conn);
-    if (!this.bufferMessage) {
+    if (!this.bufferAppMessage) {
       return;
     }
     const preMessages = this.preMessages;