feat: 增加 QQBot 更新登录 SSE 进度流

This commit is contained in:
sunlei 2026-06-04 11:44:37 +08:00
parent fbaaee8bd7
commit 054c6ce23e
5 changed files with 390 additions and 34 deletions

View File

@ -6,6 +6,7 @@ import {
HttpStatus,
Post,
Query,
Sse,
UseGuards,
} from '@nestjs/common';
import { ApiOperation, ApiQuery, ApiTags } from '@nestjs/swagger';
@ -78,6 +79,12 @@ export class QqbotAccountController {
return vbenSuccess(await this.napcatLoginService.status(query.sessionId));
}
@Sse('scan/events')
@ApiOperation({ summary: '订阅 QQBot 扫码登录进度' })
scanEvents(@Query() query: QqbotAccountScanStatusDto) {
return this.napcatLoginService.events(query.sessionId);
}
@Post('scan/qrcode/refresh')
@HttpCode(HttpStatus.OK)
@ApiOperation({ summary: '刷新 QQBot 扫码二维码' })

View File

@ -3,6 +3,7 @@ import * as https from 'https';
import { createHash, randomUUID } from 'crypto';
import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { Observable } from 'rxjs';
import { throwVbenError, ToolsService } from '@/common';
import type {
NapcatApiResponse,
@ -12,6 +13,7 @@ import type {
NapcatQrcode,
NapcatRestartOptions,
QqbotLoginScanMode,
QqbotLoginScanEvent,
QqbotLoginScanResult,
QqbotLoginScanSession,
QqbotLoginScanStatus,
@ -25,6 +27,11 @@ import { QqbotAccountService } from './qqbot-account.service';
@Injectable()
export class QqbotNapcatLoginService {
private readonly sessions = new Map<string, QqbotLoginScanSession>();
private readonly sessionEventLogs = new Map<string, QqbotLoginScanEvent[]>();
private readonly sessionEventListeners = new Map<
string,
Set<(event: QqbotLoginScanEvent) => void>
>();
private readonly credentials = new Map<
string,
{ credential: string; expiresAt: number }
@ -67,6 +74,12 @@ export class QqbotNapcatLoginService {
if (session.status !== 'pending') {
return this.toResult(session);
}
if (session.preparingRelogin) {
return this.keepSessionPending(
session,
session.errorMessage || 'NapCat 正在重置登录态并生成二维码,请稍后',
);
}
const container = await this.getSessionContainer(session);
let loginStatus: NapcatLoginStatus;
@ -85,7 +98,19 @@ export class QqbotNapcatLoginService {
if (loginStatus.isOffline) {
if (session.mode === 'refresh') {
await this.resetNapcatForLogin(container, { waitForReady: false });
this.publishScanResultEvent(
session,
'relogin-reset-start',
'processing',
'开始重置 NapCat 登录态',
);
await this.resetNapcatForLogin(
container,
{ waitForReady: false },
(step, message) => {
this.publishScanResultEvent(session, step, 'processing', message);
},
);
} else {
await this.restartNapcatForLogin(container, { waitForReady: false });
}
@ -107,6 +132,12 @@ export class QqbotNapcatLoginService {
session.expiresAt = Date.now() + this.getSessionTtlMs();
session.errorMessage = undefined;
this.sessions.set(session.id, session);
this.publishScanResultEvent(
session,
'qrcode-ready',
'success',
'登录二维码已刷新',
);
return this.toResult(session);
} catch (err) {
if (!this.toolsService.isNapcatTemporaryError(err)) throw err;
@ -123,6 +154,9 @@ export class QqbotNapcatLoginService {
if (Date.now() > session.expiresAt) {
return this.expireSession(session);
}
if (session.preparingRelogin) {
return this.toResult(session);
}
const container = await this.getSessionContainer(session);
let status: NapcatLoginStatus;
@ -141,7 +175,16 @@ export class QqbotNapcatLoginService {
status.qrcodeurl &&
!this.toolsService.isNapcatExpiredQrcodeStatus(status)
) {
const qrcodeChanged = session.qrcode !== status.qrcodeurl;
session.qrcode = status.qrcodeurl;
if (qrcodeChanged) {
this.publishScanResultEvent(
session,
'qrcode-ready',
'success',
'登录二维码已生成',
);
}
} else if (status.isOffline) {
session.qrcode = undefined;
} else if (!this.toolsService.isNapcatExpiredQrcodeStatus(status)) {
@ -154,11 +197,40 @@ export class QqbotNapcatLoginService {
return this.completeLogin(session, container);
}
events(sessionId: string) {
this.getSession(sessionId);
return new Observable<{ data: QqbotLoginScanEvent }>((subscriber) => {
const listener = (event: QqbotLoginScanEvent) => {
subscriber.next({ data: event });
};
(this.sessionEventLogs.get(sessionId) || []).forEach(listener);
const listeners =
this.sessionEventListeners.get(sessionId) ||
new Set<(event: QqbotLoginScanEvent) => void>();
listeners.add(listener);
this.sessionEventListeners.set(sessionId, listeners);
return () => {
listeners.delete(listener);
if (listeners.size <= 0) {
this.sessionEventListeners.delete(sessionId);
}
};
});
}
async cancel(sessionId: string) {
const session = this.sessions.get(sessionId);
if (session) {
this.publishScanEvent(session, {
message: '扫码会话已取消',
result: this.toResult(session),
status: 'info',
step: 'session-cancelled',
});
this.sessions.delete(sessionId);
await this.cleanupSessionContainer(session);
this.cleanupSessionEvents(sessionId);
}
return true;
}
@ -174,18 +246,30 @@ export class QqbotNapcatLoginService {
): Promise<QqbotLoginScanResult> {
await this.cleanupSessions();
try {
if (options.forceRelogin) {
await this.resetNapcatForLogin(container);
}
if (options.forceRelogin) {
const session = this.createSession({
...options,
container,
preparingRelogin: true,
status: 'pending',
});
session.lastRestartedAt = Date.now();
session.errorMessage = 'NapCat 正在重置登录态并生成二维码,请稍后';
this.sessions.set(session.id, session);
this.publishScanResultEvent(
session,
'session-created',
'processing',
'已创建更新登录会话',
);
void this.prepareReloginQrcode(session, container);
return this.toResult(session);
}
try {
const loginStatus = await this.getLoginStatus(container, true);
if (loginStatus.isOffline) {
if (options.forceRelogin) {
await this.resetNapcatForLogin(container, { waitForReady: false });
} else {
await this.restartNapcatForLogin(container, { waitForReady: false });
}
await this.restartNapcatForLogin(container, { waitForReady: false });
const session = this.createSession({
...options,
container,
@ -196,17 +280,12 @@ export class QqbotNapcatLoginService {
loginStatus.loginError ||
'NapCat 账号已离线,已重新生成二维码';
this.sessions.set(session.id, session);
return this.toResult(session);
}
if (options.forceRelogin && loginStatus.isLogin) {
const session = this.createSession({
...options,
container,
status: 'pending',
});
session.errorMessage = 'NapCat 正在重新生成二维码,请稍后刷新';
this.sessions.set(session.id, session);
this.publishScanResultEvent(
session,
'container-restarted',
'processing',
session.errorMessage,
);
return this.toResult(session);
}
@ -233,6 +312,18 @@ export class QqbotNapcatLoginService {
status: 'pending',
});
this.sessions.set(session.id, session);
this.publishScanResultEvent(
session,
'qrcode-ready',
'success',
'登录二维码已生成',
);
this.publishScanResultEvent(
session,
'waiting-scan',
'processing',
'等待扫码确认',
);
return this.toResult(session);
} catch (err) {
const cleanupError = await this.cleanupRuntimeContainer(container);
@ -277,11 +368,18 @@ export class QqbotNapcatLoginService {
session.status = 'success';
session.errorMessage = undefined;
this.sessions.set(session.id, session);
return {
const result = {
...this.toResult(session),
accountId,
selfId,
};
this.publishScanEvent(session, {
message: '扫码登录成功',
result,
status: 'success',
step: 'login-success',
});
return result;
}
private createSession(input: {
@ -289,6 +387,7 @@ export class QqbotNapcatLoginService {
container: QqbotNapcatRuntime;
expectedSelfId?: string;
mode: QqbotLoginScanMode;
preparingRelogin?: boolean;
qrcode?: string;
status: QqbotLoginScanStatus;
}): QqbotLoginScanSession {
@ -302,6 +401,7 @@ export class QqbotNapcatLoginService {
expiresAt: now + this.getSessionTtlMs(),
id: randomUUID(),
mode: input.mode,
preparingRelogin: input.preparingRelogin,
qrcode: input.qrcode,
status: input.status,
webuiPort: input.container.webuiPort,
@ -323,6 +423,41 @@ export class QqbotNapcatLoginService {
};
}
private publishScanEvent(
session: QqbotLoginScanSession,
input: Omit<QqbotLoginScanEvent, 'createdAt'>,
) {
const event: QqbotLoginScanEvent = {
...input,
createdAt: Date.now(),
};
const logs = this.sessionEventLogs.get(session.id) || [];
logs.push(event);
this.sessionEventLogs.set(session.id, logs.slice(-50));
this.sessionEventListeners
.get(session.id)
?.forEach((listener) => listener(event));
}
private publishScanResultEvent(
session: QqbotLoginScanSession,
step: string,
status: QqbotLoginScanEvent['status'],
message: string,
) {
this.publishScanEvent(session, {
message,
result: this.toResult(session),
status,
step,
});
}
private cleanupSessionEvents(sessionId: string) {
this.sessionEventLogs.delete(sessionId);
this.sessionEventListeners.delete(sessionId);
}
private keepSessionPending(
session: QqbotLoginScanSession,
errorMessage: string,
@ -354,6 +489,7 @@ export class QqbotNapcatLoginService {
this.sessions.forEach((session, sessionId) => {
if (session.status !== 'pending' || now > session.expiresAt) {
this.sessions.delete(sessionId);
this.cleanupSessionEvents(sessionId);
expiredSessions.push(session);
}
});
@ -365,8 +501,15 @@ export class QqbotNapcatLoginService {
private async expireSession(session: QqbotLoginScanSession) {
session.status = 'expired';
session.errorMessage = session.errorMessage || '扫码会话已过期';
this.publishScanResultEvent(
session,
'session-expired',
'error',
session.errorMessage,
);
this.sessions.delete(session.id);
await this.cleanupSessionContainer(session);
this.cleanupSessionEvents(session.id);
return this.toResult(session);
}
@ -376,8 +519,10 @@ export class QqbotNapcatLoginService {
) {
session.status = 'error';
session.errorMessage = errorMessage;
this.publishScanResultEvent(session, 'login-error', 'error', errorMessage);
this.sessions.delete(session.id);
await this.cleanupSessionContainer(session);
this.cleanupSessionEvents(session.id);
return this.toResult(session);
}
@ -406,8 +551,17 @@ export class QqbotNapcatLoginService {
staleQrcode: session.qrcode || status.qrcodeurl,
});
if (qrcode) {
const qrcodeChanged = session.qrcode !== qrcode;
session.qrcode = qrcode;
session.errorMessage = status.loginError || undefined;
if (qrcodeChanged) {
this.publishScanResultEvent(
session,
'qrcode-ready',
'success',
'登录二维码已生成',
);
}
}
} catch (err) {
if (!this.toolsService.isNapcatTemporaryError(err)) throw err;
@ -613,13 +767,94 @@ export class QqbotNapcatLoginService {
return this.requestNapcat<T>(container, path, body, credential);
}
private async prepareReloginQrcode(
session: QqbotLoginScanSession,
container: QqbotNapcatRuntime,
) {
try {
this.publishScanResultEvent(
session,
'relogin-reset-start',
'processing',
'开始重置 NapCat 登录态',
);
await this.resetNapcatForLogin(
container,
{ waitForReady: false },
(step, message) => {
this.publishScanResultEvent(session, step, 'processing', message);
},
);
session.lastRestartedAt = Date.now();
this.publishScanResultEvent(
session,
'napcat-ready-wait',
'processing',
'等待 NapCat WebUI 启动',
);
await this.toolsService.sleep(this.getRestartDelayMs());
this.publishScanResultEvent(
session,
'qrcode-fetch',
'processing',
'正在获取登录二维码',
);
session.qrcode = await this.refreshOrGetQrcode(container, true, {
requireFresh: true,
});
session.errorMessage = undefined;
session.expiresAt = Date.now() + this.getSessionTtlMs();
this.publishScanResultEvent(
session,
'qrcode-ready',
'success',
'登录二维码已生成',
);
this.publishScanResultEvent(
session,
'waiting-scan',
'processing',
'等待扫码确认',
);
} catch (err) {
const message = this.toolsService.getErrorMessage(err);
if (this.toolsService.isNapcatTemporaryError(err)) {
session.errorMessage =
'NapCat 正在重新生成二维码,请稍后刷新或等待自动更新';
this.publishScanResultEvent(
session,
'qrcode-pending',
'processing',
session.errorMessage,
);
} else {
session.status = 'error';
session.errorMessage = message || 'NapCat 重置登录态失败';
this.publishScanResultEvent(
session,
'relogin-error',
'error',
session.errorMessage,
);
}
} finally {
const current = this.sessions.get(session.id);
if (current === session && current.status === 'pending') {
current.preparingRelogin = false;
this.sessions.set(current.id, current);
}
}
}
private async resetNapcatForLogin(
container: QqbotNapcatRuntime,
options: NapcatRestartOptions = {},
onProgress?: (step: string, message: string) => void,
) {
const resetByContainer =
await this.containerService.resetRuntimeLoginState(container);
await this.containerService.resetRuntimeLoginState(container, onProgress);
if (!resetByContainer) {
onProgress?.('napcat-restart-webui', '正在调用 NapCat 重启接口');
await this.restartNapcatForLogin(container, options);
return;
}
@ -627,6 +862,7 @@ export class QqbotNapcatLoginService {
this.credentials.delete(this.getCredentialCacheKey(container));
if (options.waitForReady === false) return;
onProgress?.('napcat-ready-wait', '等待 NapCat WebUI 启动');
await this.toolsService.sleep(this.getRestartDelayMs());
await this.getLoginStatus(container, true);
}

View File

@ -166,7 +166,10 @@ export class QqbotNapcatContainerService {
return true;
}
async resetRuntimeLoginState(runtime: QqbotNapcatRuntime) {
async resetRuntimeLoginState(
runtime: QqbotNapcatRuntime,
onProgress?: (step: string, message: string) => void,
) {
if (this.getManagedMode() !== 'ssh' || !runtime.id || !runtime.name) {
return false;
}
@ -182,7 +185,15 @@ export class QqbotNapcatContainerService {
}
const script = this.buildRemoteResetLoginStateScript(container);
await this.runProcess('ssh', [...this.getSshArgs(), 'sh -s'], script);
await this.runProcess(
'ssh',
[...this.getSshArgs(), 'sh -s'],
script,
(line) => {
const matched = line.match(/^__KT_PROGRESS__:([^:]+):(.+)$/);
if (matched) onProgress?.(matched[1], matched[2]);
},
);
await this.containerRepository.update(
{ id: runtime.id },
{
@ -275,10 +286,14 @@ case "$DATA_DIR" in
esac
docker exec "$NAME" rm -f /app/napcat/cache/qrcode.png >/dev/null 2>&1 || true
echo "__KT_PROGRESS__:container-stop:正在停止 NapCat 容器"
docker stop "$NAME" >/dev/null 2>&1 || true
echo "__KT_PROGRESS__:login-data-clean:正在清理旧 QQ 登录态"
mkdir -p "$DATA_DIR/QQ"
find "$DATA_DIR/QQ" -mindepth 1 -maxdepth 1 -exec rm -rf -- {} +
echo "__KT_PROGRESS__:container-start:正在启动 NapCat 容器"
docker start "$NAME" >/dev/null
echo "__KT_PROGRESS__:container-started:NapCat 容器已启动"
`;
}
@ -608,7 +623,12 @@ docker run -d \\
return `'${`${value}`.replace(/'/g, `'\\''`)}'`;
}
private runProcess(command: string, args: string[], input: string) {
private runProcess(
command: string,
args: string[],
input: string,
onStdoutLine?: (line: string) => void,
) {
return new Promise<void>((resolve, reject) => {
const child = spawn(command, args, {
windowsHide: true,
@ -616,6 +636,7 @@ docker run -d \\
let settled = false;
let stdout = '';
let stderr = '';
let stdoutLineBuffer = '';
const timeoutMs = this.getProcessTimeoutMs();
const timer = setTimeout(() => {
if (settled) return;
@ -630,7 +651,16 @@ docker run -d \\
callback();
};
child.stdout.on('data', (chunk) => {
stdout += Buffer.from(chunk).toString('utf8');
const text = Buffer.from(chunk).toString('utf8');
stdout += text;
if (onStdoutLine) {
const lines = `${stdoutLineBuffer}${text}`.split(/\r?\n/);
stdoutLineBuffer = lines.pop() || '';
lines
.map((line) => line.trim())
.filter(Boolean)
.forEach((line) => onStdoutLine(line));
}
});
child.stderr.on('data', (chunk) => {
stderr += Buffer.from(chunk).toString('utf8');
@ -640,6 +670,9 @@ docker run -d \\
});
child.on('close', (code) => {
finish(() => {
if (onStdoutLine && stdoutLineBuffer.trim()) {
onStdoutLine(stdoutLineBuffer.trim());
}
if (code === 0) {
resolve();
return;

View File

@ -246,6 +246,20 @@ export type QqbotLoginScanResult = {
webuiPort?: null | number;
};
export type QqbotLoginScanEventStatus =
| 'error'
| 'info'
| 'processing'
| 'success';
export type QqbotLoginScanEvent = {
createdAt: number;
message: string;
result?: QqbotLoginScanResult;
status: QqbotLoginScanEventStatus;
step: string;
};
export type QqbotLoginScanSession = {
accountId?: string;
containerId?: string;
@ -257,6 +271,7 @@ export type QqbotLoginScanSession = {
id: string;
lastRestartedAt?: number;
mode: QqbotLoginScanMode;
preparingRelogin?: boolean;
qrcode?: string;
status: QqbotLoginScanStatus;
webuiPort?: null | number;

View File

@ -24,6 +24,8 @@ describe('QqbotNapcatLoginService', () => {
beforeEach(() => {
(service as any).sessions.clear();
(service as any).sessionEventLogs.clear();
(service as any).sessionEventListeners.clear();
jest.restoreAllMocks();
});
@ -97,14 +99,20 @@ describe('QqbotNapcatLoginService', () => {
);
});
it('resets the existing container login state before refresh login scan', async () => {
it('returns refresh login scan immediately while resetting in background', async () => {
const container = {
baseUrl: 'http://127.0.0.1:6103/',
id: 'container-current',
name: 'napcat-10001',
};
let resolveReset!: () => void;
const containerService = {
resetRuntimeLoginState: jest.fn().mockResolvedValue(true),
resetRuntimeLoginState: jest.fn(
() =>
new Promise<boolean>((resolve) => {
resolveReset = () => resolve(true);
}),
),
};
const refreshService = new QqbotNapcatLoginService(
{ get: jest.fn() } as unknown as ConfigService,
@ -118,10 +126,6 @@ describe('QqbotNapcatLoginService', () => {
jest
.spyOn(refreshService as any, 'cleanupSessions')
.mockResolvedValue(undefined);
jest
.spyOn(refreshService as any, 'getLoginStatus')
.mockResolvedValueOnce({ isLogin: false })
.mockResolvedValueOnce({ isLogin: false });
jest
.spyOn(refreshService as any, 'refreshOrGetQrcode')
.mockResolvedValue('fresh-qrcode');
@ -137,10 +141,71 @@ describe('QqbotNapcatLoginService', () => {
);
expect(result.status).toBe('pending');
expect(result.qrcode).toBe('fresh-qrcode');
expect(result.qrcode).toBeUndefined();
expect(result.errorMessage).toBe('NapCat 正在重置登录态并生成二维码,请稍后');
expect(containerService.resetRuntimeLoginState).toHaveBeenCalledWith(
container,
expect.any(Function),
);
resolveReset();
});
it('does not complete refresh login from stale status while relogin is preparing', async () => {
(service as any).sessions.set('session-preparing', {
containerId: 'container-preparing',
containerName: 'napcat-preparing',
createdAt: Date.now(),
errorMessage: 'NapCat 正在重置登录态并生成二维码,请稍后',
expiresAt: Date.now() + 60_000,
id: 'session-preparing',
mode: 'refresh',
preparingRelogin: true,
status: 'pending',
webuiPort: 6106,
});
const getLoginStatus = jest.spyOn(service as any, 'getLoginStatus');
const result = await service.status('session-preparing');
expect(result.status).toBe('pending');
expect(result.errorMessage).toBe(
'NapCat 正在重置登录态并生成二维码,请稍后',
);
expect(getLoginStatus).not.toHaveBeenCalled();
});
it('replays scan progress events to late SSE subscribers', () => {
const session = {
containerId: 'container-events',
containerName: 'napcat-events',
createdAt: Date.now(),
expiresAt: Date.now() + 60_000,
id: 'session-events',
mode: 'refresh',
status: 'pending',
webuiPort: 6107,
};
(service as any).sessions.set('session-events', session);
(service as any).publishScanResultEvent(
session,
'container-stop',
'processing',
'正在停止 NapCat 容器',
);
const events: any[] = [];
const subscription = service
.events('session-events')
.subscribe((event) => events.push(event.data));
expect(events).toEqual([
expect.objectContaining({
message: '正在停止 NapCat 容器',
status: 'processing',
step: 'container-stop',
}),
]);
subscription.unsubscribe();
});
it('requires a qrcode different from the current one when refreshing qrcode', async () => {