refactor: replace polling with event-driven updates

Trigger account detection and group requests from observed changes. Remove native stage log forwarding and its lifecycle state, and simplify the README.
This commit is contained in:
手瓜一十雪
2026-10-19 09:56:20 +08:00
parent 1b74747251
commit e5cfec01bd
33 changed files with 1312 additions and 1956 deletions
+1 -2
View File
@@ -53,8 +53,7 @@ SnowLuma 是面向 QQ 客户端的 TypeScript 互操作运行时,将 QQ 原生
3. 打开 [`http://localhost:5099`](http://localhost:5099),使用启动日志中的初始密码登录 WebUI,按引导接入 QQ 进程并配置 OneBot 连接。
在「设置 → 关于 → 软件更新」下载新版本,再点「重启更新」。文件优先暂存于系统临时目录,不可用时使用安装目录的 `tmp/update`,下次启动时替换并清理,配置、日志与消息数据保留。更新沿用当前 Node.js;容器部署请更新镜像。
版本号遵循 [SemVer](https://semver.org/),在线更新仅安装正式版;构建元数据不影响版本排序。
账号识别由登录数据库的文件变化、进程和登录事件或手动操作触发,不做周期探测。未载入的 QQ 进程也不会持续调用 Windows Restart Manager;需要重查时可点击 WebUI 进程行的刷新/重连。
## 接入与文档
+59 -34
View File
@@ -14,13 +14,15 @@ import {
import { PipeWatcher } from './pipe-watcher';
import { createProcessEnumerator, type ProcessEnumerator } from './process-enumerator';
import { QqHookClient } from './qq-hook-client';
import { detectAccountLoginsForPids, type AccountLoginState } from './qq-account';
import {
detectAccountLoginsForPids,
watchAccountChangesForPids,
type AccountLoginState,
} from './qq-account';
import type { HookProcessAccount, HookProcessInfo } from './types';
const AUTO_LOAD_MAX_ATTEMPTS = 3;
const DEFAULT_ACCOUNT_REPROBE_INTERVAL_MS = 5000;
type AutoLoadAttempt = {
attempts: number;
inFlight: boolean;
@@ -61,9 +63,8 @@ export type HookManagerDeps = {
* the HOME it actually runs under (a container can give every QQ instance
* its own). Defaults to reading QQ's own account data. */
detectAccounts?: (pids: readonly number[]) => Promise<AccountLoginState[]>;
/** How often the watcher re-probes the account label of each process, in
* milliseconds. Defaults to `DEFAULT_ACCOUNT_REPROBE_INTERVAL_MS`. */
accountReprobeIntervalMs?: number;
/** Subscribe to login database changes. Returns the subscription cleanup. */
watchAccounts?: (pids: readonly number[], onChange: () => void) => () => void;
/** When true, every newly-discovered QQ process is auto-injected (fires
* `loadProcess(pid)` from the watcher's 'process-discovered' handler).
* Failed loads are logged and leave the session in the 'error' state.
@@ -106,7 +107,7 @@ export class HookManager {
private readonly ownsPipeWatcher: boolean;
private readonly enumerator: ProcessEnumerator;
private readonly detectAccounts: (pids: readonly number[]) => Promise<AccountLoginState[]>;
private readonly accountReprobeIntervalMs: number;
private readonly watchAccounts: NonNullable<HookManagerDeps['watchAccounts']>;
private readonly autoLoadOnDiscovery: boolean;
private readonly onSessionsChangedRaw?: () => void;
private readonly log: Logger;
@@ -116,7 +117,8 @@ export class HookManager {
private accounts = new Map<number, HookProcessAccount>();
private accountProbe: Promise<void> | null = null;
private nextAccountProbeAt = 0;
private accountsDirty = true;
private stopWatchingAccounts?: () => void;
private lastAccountProbeFailure = '';
private disposed = false;
@@ -133,8 +135,9 @@ export class HookManager {
};
this.makeClient = deps.makeClient ?? ((pid: number) => new QqHookClient(pid));
this.detectAccounts = deps.detectAccounts ?? detectAccountLoginsForPids;
this.accountReprobeIntervalMs =
deps.accountReprobeIntervalMs ?? DEFAULT_ACCOUNT_REPROBE_INTERVAL_MS;
this.watchAccounts =
deps.watchAccounts ??
((pids, onChange) => watchAccountChangesForPids(pids, onChange, this.log));
this.autoLoadOnDiscovery = deps.autoLoadOnDiscovery ?? false;
this.enumerator = createProcessEnumerator({
@@ -186,6 +189,7 @@ export class HookManager {
await this.startPromise;
this.autoLoadAttempts.delete(pid);
const session = this.ensureSession(pid);
this.invalidateAccounts();
await session.load();
this.pipeWatcher.wake();
await this.refreshAccounts();
@@ -197,6 +201,7 @@ export class HookManager {
await this.startPromise;
this.autoLoadAttempts.delete(pid);
const session = this.ensureSession(pid);
this.invalidateAccounts();
await session.unload();
await this.refreshAccounts();
return this.processInfo(session);
@@ -206,6 +211,8 @@ export class HookManager {
this.assertValidPid(pid);
await this.startPromise;
const session = this.ensureSession(pid);
this.updateAccountWatcher();
this.invalidateAccounts();
await session.refresh();
await this.refreshAccounts();
return this.processInfo(session);
@@ -217,6 +224,7 @@ export class HookManager {
}
this.disposed = true;
this.stopWatchingAccounts?.();
for (const session of this.sessions.values()) {
session.dispose();
@@ -238,45 +246,52 @@ export class HookManager {
return { ...session.toInfo(), account: this.accounts.get(session.pid) ?? null };
}
private invalidateAccounts(): void {
this.accountsDirty = true;
}
private updateAccountWatcher(): void {
this.stopWatchingAccounts?.();
this.stopWatchingAccounts = this.watchAccounts([...this.sessions.keys()], () => {
this.invalidateAccounts();
void this.refreshAccounts();
});
}
/** REST, watcher ticks and login reconciliation share the same account snapshot. */
private async refreshAccounts(): Promise<void> {
if (this.disposed) {
if (this.disposed || this.sessions.size === 0) {
return;
}
if (this.accountProbe) {
await this.accountProbe;
// Process discovery invalidates the snapshot while a read is in flight.
if (this.nextAccountProbeAt === 0) {
await this.refreshAccounts();
}
return;
}
if (Date.now() < this.nextAccountProbeAt || this.sessions.size === 0) {
if (!this.accountsDirty) {
return;
}
this.nextAccountProbeAt = Date.now() + this.accountReprobeIntervalMs;
this.accountProbe = runWithTraceRequest(() => this.readAccounts()).finally(() => {
this.accountProbe = runWithTraceRequest(async () => {
// All callers await the replacement read too if a lifecycle change
// invalidates an in-flight result.
do {
this.accountsDirty = false;
await this.readAccounts();
} while (!this.disposed && this.sessions.size > 0 && this.accountsDirty);
}).finally(() => {
this.accountProbe = null;
});
await this.accountProbe;
}
private async readAccounts(): Promise<void> {
const sessions = new Map(this.sessions);
const accounts = new Map<number, HookProcessAccount>();
try {
for (const state of await this.detectAccounts([...sessions.keys()])) {
if (
state.pid !== null &&
sessions.has(state.pid) &&
this.sessions.get(state.pid) === sessions.get(state.pid)
) {
for (const state of await this.detectAccounts([...this.sessions.keys()])) {
if (state.pid !== null && this.sessions.has(state.pid)) {
accounts.set(state.pid, { uin: state.uin, nickname: state.nickname });
}
}
@@ -291,7 +306,7 @@ export class HookManager {
}
}
if (this.disposed) {
if (this.disposed || this.accountsDirty) {
return;
}
@@ -364,6 +379,7 @@ export class HookManager {
const session = this.sessions.get(pid);
if (session) {
this.invalidateAccounts();
session.onPipeUp();
}
});
@@ -375,6 +391,7 @@ export class HookManager {
const session = this.sessions.get(pid);
if (session) {
this.invalidateAccounts();
session.onPipeDown();
}
});
@@ -570,7 +587,10 @@ export class HookManager {
session.attachProcessInfo({ name: defaultProcessName() });
session.on('login', (uin: string, sender) => {
// Account switches can keep the same 'online' status.
this.invalidateAccounts();
this.bridgeManager.onHookLogin(pid, uin, sender);
this.notifySessionsChanged();
});
session.on('disconnected', (wasLoggedIn: boolean) => {
if (wasLoggedIn) {
@@ -580,20 +600,25 @@ export class HookManager {
session.on('receive-health-changed', (healthy: boolean) => {
this.bridgeManager.onPidReceiveHealthChanged(pid, healthy);
});
// status-changed fires on EVERY internal status mutation, including the
// ones reached via login / disconnected / refresh — subscribing here
// alone covers every transition the WebUI processes view cares about.
session.on('status-changed', () => {
let previousStatus = session.status;
session.on('status-changed', (status) => {
if (previousStatus === 'online' && status !== 'online') {
this.invalidateAccounts();
}
previousStatus = status;
this.notifySessionsChanged();
});
session.on('disposed', () => {
this.sessions.delete(pid);
this.accounts.delete(pid);
this.nextAccountProbeAt = 0;
this.updateAccountWatcher();
this.invalidateAccounts();
});
this.sessions.set(pid, session);
this.nextAccountProbeAt = 0;
this.updateAccountWatcher();
this.invalidateAccounts();
return session;
}
+1
View File
@@ -12,6 +12,7 @@ export {
type LoginDbFailure,
} from './login-db';
export { accountDirName, loginDbPaths, ntMsgDbPath, qqDataRoots, qqDataRootsForPid } from './paths';
export { watchAccountChangesForPids } from './watch';
export {
decryptDatabase,
detectAlgorithm,
+115
View File
@@ -0,0 +1,115 @@
import { errorMessage } from '@snowluma/common/errors';
import { createLogger, type Logger } from '@snowluma/common/logger';
import { watch, type FSWatcher } from 'node:fs';
import path from 'node:path';
import { loginDbPaths, qqDataRootsForPid } from './paths';
export function watchAccountChangesForPids(
pids: readonly number[],
onChange: () => void,
log: Logger = createLogger('QqAccount'),
): () => void {
const roots = new Set<string>();
for (const pid of pids) {
try {
for (const root of qqDataRootsForPid(pid)) {
roots.add(root);
}
} catch (error) {
log.warn('account watch root failed: PID=%d err=%s', pid, errorMessage(error));
}
}
return watchAccountFiles([...roots].flatMap(loginDbPaths), onChange, log);
}
/** Watch directories so replacing login.db or recreating its WAL keeps working. */
export function watchAccountFiles(
dbPaths: readonly string[],
onChange: () => void,
log: Logger = createLogger('QqAccount'),
): () => void {
const watchers = new Map<string, FSWatcher>();
let timer: NodeJS.Timeout | undefined;
const changed = () => {
clearTimeout(timer);
timer = setTimeout(() => {
timer = undefined;
onChange();
}, 250);
timer.unref();
};
const observe = (dbPath: string) => {
watchers.get(dbPath)?.close();
watchers.delete(dbPath);
const dbDirectory = path.dirname(dbPath);
let directory = dbDirectory;
for (;;) {
const target = path.relative(directory, dbPath).split(path.sep)[0]!;
try {
const watcher = watch(directory, { persistent: false }, (_event, filename) => {
if (
watchers.get(dbPath) !== watcher ||
(filename !== null &&
filename !== target &&
!(directory === dbDirectory && filename === `${target}-wal`))
) {
return;
}
// QQ may create its data directories after process discovery.
if (directory !== dbDirectory) {
observe(dbPath);
}
changed();
});
watcher.on('error', (error) => {
if (watchers.get(dbPath) !== watcher) {
return;
}
watcher.close();
watchers.delete(dbPath);
log.warn('account watch failed: path=%s err=%s', dbPath, errorMessage(error));
});
watchers.set(dbPath, watcher);
return;
} catch (error) {
const parent = path.dirname(directory);
if (
error instanceof Error &&
'code' in error &&
error.code === 'ENOENT' &&
parent !== directory
) {
directory = parent;
continue;
}
log.warn('account watch failed: path=%s err=%s', dbPath, errorMessage(error));
return;
}
}
};
for (const dbPath of new Set(dbPaths)) {
observe(dbPath);
}
return () => {
clearTimeout(timer);
for (const watcher of watchers.values()) {
watcher.close();
}
watchers.clear();
};
}
@@ -24,6 +24,8 @@ interface Harness {
setLivePipes: (next: number[]) => void;
onSessionsChanged: ReturnType<typeof vi.fn>;
clientFactory: ReturnType<typeof vi.fn>;
notifyAccountChange: () => void;
watchAccounts: ReturnType<typeof vi.fn>;
}
function makeManager(
@@ -31,7 +33,6 @@ function makeManager(
processes?: number[];
livePipes?: number[];
detectAccounts?: (pids: readonly number[]) => Promise<AccountLoginState[]>;
accountReprobeIntervalMs?: number;
} = {},
): Harness {
let pids = opts.processes ?? [];
@@ -64,6 +65,13 @@ function makeManager(
onPidReceiveHealthChanged: vi.fn(),
} as unknown as BridgeManagerSink & { onPidReceiveHealthChanged: ReturnType<typeof vi.fn> };
const onSessionsChanged = vi.fn();
let accountChanged = () => {};
const watchAccounts = vi.fn((_pids: readonly number[], onChange: () => void) => {
accountChanged = onChange;
return vi.fn();
});
const manager = new HookManager({
bridgeManager,
pipeWatcher,
@@ -77,7 +85,7 @@ function makeManager(
makeClient: clientFactory,
listProcesses: async () => pids.map((pid) => ({ pid, name: 'qq', path: '' })),
detectAccounts: opts.detectAccounts ?? (async () => []),
accountReprobeIntervalMs: opts.accountReprobeIntervalMs,
watchAccounts,
onSessionsChanged,
});
return {
@@ -86,6 +94,8 @@ function makeManager(
bridgeManager,
onSessionsChanged,
clientFactory,
notifyAccountChange: () => accountChanged(),
watchAccounts,
setProcesses: (next) => {
pids = next;
},
@@ -198,6 +208,7 @@ describe('HookManager.onSessionsChanged', () => {
}),
listProcesses: async () => pids.map((pid) => ({ pid, name: 'qq', path: '' })),
detectAccounts: async () => [],
watchAccounts: () => () => {},
// onSessionsChanged deliberately omitted
});
await pipeWatcher.start();
@@ -210,7 +221,111 @@ describe('HookManager.onSessionsChanged', () => {
});
});
describe('HookManager periodic account re-probe', () => {
describe('HookManager account re-probe', () => {
it('reuses the account snapshot across ticks and REST reads while all sessions are online (#549)', async () => {
const detectAccounts = vi.fn(async (): Promise<AccountLoginState[]> => [
{ uin: '10001', uid: 'u_a', nickname: 'Alice', pid: 4242 },
{ uin: '10002', uid: 'u_b', nickname: 'Bob', pid: 5151 },
]);
const ctx = makeManager({
processes: [4242, 5151],
livePipes: [4242, 5151],
detectAccounts,
});
const now = vi.spyOn(Date, 'now');
try {
await ctx.pipeWatcher.start();
await flush();
for (const [index, uin] of ['10001', '10002'].entries()) {
const client = ctx.clientFactory.mock.results[index]!.value as EventEmitter;
client.emit('loginState', { loggedIn: true, uin, uinNumber: BigInt(uin) });
}
const initial = await ctx.manager.listProcesses();
expect(initial.map((info) => info.status)).toEqual(['online', 'online']);
const baseline = detectAccounts.mock.calls.length;
ctx.onSessionsChanged.mockClear();
const startedAt = Date.now();
for (let tick = 1; tick <= 10; tick++) {
now.mockReturnValue(startedAt + tick * 6000);
await ctx.pipeWatcher.tickNow();
expect(await ctx.manager.listProcesses()).toEqual(initial);
}
expect(detectAccounts).toHaveBeenCalledTimes(baseline);
expect(ctx.onSessionsChanged).not.toHaveBeenCalled();
} finally {
now.mockRestore();
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
});
it.each(['loadProcess', 'unloadProcess', 'refreshProcess'] as const)(
'refreshes account labels on %s even within the probe interval',
async (operation) => {
let nickname = 'Alice';
const ctx = makeManager({
processes: [4242],
detectAccounts: async () => [{ uin: '10001', uid: 'u_a', nickname, pid: 4242 }],
});
try {
expect((await ctx.manager.listProcesses())[0]!.account?.nickname).toBe('Alice');
nickname = 'Updated';
expect((await ctx.manager[operation](4242)).account?.nickname).toBe('Updated');
} finally {
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
},
);
it('discards an in-flight account result when an online process switches accounts', async () => {
const stale = Promise.withResolvers<AccountLoginState[]>();
const fresh = Promise.withResolvers<AccountLoginState[]>();
const detectAccounts = vi.fn(async (): Promise<AccountLoginState[]> => [
{ uin: '10001', uid: 'u_a', nickname: 'Alice', pid: 4242 },
]);
const ctx = makeManager({
processes: [4242],
livePipes: [4242],
detectAccounts,
});
try {
await ctx.pipeWatcher.start();
await flush();
const client = ctx.clientFactory.mock.results[0]!.value as EventEmitter;
client.emit('loginState', { loggedIn: true, uin: '10001', uinNumber: 10001n });
await ctx.manager.listProcesses();
detectAccounts.mockImplementationOnce(() => stale.promise);
detectAccounts.mockImplementationOnce(() => fresh.promise);
const refreshing = ctx.manager.refreshProcess(4242);
await flush();
const baseline = detectAccounts.mock.calls.length;
client.emit('loginState', { loggedIn: true, uin: '10002', uinNumber: 10002n });
const snapshot = ctx.manager.listProcesses();
await flush();
ctx.onSessionsChanged.mockClear();
stale.resolve([{ uin: '10003', uid: 'u_c', nickname: 'Stale', pid: 4242 }]);
await flush();
expect(ctx.onSessionsChanged).not.toHaveBeenCalled();
expect(detectAccounts).toHaveBeenCalledTimes(baseline + 1);
fresh.resolve([{ uin: '10002', uid: 'u_b', nickname: 'Bob', pid: 4242 }]);
expect((await snapshot)[0]!.account).toEqual({ uin: '10002', nickname: 'Bob' });
expect((await refreshing).account).toEqual({ uin: '10002', nickname: 'Bob' });
} finally {
stale.resolve([]);
fresh.resolve([]);
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
});
it('shares one account probe between watcher ticks and concurrent snapshots', async () => {
const probe = Promise.withResolvers<AccountLoginState[]>();
const detectAccounts = vi.fn(() => probe.promise);
@@ -234,6 +349,86 @@ describe('HookManager periodic account re-probe', () => {
}
});
it.each(['pipe-down', 'client-close', 'logout'] as const)(
're-probes after %s and observes later login data changes',
async (event) => {
let accounts: AccountLoginState[] = [
{ uin: '10001', uid: 'u_a', nickname: 'Alice', pid: 4242 },
];
const detectAccounts = vi.fn(async () => accounts);
const ctx = makeManager({
processes: [4242],
livePipes: [4242],
detectAccounts,
});
const now = vi.spyOn(Date, 'now');
try {
await ctx.pipeWatcher.start();
await flush();
const client = ctx.clientFactory.mock.results[0]!.value as EventEmitter;
client.emit('loginState', { loggedIn: true, uin: '10001', uinNumber: 10001n });
await ctx.manager.listProcesses();
accounts = [];
if (event === 'pipe-down') {
ctx.setLivePipes([]);
await ctx.pipeWatcher.tickNow();
} else if (event === 'client-close') {
client.emit('close');
} else {
client.emit('loginState', { loggedIn: false, uin: '0', uinNumber: 0n });
}
await flush();
expect((await ctx.manager.listProcesses())[0]!.account).toBeNull();
accounts = [{ uin: '10002', uid: 'u_b', nickname: 'Bob', pid: 4242 }];
now.mockReturnValue(Date.now() + 6000);
ctx.notifyAccountChange();
await ctx.pipeWatcher.tickNow();
expect((await ctx.manager.listProcesses())[0]!.account?.uin).toBe('10002');
} finally {
now.mockRestore();
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
},
);
it('re-probes when another process appears and when it exits', async () => {
let nickname = 'Alice';
const detectAccounts = vi.fn(async (): Promise<AccountLoginState[]> => [
{ uin: '10001', uid: 'u_a', nickname, pid: 4242 },
{ uin: '10002', uid: 'u_b', nickname: 'Bob', pid: 5151 },
]);
const ctx = makeManager({ processes: [4242], livePipes: [4242], detectAccounts });
try {
await ctx.pipeWatcher.start();
await flush();
const client = ctx.clientFactory.mock.results[0]!.value as EventEmitter;
client.emit('loginState', { loggedIn: true, uin: '10001', uinNumber: 10001n });
await ctx.manager.listProcesses();
nickname = 'Updated';
ctx.setProcesses([4242, 5151]);
await ctx.pipeWatcher.tickNow();
expect((await ctx.manager.listProcesses()).map((info) => info.account)).toEqual([
{ uin: '10001', nickname: 'Updated' },
{ uin: '10002', nickname: 'Bob' },
]);
nickname = 'Final';
ctx.setProcesses([4242]);
await ctx.pipeWatcher.tickNow();
await flush();
const list = await ctx.manager.listProcesses();
expect(list).toHaveLength(1);
expect(list[0]!.account?.nickname).toBe('Final');
} finally {
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
});
it('keeps all process labels when an operation reads just one process', async () => {
const accounts: AccountLoginState[] = [
{ uin: '10001', uid: 'u_a', nickname: 'Alice', pid: 4242 },
@@ -246,8 +441,8 @@ describe('HookManager periodic account re-probe', () => {
try {
await ctx.manager.listProcesses();
const baseline = detectAccounts.mock.calls.length;
expect((await ctx.manager.refreshProcess(4242)).account?.uin).toBe('10001');
const baseline = detectAccounts.mock.calls.length;
ctx.onSessionsChanged.mockClear();
await ctx.pipeWatcher.tickNow();
await flush();
@@ -282,17 +477,13 @@ describe('HookManager periodic account re-probe', () => {
}
});
// A login can happen with no hook edge at all (QQ signs in after SnowLuma
// started, or before the hook was injected), so no session status changes
// and nothing pushes a fresh processes snapshot. These cover the watcher
// pass that re-reads the account labels and invalidates when they move.
// Login database changes can reveal login before the hook is loaded.
it('publishes when a process gains an account label', async () => {
let accounts: AccountLoginState[] = [{ uin: '1', uid: 'u1', nickname: 'A', pid: null }];
const ctx = makeManager({
processes: [4242],
detectAccounts: async () => accounts,
accountReprobeIntervalMs: 0,
});
await ctx.pipeWatcher.start();
await flush();
@@ -300,7 +491,7 @@ describe('HookManager periodic account re-probe', () => {
// QQ finished signing in: the account's nt_msg.db is now held by 4242.
accounts = [{ uin: '1', uid: 'u1', nickname: 'A', pid: 4242 }];
await ctx.pipeWatcher.tickNow();
ctx.notifyAccountChange();
await flush();
await flush();
@@ -317,14 +508,12 @@ describe('HookManager periodic account re-probe', () => {
const ctx = makeManager({
processes: [4242],
detectAccounts,
accountReprobeIntervalMs: 0,
});
await ctx.pipeWatcher.start();
await flush();
ctx.onSessionsChanged.mockClear();
await ctx.pipeWatcher.tickNow();
await ctx.pipeWatcher.tickNow();
ctx.notifyAccountChange();
await flush();
await flush();
@@ -340,7 +529,6 @@ describe('HookManager periodic account re-probe', () => {
const ctx = makeManager({
processes: [4242],
detectAccounts: async () => accounts,
accountReprobeIntervalMs: 0,
});
await ctx.pipeWatcher.start();
await flush();
@@ -352,7 +540,7 @@ describe('HookManager periodic account re-probe', () => {
{ uin: '1', uid: 'u1', nickname: 'A', pid: null },
{ uin: '2', uid: 'u2', nickname: 'B', pid: null },
];
await ctx.pipeWatcher.tickNow();
ctx.notifyAccountChange();
await flush();
await flush();
@@ -361,24 +549,89 @@ describe('HookManager periodic account re-probe', () => {
ctx.manager.dispose();
});
it('throttles the probe to the configured interval', async () => {
it('does not poll unlabelled, unloaded processes even after the old interval expires', async () => {
const detectAccounts = vi.fn(async () => [] as AccountLoginState[]);
const ctx = makeManager({
processes: [4242],
detectAccounts,
accountReprobeIntervalMs: 60_000,
});
await ctx.pipeWatcher.start();
await flush();
const baseline = detectAccounts.mock.calls.length;
const now = vi.spyOn(Date, 'now');
await ctx.pipeWatcher.tickNow();
await ctx.pipeWatcher.tickNow();
await flush();
try {
await ctx.manager.listProcesses();
const baseline = detectAccounts.mock.calls.length;
const startedAt = Date.now();
expect(detectAccounts.mock.calls.length).toBe(baseline);
for (let tick = 1; tick <= 10; tick++) {
now.mockReturnValue(startedAt + tick * 6000);
await ctx.pipeWatcher.tickNow();
expect((await ctx.manager.listProcesses())[0]!.account).toBeNull();
}
ctx.manager.dispose();
expect(detectAccounts).toHaveBeenCalledTimes(baseline);
} finally {
now.mockRestore();
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
});
it('does not re-probe a mixed set of online and unloaded processes', async () => {
const detectAccounts = vi.fn(async (): Promise<AccountLoginState[]> => [
{ uin: '10001', uid: 'u_a', nickname: 'Alice', pid: 4242 },
]);
const ctx = makeManager({
processes: [4242, 5151],
livePipes: [4242],
detectAccounts,
});
const now = vi.spyOn(Date, 'now');
try {
await ctx.pipeWatcher.start();
await flush();
const client = ctx.clientFactory.mock.results[0]!.value as EventEmitter;
client.emit('loginState', { loggedIn: true, uin: '10001', uinNumber: 10001n });
expect((await ctx.manager.listProcesses()).map((info) => info.status)).toEqual([
'online',
'available',
]);
const baseline = detectAccounts.mock.calls.length;
now.mockReturnValue(Date.now() + 60_000);
await ctx.pipeWatcher.tickNow();
await ctx.manager.listProcesses();
expect(detectAccounts).toHaveBeenCalledTimes(baseline);
} finally {
now.mockRestore();
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
});
it('releases file subscriptions when the process set changes and the manager stops', async () => {
const ctx = makeManager({ processes: [4242] });
try {
await ctx.pipeWatcher.start();
expect(ctx.watchAccounts).toHaveBeenLastCalledWith([4242], expect.any(Function));
const firstStop = ctx.watchAccounts.mock.results.at(-1)!.value;
ctx.setProcesses([4242, 5151]);
await ctx.pipeWatcher.tickNow();
expect(firstStop).toHaveBeenCalledOnce();
expect(ctx.watchAccounts).toHaveBeenLastCalledWith([4242, 5151], expect.any(Function));
const secondStop = ctx.watchAccounts.mock.results.at(-1)!.value;
ctx.setProcesses([5151]);
await ctx.pipeWatcher.tickNow();
await flush();
expect(secondStop).toHaveBeenCalledOnce();
expect(ctx.watchAccounts).toHaveBeenLastCalledWith([5151], expect.any(Function));
const finalStop = ctx.watchAccounts.mock.results.at(-1)!.value;
ctx.manager.dispose();
expect(finalStop).toHaveBeenCalledOnce();
} finally {
ctx.manager.dispose();
ctx.pipeWatcher.dispose();
}
});
it('keeps working when detection fails, and reports the failure once', async () => {
@@ -387,7 +640,6 @@ describe('HookManager periodic account re-probe', () => {
detectAccounts: async () => {
throw new Error('login.db unreadable');
},
accountReprobeIntervalMs: 0,
});
await ctx.pipeWatcher.start();
await flush();
@@ -80,6 +80,7 @@ function makeManager(opts: {
listProcesses: async () => pids.map((pid) => ({ pid, name: 'qq', path: '' })),
// Detection reads QQ's own data directories; tests stay off the host.
detectAccounts: opts.detectAccounts ?? (async () => []),
watchAccounts: () => () => {},
});
return {
manager,
@@ -72,6 +72,7 @@ describe('HookManager — Docker reconnect-without-login leak guard', () => {
autoLoadOnDiscovery: true,
listProcesses: async () => [{ pid: PID, name: 'qq', path: '' }],
detectAccounts: async () => [],
watchAccounts: () => () => {},
});
// Initial discovery + first connect.
@@ -0,0 +1,119 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import * as fs from 'node:fs';
import { mkdtemp, mkdir, rename, rm, writeFile } from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { setTimeout as delay } from 'node:timers/promises';
import { createLogger } from '@snowluma/common/logger';
import { watchAccountFiles } from '../src/qq-account/watch';
vi.mock('node:fs', async (importOriginal) => {
const actual = await importOriginal<typeof fs>();
return { ...actual, watch: vi.fn(actual.watch) };
});
let root: string;
const subscriptions: Array<() => void> = [];
beforeEach(async () => {
root = await mkdtemp(path.join(os.tmpdir(), 'snowluma-account-watch-'));
});
afterEach(async () => {
for (const stop of subscriptions.splice(0)) {
stop();
}
await rm(root, { recursive: true, force: true });
vi.restoreAllMocks();
vi.clearAllMocks();
});
function subscribe(dbPaths = [path.join(root, 'login.db')]) {
const changed = vi.fn();
const log = createLogger('AccountWatchTest');
const warning = vi.spyOn(log, 'warn');
const stop = watchAccountFiles(dbPaths, changed, log);
subscriptions.push(stop);
return { changed, stop, warning };
}
describe('account file changes', () => {
it('coalesces login database and WAL changes while ignoring message data and SHM', async () => {
const dbPath = path.join(root, 'login.db');
const { changed } = subscribe([dbPath, dbPath]);
expect(fs.watch).toHaveBeenCalledTimes(1);
await writeFile(dbPath, 'login');
await writeFile(`${dbPath}-wal`, 'wal');
await vi.waitFor(() => expect(changed).toHaveBeenCalledTimes(1));
await writeFile(path.join(root, 'nt_msg.db'), 'message');
await writeFile(`${dbPath}-shm`, 'shared memory');
await delay(350);
expect(changed).toHaveBeenCalledTimes(1);
});
it('observes database replacement and WAL recreation', async () => {
const dbPath = path.join(root, 'login.db');
await writeFile(dbPath, 'initial');
const { changed } = subscribe();
await writeFile(path.join(root, 'replacement'), 'replacement');
await rename(path.join(root, 'replacement'), dbPath);
await vi.waitFor(() => expect(changed).toHaveBeenCalledTimes(1));
await writeFile(dbPath, 'updated');
await vi.waitFor(() => expect(changed).toHaveBeenCalledTimes(2));
await writeFile(`${dbPath}-wal`, 'wal');
await vi.waitFor(() => expect(changed).toHaveBeenCalledTimes(3));
await rm(`${dbPath}-wal`);
await writeFile(`${dbPath}-wal`, 'new wal');
await vi.waitFor(() => expect(changed).toHaveBeenCalledTimes(4));
});
it('starts watching login data created after QQ process discovery', async () => {
const dbPath = path.join(root, 'QQ', 'global', 'nt_db', 'login.db');
const { changed } = subscribe([dbPath]);
await mkdir(path.dirname(dbPath), { recursive: true });
await writeFile(dbPath, 'initial');
await vi.waitFor(() => expect(changed).toHaveBeenCalledTimes(1));
await writeFile(`${dbPath}-wal`, 'login');
await vi.waitFor(() => expect(changed).toHaveBeenCalledTimes(2));
});
it('closes watchers and cancels pending notifications on disposal', async () => {
const { changed, stop } = subscribe();
await writeFile(path.join(root, 'login.db'), 'initial');
await delay(30);
stop();
await writeFile(path.join(root, 'login.db'), 'later');
await delay(350);
expect(changed).not.toHaveBeenCalled();
});
it('reports watch failures without starting a polling fallback', async () => {
vi.mocked(fs.watch).mockImplementationOnce(() => {
throw Object.assign(new Error('access denied'), { code: 'EACCES' });
});
const { changed, warning } = subscribe();
expect(warning).toHaveBeenCalledWith(
'account watch failed: path=%s err=%s',
path.join(root, 'login.db'),
'access denied',
);
await delay(350);
expect(fs.watch).toHaveBeenCalledTimes(1);
expect(changed).not.toHaveBeenCalled();
});
it('reports asynchronous watch errors and closes the failed watcher', () => {
const { changed, warning } = subscribe();
const watcher = vi.mocked(fs.watch).mock.results[0]!.value;
const close = vi.spyOn(watcher, 'close');
watcher.emit('error', new Error('watch unavailable'));
expect(close).toHaveBeenCalledOnce();
expect(warning).toHaveBeenCalledWith(
'account watch failed: path=%s err=%s',
path.join(root, 'login.db'),
'watch unavailable',
);
expect(changed).not.toHaveBeenCalled();
});
});
+41 -4
View File
@@ -581,11 +581,48 @@ export class ContactsApi {
return requests;
}
/** Live observation write. IncomingPacketPipeline and tests only. */
rememberGroupInviteCardSequence(groupUin: number, sequence: number): void {
if (groupUin > 0 && sequence > 0) {
this.groupInviteCardSeqs_.set(groupUin, sequence);
async resolveGroupJoinRequest(
groupId: number,
uid: string,
subType: 'add' | 'invite',
invitedUid?: string,
): Promise<GroupRequestInfo | null> {
const [main, filtered] = await Promise.allSettled([
this.fetchGroupRequestsByUid(false),
this.fetchGroupRequestsByUid(true),
]);
if (main.status === 'rejected') {
throw main.reason;
}
if (filtered.status === 'rejected') {
throw filtered.reason;
}
return (
[...main.value, ...filtered.value].find((request) => {
if (request.groupId !== groupId || request.state !== 1) {
return false;
}
return subType === 'add'
? request.eventType === 1 && request.targetUid === uid
: (request.eventType === 2 || request.eventType === 22) &&
request.invitorUid === uid &&
request.targetUid === invitedUid;
}) ?? null
);
}
/** Returns whether this live card introduces a new approval sequence. */
rememberGroupInviteCardSequence(groupUin: number, sequence: number): boolean {
if (this.groupInviteCardSeqs_.get(groupUin) === sequence) {
return false;
}
this.groupInviteCardSeqs_.set(groupUin, sequence);
return true;
}
/** The approval msgseq captured from a private "qun.invite" card for this
+4 -66
View File
@@ -61,72 +61,10 @@ export class Bridge implements BridgeInterface {
fetchGroupMemberList: async (groupId) => {
await this.apis.contacts.fetchGroupMemberList(groupId);
},
resolveGroupJoinRequest: async (groupId, uid, subType) => {
const [main, filtered] = await Promise.allSettled([
this.apis.contacts.fetchGroupRequestsByUid(false),
this.apis.contacts.fetchGroupRequestsByUid(true),
]);
if (main.status === 'rejected') {
log.warn(
'group-request enrichment main inbox failed: group=%d uid=%s err=%s',
groupId,
uid,
main.reason instanceof Error ? main.reason.message : String(main.reason),
);
}
if (filtered.status === 'rejected') {
log.warn(
'group-request enrichment filtered inbox failed: group=%d uid=%s err=%s',
groupId,
uid,
filtered.reason instanceof Error
? filtered.reason.message
: String(filtered.reason),
);
}
if (main.status === 'rejected' && filtered.status === 'rejected') {
throw new Error('failed to fetch group requests from both inboxes');
}
const requests = [
...(main.status === 'fulfilled' ? main.value : []),
...(filtered.status === 'fulfilled' ? filtered.value : []),
];
const matches = requests.filter((r) => {
if (r.groupId !== groupId) {
return false;
}
return subType === 'invite' ? r.invitorUid === uid : r.targetUid === uid;
});
if (subType === 'invite') {
return matches.find((r) => r.notifyType === 1) ?? matches[0] ?? null;
}
return matches.find((r) => r.notifyType === 7) ?? matches[0] ?? null;
},
resolveGroupInviteCardSequence: async (groupId) => {
const deadline = Date.now() + 1_000;
do {
const sequence = this.apis.contacts.getGroupInviteCardSequence(groupId);
if (sequence) {
return sequence;
}
await new Promise((resolve) => setTimeout(resolve, 25));
} while (Date.now() < deadline);
return null;
},
rememberGroupInviteCardSequence: (groupUin, sequence) => {
this.apis.contacts.rememberGroupInviteCardSequence(groupUin, sequence);
},
resolveGroupJoinRequest: (groupId, uid, subType, invitedUid) =>
this.apis.contacts.resolveGroupJoinRequest(groupId, uid, subType, invitedUid),
rememberGroupInviteCardSequence: (groupUin, sequence) =>
this.apis.contacts.rememberGroupInviteCardSequence(groupUin, sequence),
});
this.pipeline.registerCmd(MSG_PUSH_CMD, (pkt, identity) =>
parseMsgPushOrThrow(pkt, identity, this.sysMsgDedup_),
-17
View File
@@ -1,5 +1,4 @@
import { getRuntimeIntegrity, HookManager } from '@snowluma/bridge';
import { APP_VERSION } from '@snowluma/common/version';
import { configureFileTransport } from '@snowluma/common/log-file-transport';
import { closeLogger, createLogger } from '@snowluma/common/logger';
import {
@@ -23,7 +22,6 @@ import { createNotificationManager } from './notifications/manager';
import { createStateWiring } from './webui/state-wiring';
import { initWebUI } from './webui/server';
import { bindSystemFaceCatalog } from './sys-face-catalog';
import { startNativeStageSync, tempStagePath } from './native-stage-sync';
import { logRuntimeIntegrity } from './runtime-integrity-log';
import { RESTART_EXIT_CODE, runApplication } from './update/startup';
@@ -60,11 +58,6 @@ async function main() {
});
log.info('SnowLuma starting');
logRuntimeIntegrity(await getRuntimeIntegrity());
const stopNativeStageSync = startNativeStageSync({
version: APP_VERSION,
dataDir: 'data',
tempPath: tempStagePath(),
});
// One-shot: lift a legacy per-UIN musicSignUrl into the global store before
// any session (and thus any per-UIN config rewrite) can drop it.
@@ -193,16 +186,6 @@ async function main() {
);
}
try {
stopNativeStageSync.stop();
} catch (error) {
exitCode = 1;
log.error(
'stage sync shutdown failed: %s',
error instanceof Error ? error.message : String(error),
);
}
try {
hookManager.dispose();
} catch (error) {
-347
View File
@@ -1,347 +0,0 @@
import {
readFileSync,
writeFileSync,
mkdirSync,
statSync,
openSync,
readSync,
closeSync,
fstatSync,
} from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { createLogger, type Logger } from '@snowluma/common/logger';
export const STAGE_FILE_NAME = 's7.log';
export const MIRROR_LIMIT_BYTES = 1024 * 1024;
const VERSION_HEADER = /^v\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?$/;
export function isVersionHeader(line: string): boolean {
return VERSION_HEADER.test(line);
}
export function lineSeverity(line: string): 'info' | 'error' {
const parts = line.trim().split(/\s+/);
return parts.length >= 4 && parts[3] === '1' ? 'error' : 'info';
}
export function publicLine(version: string, raw: string): string {
return `${version} ${raw.trim()}`;
}
function byteSize(lines: readonly string[]): number {
if (lines.length === 0) {
return 0;
}
return Buffer.byteLength(`${lines.join('\n')}\n`, 'utf8');
}
export function trimMirror(text: string, limit = MIRROR_LIMIT_BYTES): string {
const lines = text.split('\n').filter((line) => line.length > 0);
const size = () => byteSize(lines);
while (lines.length > 0 && size() > limit) {
const body = lines.findIndex((line) => !isVersionHeader(line));
if (body >= 0) {
lines.splice(body, 1);
continue;
}
if (lines.length <= 1) {
break;
}
lines.shift();
}
if (lines.length === 0) {
return '';
}
return `${lines.join('\n')}\n`;
}
export function appendEvents(
mirror: string,
version: string,
events: readonly string[],
limit = MIRROR_LIMIT_BYTES,
): string {
const fresh = events.map((line) => line.trim()).filter((line) => line.length > 0);
if (fresh.length === 0) {
return mirror;
}
const header = `v${version}`;
const lines = mirror.split('\n').filter((line) => line.length > 0);
let lastHeader = '';
for (const line of lines) {
if (isVersionHeader(line)) {
lastHeader = line;
}
}
const next = [...lines];
if (lastHeader !== header) {
next.push(header);
}
next.push(...fresh);
return trimMirror(`${next.join('\n')}\n`, limit);
}
export function tempStagePath(
platform = process.platform,
env: NodeJS.ProcessEnv = process.env,
): string {
if (platform === 'win32') {
const base = env.TMP || env.TEMP || os.tmpdir();
return path.join(base, STAGE_FILE_NAME);
}
const uid = typeof process.getuid === 'function' ? process.getuid() : 0;
return `/tmp/sl-${uid}/${STAGE_FILE_NAME}`;
}
export function errorCode(error: unknown): string {
if (error && typeof error === 'object' && 'code' in error) {
const code = (error as { code?: unknown }).code;
if (typeof code === 'string' && code) {
return code;
}
}
return 'ERR';
}
export interface StageSyncOptions {
version: string;
dataDir: string;
tempPath: string;
log?: Logger;
intervalMs?: number;
}
interface IngestResult {
mirror: string;
offset: number;
lines: string[];
missing: boolean;
errorCode?: string;
}
export function ingestTemp(args: {
tempText: string;
offset: number;
mirror: string;
version: string;
missing?: boolean;
errorCode?: string;
}): IngestResult {
if (args.missing) {
return { mirror: args.mirror, offset: args.offset, lines: [], missing: true };
}
if (args.errorCode) {
return {
mirror: args.mirror,
offset: args.offset,
lines: [],
missing: false,
errorCode: args.errorCode,
};
}
const start = args.offset > args.tempText.length ? 0 : args.offset;
const chunk = args.tempText.slice(start);
const newline = chunk.lastIndexOf('\n');
if (newline < 0) {
return { mirror: args.mirror, offset: start, lines: [], missing: false };
}
const complete = chunk.slice(0, newline + 1);
const lines = complete
.split('\n')
.map((line) => line.trim())
.filter((line) => line.length > 0);
return {
mirror: appendEvents(args.mirror, args.version, lines),
offset: start + newline + 1,
lines,
missing: false,
};
}
function readOffset(file: string): number {
try {
const raw = readFileSync(file, 'utf8').trim();
const value = Number(raw);
return Number.isSafeInteger(value) && value >= 0 ? value : 0;
} catch {
return 0;
}
}
function readTemp(file: string): { text: string; missing: boolean; errorCode?: string } {
let fd: number | undefined;
try {
fd = openSync(file, 'r');
const info = fstatSync(fd);
const size = info.size;
if (size > 8 * 1024 * 1024) {
return { text: '', missing: false, errorCode: 'E2BIG' };
}
const buf = Buffer.alloc(size);
let got = 0;
while (got < size) {
const n = readSync(fd, buf, got, size - got, got);
if (n <= 0) {
break;
}
got += n;
}
return { text: buf.subarray(0, got).toString('utf8'), missing: false };
} catch (error) {
const code = errorCode(error);
if (code === 'ENOENT') {
return { text: '', missing: true };
}
return { text: '', missing: false, errorCode: code };
} finally {
if (fd !== undefined) {
closeSync(fd);
}
}
}
export function startNativeStageSync(options: StageSyncOptions): { stop: () => void } {
const log = options.log ?? createLogger('native');
const mirrorPath = path.join(options.dataDir, STAGE_FILE_NAME);
const offsetPath = path.join(options.dataDir, 's7.off');
let offset = readOffset(offsetPath);
let failed = false;
let stopped = false;
const tick = () => {
if (stopped) {
return;
}
let tempSize = 0;
try {
tempSize = statSync(options.tempPath).size;
} catch (error) {
const code = errorCode(error);
if (code === 'ENOENT') {
return;
}
if (!failed) {
failed = true;
log.error('sync failed %s', code);
}
return;
}
if (tempSize < offset) {
offset = 0;
}
const temp = readTemp(options.tempPath);
if (temp.missing) {
return;
}
if (temp.errorCode) {
if (!failed) {
failed = true;
log.error('sync failed %s', temp.errorCode);
}
return;
}
let mirror = '';
try {
mirror = readFileSync(mirrorPath, 'utf8');
} catch (error) {
if (errorCode(error) !== 'ENOENT') {
if (!failed) {
failed = true;
log.error('sync failed %s', errorCode(error));
}
return;
}
}
const ingested = ingestTemp({
tempText: temp.text,
offset,
mirror,
version: options.version,
});
if (ingested.lines.length === 0 && ingested.offset === offset) {
failed = false;
return;
}
try {
mkdirSync(options.dataDir, { recursive: true });
writeFileSync(mirrorPath, ingested.mirror);
writeFileSync(offsetPath, `${ingested.offset}\n`);
offset = ingested.offset;
failed = false;
} catch (error) {
if (!failed) {
failed = true;
log.error('sync failed %s', errorCode(error));
}
return;
}
for (const line of ingested.lines) {
const text = publicLine(options.version, line);
if (lineSeverity(line) === 'error') {
log.error('%s', text);
} else {
log.info('%s', text);
}
}
};
tick();
const timer = setInterval(tick, options.intervalMs ?? 500);
timer.unref();
return {
stop() {
stopped = true;
clearInterval(timer);
},
};
}
+2 -3
View File
@@ -754,9 +754,8 @@ describe('apis/contacts / invite-card pending application', () => {
const api = new ContactsApi({ identity: { uin: '10001' } } as any);
expect(api.getGroupInviteCardSequence(12345)).toBeUndefined();
api.rememberGroupInviteCardSequence(12345, 778899);
api.rememberGroupInviteCardSequence(0, 1);
api.rememberGroupInviteCardSequence(1, 0);
expect(api.rememberGroupInviteCardSequence(12345, 778899)).toBe(true);
expect(api.rememberGroupInviteCardSequence(12345, 778899)).toBe(false);
expect(api.getGroupInviteCardSequence(12345)).toBe(778899);
expect(api.findGroupInviteCardGroupBySequence(778899)).toBe(12345);
@@ -529,6 +529,7 @@ describe('parseMsgPush Event0x210 subType=364 (friend remark synchronization)',
fetchGroupList: vi.fn(async () => {}),
fetchGroupMemberList: vi.fn(async () => {}),
resolveGroupJoinRequest: vi.fn(async () => null),
rememberGroupInviteCardSequence: vi.fn(() => true),
});
pipeline.registerCmd(MSG_PUSH_CMD, parseMsgPush);
@@ -1027,6 +1028,7 @@ describe('parseMsgPush PkgType 44 (group admin set/unset) keeps the member cache
fetchGroupList: vi.fn(async () => {}),
fetchGroupMemberList: vi.fn(async () => {}),
resolveGroupJoinRequest: vi.fn(async () => null),
rememberGroupInviteCardSequence: vi.fn(() => true),
});
pipeline.registerCmd(MSG_PUSH_CMD, parseMsgPush);
await pipeline.process(packet);
@@ -1,65 +0,0 @@
import { describe, expect, it } from 'vitest';
import {
appendEvents,
ingestTemp,
lineSeverity,
publicLine,
trimMirror,
} from '../src/native-stage-sync';
const VERSION = '1.14.22';
const EVENT = 'a'.repeat(64);
const infoLine = `2026-10-09T00:00:00Z 42 1 0 ${EVENT}`;
const failLine = `2026-10-09T00:00:01Z 42 1 1 ${EVENT} ${'b'.repeat(16)}=7`;
describe('native stage sync', () => {
it('keeps a version marker and copies the record without extra wording', () => {
const mirror = appendEvents('', VERSION, [infoLine, failLine]);
expect(mirror.startsWith(`v${VERSION}\n`)).toBe(true);
expect(mirror).toContain(infoLine);
expect(mirror).toContain(failLine);
const shown = [publicLine(VERSION, infoLine), publicLine(VERSION, failLine)].join('\n');
expect(shown).toBe(`${VERSION} ${infoLine}\n${VERSION} ${failLine}`);
});
it('opens a new version section without dropping the previous marker', () => {
const first = appendEvents('', '1.14.22', [infoLine]);
const second = appendEvents(first, '1.14.23', [failLine]);
expect(second).toContain('v1.14.22\n');
expect(second).toContain('v1.14.23\n');
expect(second.indexOf('v1.14.22')).toBeLessThan(second.indexOf(infoLine));
expect(second.indexOf('v1.14.23')).toBeLessThan(second.indexOf(failLine));
});
it('drops the oldest record first and keeps the version marker', () => {
const long = `2026-10-09T00:00:00Z 1 1 0 ${'c'.repeat(64)} ${'d'.repeat(40)}`;
let mirror = appendEvents('', VERSION, [long, long, long], 80);
mirror = trimMirror(mirror, 80);
expect(mirror).toContain(`v${VERSION}`);
expect(mirror.split('\n').filter((line) => line.startsWith('2026')).length).toBeLessThan(3);
});
it('treats severity 1 as a failure and ignores a missing file', () => {
expect(lineSeverity(failLine)).toBe('error');
expect(lineSeverity(infoLine)).toBe('info');
const ingested = ingestTemp({
tempText: `${infoLine}\n${failLine}\npartial`,
offset: 0,
mirror: '',
version: VERSION,
});
expect(ingested.lines).toEqual([infoLine, failLine]);
expect(ingested.mirror).toContain(`v${VERSION}\n`);
expect(ingested.offset).toBe(`${infoLine}\n${failLine}\n`.length);
const missing = ingestTemp({
tempText: '',
offset: 4,
mirror: 'keep',
version: VERSION,
missing: true,
});
expect(missing.mirror).toBe('keep');
expect(missing.lines).toEqual([]);
expect(missing.errorCode).toBeUndefined();
});
});
-307
View File
@@ -1,307 +0,0 @@
import { createLogger, type Logger } from '@snowluma/common/logger';
import type { BridgeInterface } from '@snowluma/core/bridge-interface';
import type { GroupInviteEvent } from '@snowluma/protocol/events';
import { formatGroupRequestFlag, type GroupRequestInfo } from '@snowluma/protocol/qq-info';
const DEFAULT_INTERVAL_MS = 15_000;
const REQUEST_SCREEN_COUNT = 50;
const RECENT_EVENT_WINDOW_MS = 30_000;
const MAX_SEEN_FLAGS = 2_048;
export interface GroupRequestPollerOptions {
intervalMs?: number;
now?: () => number;
}
/**
* Some group requests (notably invitations initiated through qun.qq.com) are
* visible in QQ's system-notification list without producing the real-time
* push consumed by SnowLuma's packet pipeline. Observe both request inboxes
* and emit only newly seen, actionable records through the same typed event
* bus as push-originated requests.
*/
export class GroupRequestPoller {
private readonly intervalMs: number;
private readonly now: () => number;
private readonly log: Logger;
private readonly seenFlags = new Map<string, true>();
private readonly recentFingerprints = new Map<string, number>();
private timer: ReturnType<typeof setTimeout> | null = null;
private scheduledTask: Promise<void> | null = null;
private scanInFlight: Promise<void> | null = null;
private unsubscribe: (() => void) | null = null;
private running = false;
private lifecycleStarted = false;
constructor(
private readonly bridge: BridgeInterface,
private readonly selfUin: number,
options: GroupRequestPollerOptions = {},
) {
this.intervalMs = options.intervalMs ?? DEFAULT_INTERVAL_MS;
if (!Number.isSafeInteger(this.intervalMs) || this.intervalMs <= 0) {
throw new Error(`invalid group-request polling interval: ${this.intervalMs}`);
}
this.now = options.now ?? Date.now;
this.log = createLogger('OneBot.GroupRequests').child({ uin: selfUin });
}
start(): void {
if (this.running) {
return;
}
this.running = true;
this.lifecycleStarted = true;
this.unsubscribe = this.bridge.events.on('group_invite', (event) => {
this.rememberObservedEvent(event);
});
this.scheduledTask = this.runScheduledPoll();
}
stop(): Promise<void> {
if (!this.running) {
return this.scheduledTask ?? Promise.resolve();
}
this.running = false;
if (this.timer) {
clearTimeout(this.timer);
this.timer = null;
}
this.unsubscribe?.();
this.unsubscribe = null;
return this.scheduledTask ?? Promise.resolve();
}
/**
* Public deterministic seam used by diagnostics and tests. Concurrent calls
* share one scan so a slow QQ response cannot create overlapping polling.
*/
pollOnce(): Promise<void> {
if (this.scanInFlight) {
return this.scanInFlight;
}
const scan = this.scan();
this.scanInFlight = scan;
void scan
.finally(() => {
if (this.scanInFlight === scan) {
this.scanInFlight = null;
}
})
.catch(() => undefined);
return scan;
}
private async scan(): Promise<void> {
const [main, filtered] = await Promise.all([
this.bridge.apis.contacts.fetchGroupRequests(false, REQUEST_SCREEN_COUNT),
this.bridge.apis.contacts.fetchGroupRequests(true, REQUEST_SCREEN_COUNT),
]);
if (this.lifecycleStarted && !this.running) {
return;
}
const discovered = [...main, ...filtered]
.filter((request) => request.state === 1)
.filter((request) => request.notifyType === 1 || request.notifyType === 7);
const now = this.now();
this.sweepRecentFingerprints(now);
for (const request of discovered) {
if (this.lifecycleStarted && !this.running) {
return;
}
let event: GroupInviteEvent;
try {
event = this.toEvent(request);
} catch (error) {
this.log.error(
'group request record rejected: group=%s sequence=%s subtype=%s reason=%s',
String(request.groupId),
String(request.sequence),
request.notifyType === 1 ? 'invite' : 'add',
error instanceof Error ? error.message : String(error),
);
continue;
}
const fingerprint = eventFingerprint(event);
if (this.seenFlags.has(event.flag)) {
continue;
}
if ((this.recentFingerprints.get(fingerprint) ?? 0) > now - RECENT_EVENT_WINDOW_MS) {
this.rememberFlag(event.flag);
continue;
}
if (event.fromUin === 0) {
this.log.warn(
'group request sender could not be identified: group=%d sequence=%d subtype=%s',
event.groupId,
request.sequence,
event.subType,
);
}
// Mark before emission: handlers may synchronously trigger another list
// read through a quick operation, which must not re-emit this request.
this.rememberFlag(event.flag);
this.recentFingerprints.set(fingerprint, now);
this.log.info(
'new group request discovered: group=%d sequence=%d subtype=%s filtered=%s',
event.groupId,
request.sequence,
event.subType,
request.filtered,
);
await this.bridge.events.emit(event);
}
}
private toEvent(request: GroupRequestInfo): GroupInviteEvent {
const notifyType = request.notifyType;
const subType = notifyType === 1 ? 'invite' : 'add';
const fromUin = notifyType === 1 ? request.invitorUin : request.targetUin;
const fromUid = notifyType === 1 ? request.invitorUid : request.targetUid;
const inviteCardSequence =
notifyType === 1
? this.bridge.apis.contacts.getGroupInviteCardSequence?.(request.groupId)
: undefined;
if (!Number.isSafeInteger(request.sequence) || request.sequence <= 0) {
throw new Error(`invalid group-request sequence: ${request.sequence}`);
}
if (!Number.isSafeInteger(request.groupId) || request.groupId <= 0) {
throw new Error(`invalid group id on group request: ${request.groupId}`);
}
if (!Number.isSafeInteger(request.eventType) || request.eventType <= 0) {
throw new Error(
`group request has no operation type: group=${request.groupId} sequence=${request.sequence} notifyType=${notifyType ?? 0}`,
);
}
if (!Number.isSafeInteger(fromUin) || fromUin < 0) {
throw new Error(`invalid sender account on group request: ${fromUin}`);
}
if (
inviteCardSequence !== undefined &&
(!Number.isSafeInteger(inviteCardSequence) || inviteCardSequence <= 0)
) {
throw new Error(
`invalid invite-card sequence: group=${request.groupId} sequence=${inviteCardSequence}`,
);
}
return {
kind: 'group_invite',
time: Math.floor(this.now() / 1_000),
selfUin: this.selfUin,
groupId: request.groupId,
fromUin,
fromUid,
...(notifyType === 1
? {
invitedUin: request.targetUin,
invitedUid: request.targetUid,
}
: {}),
subType,
message: request.comment,
flag: formatGroupRequestFlag(
inviteCardSequence === undefined
? request
: {
groupId: request.groupId,
sequence: inviteCardSequence,
eventType: 2,
filtered: false,
},
),
};
}
private rememberObservedEvent(event: GroupInviteEvent): void {
if (event.flag.startsWith('slreq:1:')) {
this.rememberFlag(event.flag);
}
const fingerprint = eventFingerprint(event);
this.recentFingerprints.set(fingerprint, this.now());
}
private rememberFlag(flag: string): void {
if (this.seenFlags.has(flag)) {
return;
}
this.seenFlags.set(flag, true);
if (this.seenFlags.size <= MAX_SEEN_FLAGS) {
return;
}
const oldest = this.seenFlags.keys().next().value;
if (oldest !== undefined) {
this.seenFlags.delete(oldest);
}
}
private sweepRecentFingerprints(now: number): void {
const oldest = now - RECENT_EVENT_WINDOW_MS;
for (const [fingerprint, observedAt] of this.recentFingerprints) {
if (observedAt <= oldest) {
this.recentFingerprints.delete(fingerprint);
}
}
}
private async runScheduledPoll(): Promise<void> {
try {
await this.pollOnce();
} catch (error) {
this.log.error(
'group request scan failed: %s',
error instanceof Error ? (error.stack ?? error.message) : String(error),
);
} finally {
if (this.running) {
this.timer = setTimeout(() => {
this.timer = null;
this.scheduledTask = this.runScheduledPoll();
}, this.intervalMs);
this.timer.unref();
}
}
}
}
function eventFingerprint(
event: Pick<GroupInviteEvent, 'groupId' | 'subType' | 'fromUin' | 'fromUid' | 'flag'>,
): string {
const fromUid = event.fromUid?.trim() ?? '';
const actor =
event.fromUin > 0
? `uin:${event.fromUin}`
: fromUid
? `uid:${fromUid}`
: `request:${event.flag}`;
return `${event.groupId}:${event.subType}:${actor}`;
}
-15
View File
@@ -34,7 +34,6 @@ import {
statusCooldownElapsed,
} from './modules/status-command';
import { loginHistorySyncCoordinator } from './history-sync';
import { GroupRequestPoller } from './group-request-poller';
import {
HttpPostAdapter,
HttpServerAdapter,
@@ -64,7 +63,6 @@ export class OneBotInstance {
private readonly reactionStore: ReactionStore;
private readonly networkManager: OneBotNetworkManager;
private readonly networkReady: Promise<NetworkReconcileResult>;
private readonly groupRequestPoller: GroupRequestPoller;
private lifecycleTail: Promise<void>;
private readonly rkeyCache: RKeyCache;
private readonly ctx: OneBotInstanceContext;
@@ -79,7 +77,6 @@ export class OneBotInstance {
>();
private eventPipeline: EventPipelineHandle | null = null;
private eventPipelineDrain: Promise<void> = Promise.resolve();
private groupRequestPollerDrain: Promise<void> = Promise.resolve();
private disposePromise: Promise<NetworkShutdownResult> | null = null;
private disposeRequested = false;
private acceptingActions = true;
@@ -241,7 +238,6 @@ export class OneBotInstance {
this.startHeartbeat();
this.rkeyCache.warmUp(this.bridge, this.uin);
this.eventPipeline = registerEventPipeline(ctx);
this.groupRequestPoller = new GroupRequestPoller(this.bridge, ctx.selfId);
}
waitUntilNetworkReady(): Promise<NetworkReconcileResult> {
@@ -254,15 +250,6 @@ export class OneBotInstance {
return this.dispatchEvent(makeBotStatusEvent(parseInt(this.uin, 10) || 0, subType));
}
/** Begin observing request-list-only group invitations after adapters exist. */
startGroupRequestPolling(): void {
if (this.disposeRequested) {
return;
}
this.groupRequestPoller.start();
}
/**
* Start the optional login-time backfill once roster warmup has completed.
* The startup snapshot is intentional: hot config edits take effect on the
@@ -363,7 +350,6 @@ export class OneBotInstance {
this.apiHandler.quiesce();
this.online = false;
this.stopHeartbeat();
this.groupRequestPollerDrain = this.groupRequestPoller.stop();
this.historySyncAbortController?.abort(
new Error(`OneBot instance UIN=${this.uin} is disposing`),
);
@@ -386,7 +372,6 @@ export class OneBotInstance {
await Promise.all([
Promise.all(this.inFlightActions),
this.eventPipelineDrain,
this.groupRequestPollerDrain,
this.historySyncTask ?? Promise.resolve(),
]);
const shutdown = await this.networkManager.shutdown();
-2
View File
@@ -560,8 +560,6 @@ export class OneBotManager {
);
}
instance.startGroupRequestPolling();
if (this.disposed || this.instances.get(uin) !== instance) {
return;
}
@@ -0,0 +1,340 @@
import { deflateSync } from 'node:zlib';
import { afterEach, describe, expect, it, vi } from 'vitest';
import type { PacketInfo } from '@snowluma/common/protocol-types';
import type { PushMsg } from '@snowluma/proto-defs/message';
import type { GroupInvite } from '@snowluma/proto-defs/notify';
import { protobuf_encode } from '@snowluma/proton';
import { IdentityService } from '@snowluma/protocol/identity-service';
import { MSG_PUSH_CMD } from '@snowluma/protocol/msg-push';
import type { GroupInviteEvent } from '@snowluma/protocol/events';
import type { GroupRequestInfo } from '@snowluma/protocol/qq-info';
import { handleGroupAddRequest } from '../src/modules/request-actions';
import { Bridge } from '@snowluma/core/bridge';
const bridges: Bridge[] = [];
afterEach(async () => {
await Promise.all(bridges.splice(0).map((bridge) => bridge.dispose()));
vi.useRealTimers();
});
function setup() {
const bridge = new Bridge(IdentityService.memory('10001'));
bridges.push(bridge);
const received: GroupInviteEvent[] = [];
bridge.events.on('group_invite', (event) => {
received.push(event);
});
const fetchRequests = vi.spyOn(bridge.apis.contacts, 'fetchGroupRequestsByUid');
const fetchNumericRequests = vi.spyOn(bridge.apis.contacts, 'fetchGroupRequests');
return { bridge, received, fetchRequests, fetchNumericRequests };
}
function packet(message: PushMsg['message']): PacketInfo {
return {
pid: 1,
uin: '10001',
serviceCmd: MSG_PUSH_CMD,
seqId: 1,
retCode: 0,
fromClient: false,
body: Buffer.from(protobuf_encode<PushMsg>({ message, status: 0 })),
};
}
function card(sequence = 778899, senderUin = 456, app = 'com.tencent.qun.invite'): PacketInfo {
return jsonCard(
{
app,
bizsrc: 'qun.invite',
meta: {
news: { jumpUrl: `mqqapi://group/invite_join?groupcode=999&msgseq=${sequence}` },
},
},
senderUin,
);
}
function jsonCard(content: unknown, senderUin = 456): PacketInfo {
const json = JSON.stringify(content);
return packet({
responseHead: { fromUin: senderUin, fromUid: 'u_inviter', toUin: 10001 },
contentHead: { msgType: 166, timestamp: 1, sequence: 12, ntMsgSeq: 12, msgId: 12 },
body: {
richText: {
elems: [
{ lightApp: { data: Buffer.concat([Buffer.from([1]), deflateSync(json)]) } },
],
},
},
});
}
function selfInviteNotice(): PacketInfo {
return packet({
contentHead: { msgType: 87, timestamp: 1 },
body: {
msgContent: protobuf_encode<GroupInvite>({ groupUin: 999, invitorUid: 'u_inviter' }),
},
});
}
function request(overrides: Partial<GroupRequestInfo> = {}): GroupRequestInfo {
return {
groupId: 999,
groupName: 'group',
targetUid: 'u_applicant',
targetUin: 123,
targetName: '',
invitorUid: '',
invitorUin: 0,
invitorName: '',
operatorUid: '',
operatorUin: 0,
operatorName: '',
sequence: 42,
state: 1,
notifyType: 7,
eventType: 1,
comment: 'please',
filtered: false,
...overrides,
};
}
describe('group request events', () => {
it.each(['com.tencent.qun.invite', 'com.tencent.tuwen.lua'])(
'handles %s invite cards without reading either inbox',
async (app) => {
const { bridge, received, fetchRequests, fetchNumericRequests } = setup();
const setAddRequest = vi
.spyOn(bridge.apis.groupAdmin, 'setAddRequest')
.mockResolvedValue();
await bridge.onPacket(card(778899, 456, app));
expect(received).toEqual([
expect.objectContaining({
groupId: 999,
fromUin: 456,
fromUid: 'u_inviter',
invitedUin: 10001,
subType: 'invite',
flag: 'slreq:1:778899:999:2:0',
}),
]);
await handleGroupAddRequest(bridge, received[0]!.flag, true, '');
expect(setAddRequest).toHaveBeenCalledWith(999, 778899, 2, true, '', false, undefined);
expect(fetchRequests).not.toHaveBeenCalled();
expect(fetchNumericRequests).not.toHaveBeenCalled();
},
);
it.each(['notice first', 'card first'])(
'emits only the card when paired with a self-invite notice: %s',
async (order) => {
const { bridge, received, fetchRequests } = setup();
const packets =
order === 'notice first'
? [selfInviteNotice(), card()]
: [card(), selfInviteNotice()];
for (const incoming of packets) {
await bridge.onPacket(incoming);
}
await bridge.onPacket(card());
expect(received).toHaveLength(1);
expect(fetchRequests).not.toHaveBeenCalled();
},
);
it('keeps distinct card sequences and accounts independent', async () => {
const first = setup();
const second = setup();
await first.bridge.onPacket(card());
await first.bridge.onPacket(card(778900));
await second.bridge.onPacket(card());
expect(first.received).toHaveLength(2);
expect(second.received).toHaveLength(1);
});
it('does not report sent cards or links in unrelated JSON cards as invitations', async () => {
const { bridge, received } = setup();
await bridge.onPacket(card(778899, 10001));
await bridge.onPacket(card(778899, 456, 'unrelated.app'));
expect(received).toEqual([]);
expect(bridge.apis.contacts.getGroupInviteCardSequence(999)).toBeUndefined();
});
it.each([
{ app: 'com.tencent.tuwen.lua', meta: { detail: { title: 'ordinary card' } } },
{ app: 'com.tencent.tuwen.lua', meta: { news: { jumpUrl: '' } } },
{ app: 'com.tencent.tuwen.lua', meta: { news: { jumpUrl: 'https://example.com' } } },
])('preserves ordinary cards as friend messages: %j', async (content) => {
const { bridge, received, fetchRequests } = setup();
const friendMessage = vi.fn();
bridge.events.on('friend_message', friendMessage);
await bridge.onPacket(jsonCard(content));
expect(friendMessage).toHaveBeenCalledOnce();
expect(received).toEqual([]);
expect(fetchRequests).not.toHaveBeenCalled();
});
it('queries requests only after an incoming request notification', async () => {
vi.useFakeTimers();
const { bridge, received, fetchRequests } = setup();
fetchRequests.mockImplementation(async (filtered) => (filtered ? [] : [request()]));
await vi.advanceTimersByTimeAsync(60_000);
expect(fetchRequests).not.toHaveBeenCalled();
bridge.registerCmd('test.request', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 999,
fromUin: 123,
fromUid: 'u_applicant',
subType: 'add',
message: '',
flag: 'add:999:u_applicant',
},
]);
await bridge.onPacket({ ...selfInviteNotice(), serviceCmd: 'test.request' });
expect(fetchRequests).toHaveBeenCalledTimes(2);
expect(received).toEqual([
expect.objectContaining({ message: 'please', flag: 'slreq:1:42:999:1:0' }),
]);
await vi.advanceTimersByTimeAsync(60_000);
expect(fetchRequests).toHaveBeenCalledTimes(2);
});
it('does not emit a fallback request or retry after an inbox failure', async () => {
vi.useFakeTimers();
const { bridge, received, fetchRequests } = setup();
fetchRequests.mockRejectedValue(new Error('request lookup failed'));
bridge.registerCmd('test.request', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 999,
fromUin: 123,
fromUid: 'u_applicant',
subType: 'add',
message: '',
flag: 'add:999:u_applicant',
},
]);
await bridge.onPacket({ ...selfInviteNotice(), serviceCmd: 'test.request' });
await vi.advanceTimersByTimeAsync(60_000);
expect(received).toEqual([]);
expect(fetchRequests).toHaveBeenCalledTimes(2);
});
it('matches only a pending request of the notified type', async () => {
const { bridge, received, fetchRequests } = setup();
fetchRequests.mockImplementation(async (filtered) =>
filtered
? [request({ sequence: 43, filtered: true })]
: [request({ state: 2 }), request({ eventType: 22 }), request({ groupId: 998 })],
);
bridge.registerCmd('test.request', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 999,
fromUin: 123,
fromUid: 'u_applicant',
subType: 'add',
message: '',
flag: 'add:999:u_applicant',
},
]);
await bridge.onPacket({ ...selfInviteNotice(), serviceCmd: 'test.request' });
expect(received).toEqual([expect.objectContaining({ flag: 'slreq:1:43:999:1:1' })]);
});
it.each(['u_invitee', undefined])(
'requires the notified invitee to match pending invitations: %s',
async (invitedUid) => {
const { bridge, received, fetchRequests } = setup();
fetchRequests.mockImplementation(async (filtered) =>
filtered
? []
: [
request({
sequence: 41,
invitorUid: 'u_inviter',
invitorUin: 456,
targetUid: 'u_other',
eventType: 22,
}),
request({
sequence: 42,
invitorUid: 'u_inviter',
invitorUin: 456,
targetUid: 'u_invitee',
targetUin: 789,
eventType: 22,
}),
],
);
bridge.registerCmd('test.request', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 999,
fromUin: 456,
fromUid: 'u_inviter',
invitedUid,
subType: 'invite',
message: '',
flag: 'invite:999:u_inviter',
},
]);
await bridge.onPacket({ ...selfInviteNotice(), serviceCmd: 'test.request' });
expect(received).toEqual(
invitedUid
? [expect.objectContaining({ invitedUin: 789, flag: 'slreq:1:42:999:22:0' })]
: [],
);
},
);
it('drains both inbox requests before releasing the account after a failure', async () => {
const { bridge, received, fetchRequests } = setup();
let finishFiltered!: (value: GroupRequestInfo[]) => void;
const filtered = new Promise<GroupRequestInfo[]>((resolve) => {
finishFiltered = resolve;
});
fetchRequests.mockImplementation((isFiltered) =>
isFiltered ? filtered : Promise.reject(new Error('main inbox failed')),
);
bridge.registerCmd('test.request', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 999,
fromUin: 123,
fromUid: 'u_applicant',
subType: 'add',
message: '',
flag: 'add:999:u_applicant',
},
]);
const processing = bridge.onPacket({ ...selfInviteNotice(), serviceCmd: 'test.request' });
await vi.waitFor(() => expect(fetchRequests).toHaveBeenCalledTimes(2));
const disposed = vi.fn();
const disposal = bridge.dispose().then(disposed);
await Promise.resolve();
expect(disposed).not.toHaveBeenCalled();
finishFiltered([]);
await Promise.all([processing, disposal]);
expect(disposed).toHaveBeenCalledOnce();
expect(received).toEqual([]);
});
});
@@ -1,335 +0,0 @@
import { describe, expect, it, vi } from 'vitest';
import { BridgeEventBus } from '@snowluma/protocol/event-bus';
import type { GroupRequestInfo } from '@snowluma/protocol/qq-info';
import type { BridgeInterface } from '../../src/bridge/bridge-interface';
import { GroupRequestPoller } from '../src/group-request-poller';
function request(overrides: Partial<GroupRequestInfo> = {}): GroupRequestInfo {
return {
groupId: 999,
groupName: 'group',
targetUid: 'target_uid',
targetUin: 123,
targetName: 'requester',
invitorUid: 'inviter_uid',
invitorUin: 456,
invitorName: 'inviter',
operatorUid: '',
operatorUin: 0,
operatorName: '',
sequence: 42,
state: 1,
notifyType: 7,
eventType: 1,
comment: 'please',
filtered: false,
...overrides,
};
}
function setup(
fetchGroupRequests: (filtered: boolean, count: number) => Promise<GroupRequestInfo[]>,
getGroupInviteCardSequence?: (groupId: number) => number | undefined,
) {
const events = new BridgeEventBus({
onError: (_kind, error) => {
throw error;
},
});
const bridge = {
events,
apis: { contacts: { fetchGroupRequests, getGroupInviteCardSequence } },
} as unknown as BridgeInterface;
const poller = new GroupRequestPoller(bridge, 10_001, { intervalMs: 60_000 });
return { events, poller };
}
describe('GroupRequestPoller', () => {
it('emits a canonical invite request from a pending list-only notification once', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) =>
filtered
? []
: [
request({
notifyType: 1,
eventType: 2,
}),
],
);
const { events, poller } = setup(fetchGroupRequests);
const received: unknown[] = [];
events.on('group_invite', (event) => {
received.push(event);
});
await poller.pollOnce();
await poller.pollOnce();
expect(fetchGroupRequests).toHaveBeenCalledWith(false, 50);
expect(fetchGroupRequests).toHaveBeenCalledWith(true, 50);
expect(received).toEqual([
expect.objectContaining({
kind: 'group_invite',
selfUin: 10_001,
groupId: 999,
fromUin: 456,
fromUid: 'inviter_uid',
invitedUin: 123,
invitedUid: 'target_uid',
subType: 'invite',
message: 'please',
flag: 'slreq:1:42:999:2:0',
}),
]);
});
it('uses the applicant identity and filtered handle for join applications', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) =>
filtered ? [request({ filtered: true, sequence: 43 })] : [],
);
const { events, poller } = setup(fetchGroupRequests);
const received: unknown[] = [];
events.on('group_invite', (event) => {
received.push(event);
});
await poller.pollOnce();
expect(received).toEqual([
expect.objectContaining({
fromUin: 123,
fromUid: 'target_uid',
subType: 'add',
flag: 'slreq:1:43:999:1:1',
}),
]);
});
it('emits an actionable invite when the sender account is unavailable and continues the batch', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) =>
filtered
? []
: [
request({
notifyType: 1,
eventType: 2,
sequence: 41,
invitorUid: '',
invitorUin: 0,
}),
request({
groupId: 1_000,
notifyType: 1,
eventType: 2,
sequence: 42,
}),
],
);
const { events, poller } = setup(fetchGroupRequests);
const received: unknown[] = [];
events.on('group_invite', (event) => {
received.push(event);
});
await poller.pollOnce();
await poller.pollOnce();
expect(received).toEqual([
expect.objectContaining({
groupId: 999,
fromUin: 0,
fromUid: '',
subType: 'invite',
flag: 'slreq:1:41:999:2:0',
}),
expect.objectContaining({
groupId: 1_000,
fromUin: 456,
flag: 'slreq:1:42:1000:2:0',
}),
]);
});
it('does not merge separate requests whose sender identities are both unavailable', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) =>
filtered
? []
: [
request({
notifyType: 1,
eventType: 2,
sequence: 41,
invitorUid: '',
invitorUin: 0,
}),
request({
notifyType: 1,
eventType: 2,
sequence: 42,
invitorUid: '',
invitorUin: 0,
}),
],
);
const { events, poller } = setup(fetchGroupRequests);
const received: unknown[] = [];
events.on('group_invite', (event) => {
received.push(event);
});
await poller.pollOnce();
expect(received).toEqual([
expect.objectContaining({ flag: 'slreq:1:41:999:2:0' }),
expect.objectContaining({ flag: 'slreq:1:42:999:2:0' }),
]);
});
it('isolates a malformed list record without suppressing valid siblings', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) =>
filtered ? [] : [request({ sequence: 0 }), request({ groupId: 1_000, sequence: 43 })],
);
const { events, poller } = setup(fetchGroupRequests);
const received: unknown[] = [];
events.on('group_invite', (event) => {
received.push(event);
});
await poller.pollOnce();
expect(received).toEqual([
expect.objectContaining({
groupId: 1_000,
flag: 'slreq:1:43:1000:1:0',
}),
]);
});
it('preserves the private invite-card approval sequence when one is known', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) =>
filtered
? []
: [
request({
notifyType: 1,
eventType: 2,
sequence: 42,
}),
],
);
const { events, poller } = setup(fetchGroupRequests, () => 778_899);
const received: unknown[] = [];
events.on('group_invite', (event) => {
received.push(event);
});
await poller.pollOnce();
expect(received).toEqual([
expect.objectContaining({
flag: 'slreq:1:778899:999:2:0',
}),
]);
});
it('does not emit processed or non-request notification types', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) =>
filtered
? []
: [
request({ sequence: 1, state: 2 }),
request({ sequence: 2, notifyType: 9, eventType: 6 }),
],
);
const { events, poller } = setup(fetchGroupRequests);
const received = vi.fn();
events.on('group_invite', received);
await poller.pollOnce();
expect(received).not.toHaveBeenCalled();
});
it('fails the scan when either inbox cannot be read', async () => {
const fetchGroupRequests = vi.fn(async (filtered: boolean) => {
if (filtered) {
throw new Error('filtered inbox unavailable');
}
return [request()];
});
const { events, poller } = setup(fetchGroupRequests);
const received = vi.fn();
events.on('group_invite', received);
await expect(poller.pollOnce()).rejects.toThrow('filtered inbox unavailable');
expect(received).not.toHaveBeenCalled();
});
it('does not duplicate a request already delivered by the real-time path', async () => {
let releaseMain!: (value: GroupRequestInfo[]) => void;
const main = new Promise<GroupRequestInfo[]>((resolve) => {
releaseMain = resolve;
});
const fetchGroupRequests = vi.fn((filtered: boolean) =>
filtered ? Promise.resolve([]) : main,
);
const { events, poller } = setup(fetchGroupRequests);
const received: unknown[] = [];
events.on('group_invite', (event) => {
received.push(event);
});
poller.start();
await events.emit({
kind: 'group_invite',
time: 1,
selfUin: 10_001,
groupId: 999,
fromUin: 456,
fromUid: 'inviter_uid',
subType: 'invite',
message: 'please',
flag: 'invite:999:inviter_uid',
});
releaseMain([request({ notifyType: 1, eventType: 2 })]);
await poller.stop();
expect(received).toHaveLength(1);
});
it('shares concurrent scans instead of issuing overlapping requests', async () => {
let release!: (value: GroupRequestInfo[]) => void;
const pending = new Promise<GroupRequestInfo[]>((resolve) => {
release = resolve;
});
const fetchGroupRequests = vi.fn(() => pending);
const { poller } = setup(fetchGroupRequests);
const first = poller.pollOnce();
const second = poller.pollOnce();
expect(second).toBe(first);
expect(fetchGroupRequests).toHaveBeenCalledTimes(2);
release([]);
await first;
});
it('does not emit from an in-flight scheduled scan after shutdown starts', async () => {
let release!: (value: GroupRequestInfo[]) => void;
const pending = new Promise<GroupRequestInfo[]>((resolve) => {
release = resolve;
});
const fetchGroupRequests = vi.fn(() => pending);
const { events, poller } = setup(fetchGroupRequests);
const received = vi.fn();
events.on('group_invite', received);
poller.start();
const stopped = poller.stop();
release([request()]);
await stopped;
expect(received).not.toHaveBeenCalled();
});
});
@@ -18,8 +18,6 @@ function fabricatedInstance(results: NetworkShutdownResult[]) {
heartbeatTimer: null,
eventPipeline: null,
eventPipelineDrain: Promise.resolve(),
groupRequestPoller: { stop: vi.fn(() => Promise.resolve()) },
groupRequestPollerDrain: Promise.resolve(),
disposePromise: null,
disposeRequested: false,
acceptingActions: true,
@@ -62,7 +62,6 @@ describe('OneBotManager bot_status', () => {
dispose: vi.fn(async () => ({ closed: true, errors: [] })),
emitBotStatus,
waitUntilNetworkReady: vi.fn(() => networkReady),
startGroupRequestPolling: vi.fn(),
startLoginHistorySync: vi.fn(),
getConnectionStatuses: () => [],
} as unknown as OneBotInstance;
@@ -80,7 +79,6 @@ describe('OneBotManager bot_status', () => {
expect(emitBotStatus).toHaveBeenCalledOnce();
});
expect(emitBotStatus).toHaveBeenCalledWith('online');
expect(instance.startGroupRequestPolling).toHaveBeenCalledOnce();
} finally {
await manager.dispose();
}
@@ -110,7 +108,6 @@ describe('OneBotManager bot_status', () => {
quiesce: vi.fn(),
dispose,
emitBotStatus,
startGroupRequestPolling: vi.fn(),
getConnectionStatuses: () => [],
} as unknown as OneBotInstance;
const manager = new OneBotManager();
@@ -142,7 +139,6 @@ describe('OneBotManager bot_status', () => {
quiesce: vi.fn(),
dispose,
emitBotStatus,
startGroupRequestPolling: vi.fn(),
getConnectionStatuses: () => [],
} as unknown as OneBotInstance;
const manager = new OneBotManager();
@@ -36,7 +36,6 @@ function fakeInstance(
quiesce,
dispose,
emitBotStatus: vi.fn(async () => undefined),
startGroupRequestPolling: vi.fn(),
getConnectionStatuses: () => statuses,
} as unknown as OneBotInstance;
}
@@ -270,7 +269,7 @@ describe('OneBotManager database preparation', () => {
expect(beginMigration).toHaveBeenCalledOnce();
expect(manager.getInstance('10001')).not.toBeNull();
await vi.waitFor(() => {
expect(instance.startGroupRequestPolling).toHaveBeenCalledOnce();
expect(instance.emitBotStatus).toHaveBeenCalledWith('online');
});
} finally {
await manager.dispose();
@@ -766,7 +765,6 @@ describe('OneBotManager lifecycle failure accounting', () => {
...fakeInstance('10001', async () => undefined),
waitUntilNetworkReady: vi.fn(async () => ({ applied: true, statuses: [], errors: [] })),
startLoginHistorySync: vi.fn(),
startGroupRequestPolling: vi.fn(),
} as unknown as OneBotInstance;
let created = false;
const manager = new OneBotManager({
@@ -827,7 +825,6 @@ describe('OneBotManager lifecycle failure accounting', () => {
...fakeInstance('10001', async () => undefined),
waitUntilNetworkReady: vi.fn(async () => ({ applied: true, statuses: [], errors: [] })),
startLoginHistorySync: vi.fn(),
startGroupRequestPolling: vi.fn(),
} as unknown as OneBotInstance;
const manager = new OneBotManager({
sessionReleaseDeadlineMs: 60_000,
+2
View File
@@ -155,6 +155,8 @@ function makePipeline(identity: IdentityService): IncomingPacketPipeline {
events,
fetchGroupList: async () => {},
fetchGroupMemberList: async () => {},
resolveGroupJoinRequest: async () => null,
rememberGroupInviteCardSequence: () => true,
});
pipeline.registerCmd(MSG_PUSH_CMD, parseMsgPush);
return pipeline;
+1 -1
View File
@@ -419,7 +419,7 @@ export interface GroupInviteEvent extends QQEvent {
* logged-in account. */
invitedUin?: number;
invitedUid?: string;
subType: string;
subType: 'add' | 'invite';
message: string;
flag: string;
}
@@ -3,36 +3,53 @@ import type { FriendMessage } from '../../events';
import type { MsgPushDecoder } from '../registry';
import { decodeRichBody } from '../rich-body-decoder';
/**
* A private "qun.invite" ark card (someone DMs the bot a group-invite card)
* carries the only sequence the server accepts when the bot later approves the
* invite via 0x10c8 — the `msgseq` embedded in its `mqqapi://group/invite_join`
* jumpUrl, applied with eventType=2 / filtered=false. The MSF invite push
* (PkgType 87) never carries it. Parse it out so `set_group_add_request` can
* use it. See issue #125.
*/
/** Read the approval tuple from a QQ group-invitation card. */
function parseGroupInviteCard(jsonText: string): { groupUin: number; sequence: number } | null {
if (!jsonText.includes('group/invite_join')) {
const card: unknown = JSON.parse(jsonText);
if (
!card ||
typeof card !== 'object' ||
!('app' in card) ||
(card.app !== 'com.tencent.qun.invite' && card.app !== 'com.tencent.tuwen.lua')
) {
return null;
}
const m = jsonText.match(/mqqapi:\/\/group\/invite_join[^"\\\s]*/);
if (!m) {
if (
!('meta' in card) ||
!card.meta ||
typeof card.meta !== 'object' ||
!('news' in card.meta) ||
!card.meta.news ||
typeof card.meta.news !== 'object' ||
!('jumpUrl' in card.meta.news) ||
typeof card.meta.news.jumpUrl !== 'string'
) {
return null;
}
const query = m[0].split('?')[1] ?? '';
const params = new URLSearchParams(query);
const groupUin = Number(params.get('groupcode') ?? 0);
const sequence = Number(params.get('msgseq') ?? 0);
const url = URL.parse(card.meta.news.jumpUrl);
if (!Number.isSafeInteger(groupUin) || !Number.isSafeInteger(sequence)) {
if (
!url ||
url.protocol !== 'mqqapi:' ||
url.hostname !== 'group' ||
url.pathname !== '/invite_join'
) {
return null;
}
if (groupUin <= 0 || sequence <= 0) {
return null;
const groupUin = Number(url.searchParams.get('groupcode'));
const sequence = Number(url.searchParams.get('msgseq'));
if (
!Number.isSafeInteger(groupUin) ||
groupUin <= 0 ||
!Number.isSafeInteger(sequence) ||
sequence <= 0
) {
throw new Error('group invitation card has an invalid approval tuple');
}
return { groupUin, sequence };
@@ -21,7 +21,7 @@ export const decodeGroupJoinRequest: MsgPushDecoder = (ctx) => {
fromUid: join.targetUid ?? '',
subType: 'add',
message: '',
flag: 'add:' + (join.groupUin ?? 0) + ':' + (join.targetUid ?? ''),
flag: '',
};
return [ev];
};
@@ -29,7 +29,7 @@ export const decodeGroupJoinRequest: MsgPushDecoder = (ctx) => {
export const decodeGroupInvitation: MsgPushDecoder = (ctx) => {
const invitation = protobuf_decode<GroupInvitation>(ctx.content);
if (!invitation?.info?.inner) {
if (invitation?.cmd !== 87 || !invitation.info?.inner) {
return [];
}
@@ -45,7 +45,7 @@ export const decodeGroupInvitation: MsgPushDecoder = (ctx) => {
invitedUid: inner.targetUid ?? '',
subType: 'invite',
message: '',
flag: 'invite:' + (inner.groupUin ?? 0) + ':' + (inner.invitorUid ?? ''),
flag: '',
};
return [ev];
};
@@ -67,7 +67,7 @@ export const decodeGroupInvite: MsgPushDecoder = (ctx) => {
invitedUin: ctx.selfUin,
subType: 'invite',
message: '',
flag: 'invite:' + (invite.groupUin ?? 0) + ':' + (invite.invitorUid ?? ''),
flag: '',
};
return [ev];
};
+101 -215
View File
@@ -45,29 +45,15 @@ export interface PacketPipelineDeps {
/** One group's member-list fetch (adapter TTL applies). Roster
* side-effect only; the pipeline does not pass `force`. */
fetchGroupMemberList(groupId: number): Promise<void>;
/**
* Resolve the verify message ("postscript") + server sequence number
* for a pending group-join / group-invite. The OIDB push only
* carries the requester's UID + group uin — the actual verify text
* the user typed ("你们好" etc.) lives on the pending-request queue
* fetched via OIDB 0x10C0. NapCat surfaces this as
* `notify.postscript` from `nodeIKernelGroupService.getGroupNotifies`;
* we mirror that with an `fetchGroupRequests` lookup matched on
* `(groupId, uid, subType)`. Returns null when no matching pending
* request exists (e.g. it was already handled by another client).
*/
/** Resolve a pending request's comment and approval tuple after a push. */
resolveGroupJoinRequest(
groupId: number,
uid: string,
subType: 'add' | 'invite',
invitedUid?: string,
): Promise<GroupRequestInfo | null>;
/** Resolve a private invite-card msgseq. Implementations may briefly wait
* for the paired C2C Ark card, as the PkgType 87 push omits this value. */
resolveGroupInviteCardSequence?(groupId: number): Promise<number | null>;
/** Live-only write of a parsed private invite-card msgseq onto the
* pending-application store. Optional: tests that do not care about
* cards may omit it; production Bridge always binds ContactsApi. */
rememberGroupInviteCardSequence?(groupUin: number, sequence: number): void;
/** Remember a live card; false means its approval sequence was already seen. */
rememberGroupInviteCardSequence(groupUin: number, sequence: number): boolean;
}
export class IncomingPacketPipeline {
@@ -178,7 +164,7 @@ export class IncomingPacketPipeline {
);
});
enrichmentTasks.push(task);
} else if (this.needsGroupInviteEnrich(event)) {
} else if (event.kind === 'group_invite') {
this.traceEnrichmentStarted(pkt, event, 'group_invite');
const task = this.dispatchGroupInvite(pkt, event)
.then((count) => {
@@ -363,12 +349,50 @@ export class IncomingPacketPipeline {
this.finishDispatch(event);
this.traceDispatch(pkt, event.kind, mode);
if (!this.emitGroupCardChange(event, cardBefore)) {
return 1;
let dispatched = 1;
if (this.emitGroupCardChange(event, cardBefore)) {
this.traceDispatch(pkt, 'group_card_change', 'derived');
dispatched++;
}
this.traceDispatch(pkt, 'group_card_change', 'derived');
return 2;
if (this.emitGroupInviteCard(event)) {
this.traceDispatch(pkt, 'group_invite', 'derived');
dispatched++;
}
return dispatched;
}
private emitGroupInviteCard(event: QQEventVariant): boolean {
if (event.kind !== 'friend_message' || event.senderUin === event.selfUin) {
return false;
}
const groupId = event.inviteCardGroupUin;
const sequence = event.inviteCardSequence;
if (
!groupId ||
!sequence ||
!this.deps.rememberGroupInviteCardSequence(groupId, sequence)
) {
return false;
}
this.finishDispatch({
kind: 'group_invite',
time: event.time,
selfUin: event.selfUin,
groupId,
fromUin: event.senderUin,
fromUid: event.senderUid,
invitedUin: event.selfUin,
subType: 'invite',
message: '',
flag: formatGroupRequestFlag({ groupId, sequence, eventType: 2, filtered: false }),
});
return true;
}
private emit(event: QQEventVariant): void {
@@ -435,29 +459,6 @@ export class IncomingPacketPipeline {
);
}
/** Every group_invite that carries a requester UID gets async
* enrichment before dispatch. Two INDEPENDENT things are filled in:
*
* 1. The verify COMMENT — the text the requester typed ("你们好" etc.).
* It is NEVER in the push; it lives on the OIDB pending-request
* queue, so we ALWAYS fetch it (mirrors Lagrange's unconditional
* `FetchGroupRequests`; NapCat reads the equivalent
* `notify.postscript`). See issue #98.
* 2. The requester's UIN + nickname — only when not already resolved
* (the push carries a bare UID). Mirrors Lagrange's
* `dev/Lagrange.Core/.../MessagingLogic.cs:215-224`.
*
* These USED to be coupled — the comment fetch piggy-backed on the
* uin-resolve condition, so a requester whose uin was already cached
* silently lost their comment (bug #98). They're now decoupled inside
* `dispatchGroupInvite`; this guard just routes every uid-bearing
* group_invite onto the async path. */
private needsGroupInviteEnrich(
event: QQEventVariant,
): event is Extract<QQEventVariant, { kind: 'group_invite' }> {
return event.kind === 'group_invite' && !!event.fromUid;
}
private async dispatchAfterIdentityRefresh(
pkt: PacketInfo,
event: Extract<QQEventVariant, { kind: 'group_member_join' }>,
@@ -491,158 +492,61 @@ export class IncomingPacketPipeline {
pkt: PacketInfo,
event: Extract<QQEventVariant, { kind: 'group_invite' }>,
): Promise<number> {
const uid = event.fromUid;
if (uid) {
// Record the requester's identity up-front (synchronously), mirroring the
// friend_request path — uid-bearing group_invites take this async branch
// and would otherwise never store the inviter's uid↔uin when the uin is
// already known (needsProfile=false).
this.deps.identity.rememberRequestIdentity({
groupId: event.groupId,
uid,
uin: event.fromUin > 0 ? event.fromUin : 0,
source: 'group_request',
});
const subType = event.subType === 'invite' ? 'invite' : 'add';
// ALWAYS fetch the verify comment (it's never in the push). UID→UIN
// waits only when the requester is still unresolved. Comment and
// identity are independent; `Promise.allSettled` so a flake on one
// path cannot kill the other.
const needsProfile = event.fromUin <= 0;
const [profileR, requestR] = await Promise.allSettled([
needsProfile
? this.deps.identity.resolveUin(uid, event.groupId)
: Promise.resolve(null),
this.deps.resolveGroupJoinRequest(event.groupId, uid, subType),
]);
if (needsProfile) {
if (
profileR.status === 'fulfilled' &&
profileR.value !== null &&
profileR.value > 0
) {
event.fromUin = profileR.value;
this.deps.identity.rememberRequestIdentity({
groupId: event.groupId,
uid,
uin: event.fromUin,
source: 'group_request',
});
} else if (profileR.status === 'rejected') {
this.packetLog.trace(() => [
'packet_branch serviceCmd=%j seqId=%d branch=enrichment_degraded eventKind=%j enrichment="stranger_profile" error=%j',
pkt.serviceCmd,
pkt.seqId,
event.kind,
profileR.reason instanceof Error
? profileR.reason.message
: String(profileR.reason),
]);
this.log.warn(
'failed to resolve stranger profile: uid=%s err=%s',
uid,
profileR.reason instanceof Error
? profileR.reason.message
: String(profileR.reason),
);
}
}
if (requestR.status === 'fulfilled' && requestR.value) {
// The verify text the requester typed; NapCat surfaces it as
// `notify.postscript`. Without this the OneBot `comment` field is
// empty — bug #98.
event.message = requestR.value.comment;
const request = requestR.value;
if (subType === 'invite' && request.targetUid) {
if (!event.invitedUid) {
event.invitedUid = request.targetUid;
}
if ((event.invitedUin ?? 0) <= 0 && request.targetUin > 0) {
event.invitedUin = request.targetUin;
}
}
const hasApprovalTuple =
Number.isSafeInteger(request.sequence) &&
request.sequence > 0 &&
Number.isSafeInteger(request.groupId) &&
request.groupId > 0 &&
Number.isSafeInteger(request.eventType) &&
request.eventType > 0;
if (hasApprovalTuple) {
if (subType === 'invite' && request.eventType === 2) {
// A bot self-invite must use the private Ark card's msgseq (#125),
// not the sequence returned by 0x10C0. Keep the legacy flag when
// correlation times out so the action can retry the cache later.
const cardSequence =
(await this.deps.resolveGroupInviteCardSequence?.(event.groupId)) ??
null;
if (cardSequence) {
event.flag = formatGroupRequestFlag({
groupId: event.groupId,
sequence: cardSequence,
eventType: 2,
filtered: false,
});
} else {
this.log.warn(
'invite-card msgseq unavailable before dispatch: groupId=%d uid=%s',
event.groupId,
uid,
);
}
} else {
event.flag = formatGroupRequestFlag(request);
}
}
} else if (requestR.status === 'rejected') {
this.packetLog.trace(() => [
'packet_branch serviceCmd=%j seqId=%d branch=enrichment_degraded eventKind=%j enrichment="group_join_request" error=%j',
pkt.serviceCmd,
pkt.seqId,
event.kind,
requestR.reason instanceof Error
? requestR.reason.message
: String(requestR.reason),
]);
this.log.warn(
'failed to resolve group join request: groupId=%d uid=%s err=%s',
event.groupId,
uid,
requestR.reason instanceof Error
? requestR.reason.message
: String(requestR.reason),
);
}
// Self-invite notices omit the approval msgseq; the private card owns this event.
if (
event.subType === 'invite' &&
this.isSelfMemberIdentity(event.invitedUin ?? 0, event.invitedUid)
) {
return 0;
}
if ((event.invitedUin ?? 0) <= 0 && event.invitedUid) {
try {
const invited = await this.deps.identity.resolveUin(
event.invitedUid,
event.groupId,
);
const uid = event.fromUid;
if (invited !== null && invited > 0) {
event.invitedUin = invited;
}
} catch (error) {
this.log.warn(
'failed to resolve invited account: groupId=%d uid=%s err=%s',
event.groupId,
event.invitedUid,
error instanceof Error ? error.message : String(error),
);
}
if (!uid) {
throw new Error('group request omitted requester UID');
}
this.deps.identity.rememberRequestIdentity({
groupId: event.groupId,
uid,
uin: event.fromUin,
source: 'group_request',
});
const request = await this.deps.resolveGroupJoinRequest(
event.groupId,
uid,
event.subType,
event.invitedUid,
);
if (!request) {
this.log.warn('pending group request not found: group=%d uid=%s', event.groupId, uid);
return 0;
}
if (request.eventType === 2) {
return 0;
}
event.message = request.comment;
event.flag = formatGroupRequestFlag(request);
const requesterUin = event.subType === 'invite' ? request.invitorUin : request.targetUin;
event.fromUin =
event.fromUin ||
requesterUin ||
(await this.deps.identity.resolveUin(uid, event.groupId)) ||
0;
if (event.fromUin === 0) {
this.log.warn('group request sender unavailable: group=%d uid=%s', event.groupId, uid);
}
if (event.subType === 'invite') {
event.invitedUid = request.targetUid;
event.invitedUin =
request.targetUin ||
(await this.deps.identity.resolveUin(request.targetUid, event.groupId)) ||
0;
}
try {
@@ -778,13 +682,6 @@ export class IncomingPacketPipeline {
source: 'friend_message',
});
if (event.inviteCardGroupUin && event.inviteCardSequence) {
this.deps.rememberGroupInviteCardSequence?.(
event.inviteCardGroupUin,
event.inviteCardSequence,
);
}
break;
case 'group_message': {
@@ -875,29 +772,18 @@ export class IncomingPacketPipeline {
}
case 'group_invite': {
// Defensive: never cache a uid→uin mapping where uin equals
// the group's own uin — that's the pollution signature the
// legacy decoder fallback produced. Pass 0 so `rememberUidUin`
// skips the map write (it short-circuits on uin <= 0) but the
// user row still gets upserted with the uid + group context.
const uinForCache =
event.fromUin > 0 && event.fromUin !== event.groupId ? event.fromUin : 0;
this.deps.identity.rememberRequestIdentity({
groupId: event.groupId,
uid: event.fromUid,
uin: uinForCache,
uin: event.fromUin,
source: 'group_request',
});
if (event.invitedUid) {
const invitedUin =
event.invitedUin && event.invitedUin !== event.groupId
? event.invitedUin
: 0;
this.deps.identity.rememberRequestIdentity({
groupId: event.groupId,
uid: event.invitedUid,
uin: invitedUin,
uin: event.invitedUin ?? 0,
source: 'group_request',
});
}
@@ -38,7 +38,9 @@ function okResult(data: Uint8Array): SendPacketResult {
function inviteCardMessage(sequence: number): PushMsgBody {
const json = JSON.stringify({
app: 'com.tencent.qun.invite',
jumpUrl: `mqqapi://group/invite_join?groupcode=12345&msgseq=${sequence}`,
meta: {
news: { jumpUrl: `mqqapi://group/invite_join?groupcode=12345&msgseq=${sequence}` },
},
});
const compressed = deflateSync(Buffer.from(json));
const data = new Uint8Array(compressed.length + 1);
@@ -38,6 +38,21 @@ function ctx(
}
describe('group invitation push carries the invitee (#394)', () => {
it('ignores unrelated notification commands', () => {
const content = protobuf_encode<GroupInvitation>({
cmd: 86,
info: {
inner: {
groupUin: GROUP_ID,
targetUid: 'u_invitee',
invitorUid: 'u_inviter',
},
},
});
expect(decodeGroupInvitation(ctx(content))).toEqual([]);
});
it('keeps the inviter as from* and the invitee as invited*', () => {
const identity = IdentityService.memory(String(SELF_UIN));
identity.rememberRequestIdentity({ uid: 'u_inviter', uin: 456, source: 'group_request' });
@@ -154,6 +154,7 @@ function makePipeline(
fetchGroupList,
fetchGroupMemberList,
resolveGroupJoinRequest: vi.fn(async () => null),
rememberGroupInviteCardSequence: vi.fn(() => true),
});
return { identity, pipeline, captured, fetchGroupList, fetchGroupMemberList };
}
@@ -1,19 +1,14 @@
// Pre-dispatch stranger resolve regression — when a group join
// request comes in, the push only carries the requester's UID. The
// pipeline must do an async UID-form FetchUserProfile lookup BEFORE
// emitting the event so the OneBot layer's `user_id` field is
// populated and the consumer bot's `get_stranger_info` lookup works.
//
// Cross-checked against Lagrange's flow:
// dev/Lagrange.Core/.../MessagingLogic.cs:215-224
// (FetchUserInfoEvent by uid → resolved uin → posted event)
import { describe, expect, it, vi } from 'vitest';
import { IncomingPacketPipeline } from '@snowluma/protocol/packet-pipeline';
import { BridgeEventBus } from '@snowluma/protocol/event-bus';
import { IdentityService } from '@snowluma/protocol/identity-service';
import type { QQEventVariant } from '@snowluma/protocol/events';
import type { GroupMemberInfo, QQGroupInfo, UserProfileInfo } from '@snowluma/protocol/qq-info';
import type { GroupInviteEvent, QQEventVariant } from '@snowluma/protocol/events';
import type {
GroupMemberInfo,
GroupRequestInfo,
QQGroupInfo,
UserProfileInfo,
} from '@snowluma/protocol/qq-info';
import type { PacketInfo } from '@snowluma/common/protocol-types';
function makeProfile(uin: number, uid = '', nickname = ''): UserProfileInfo {
@@ -38,7 +33,6 @@ function makePipeline(
opts: {
fetchProfileByUid?: vi.Mock;
resolveGroupJoinRequest?: vi.Mock;
resolveGroupInviteCardSequence?: vi.Mock;
} = {},
) {
const identity = IdentityService.memory('10001');
@@ -48,16 +42,14 @@ function makePipeline(
fetchProfile: async () => makeProfile(0),
fetchProfileByUid,
});
const resolveGroupJoinRequest = opts.resolveGroupJoinRequest ?? vi.fn(async () => null);
const resolveGroupInviteCardSequence =
opts.resolveGroupInviteCardSequence ?? vi.fn(async () => null);
const resolveGroupJoinRequest = opts.resolveGroupJoinRequest ?? vi.fn(async () => request());
const pipeline = new IncomingPacketPipeline({
identity,
events,
fetchGroupList: vi.fn(async () => {}),
fetchGroupMemberList: vi.fn(async () => {}),
resolveGroupJoinRequest,
resolveGroupInviteCardSequence,
rememberGroupInviteCardSequence: vi.fn(() => true),
});
const captured: QQEventVariant[] = [];
@@ -72,456 +64,184 @@ function makePipeline(
captured,
fetchProfileByUid,
resolveGroupJoinRequest,
resolveGroupInviteCardSequence,
};
}
describe('IncomingPacketPipeline / stranger resolve on group_invite', () => {
it('calls fetchProfileByUid when group_invite has fromUin=0 + fromUid', async () => {
const fetchProfileByUid = vi.fn(async (_uid: string) =>
makeProfile(950929451, 'u_stranger_abc', '小明'),
);
function request(overrides: Partial<GroupRequestInfo> = {}): GroupRequestInfo {
return {
groupId: 12345,
groupName: 'group',
targetUid: 'u_requester',
targetUin: 0,
targetName: '',
invitorUid: '',
invitorUin: 0,
invitorName: '',
operatorUid: '',
operatorUin: 0,
operatorName: '',
sequence: 42,
state: 1,
notifyType: 7,
eventType: 1,
comment: 'please',
filtered: false,
...overrides,
};
}
function requestEvent(overrides: Partial<GroupInviteEvent> = {}): GroupInviteEvent {
return {
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 12345,
fromUin: 0,
fromUid: 'u_requester',
subType: 'add',
message: '',
flag: 'add:12345:u_requester',
...overrides,
};
}
async function receive(pipeline: IncomingPacketPipeline, event: GroupInviteEvent): Promise<void> {
pipeline.registerCmd('test.cmd', () => [event]);
await pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
}
describe('IncomingPacketPipeline / group requests', () => {
it('resolves an unknown requester before dispatch', async () => {
const fetchProfileByUid = vi.fn(async () => makeProfile(222, 'u_requester', 'Alice'));
const { pipeline, captured } = makePipeline({ fetchProfileByUid });
// Plant a parser that emits the user's exact bug-report shape:
// groupId set, fromUid is a string UID, fromUin=0 (decoder fix).
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1700000000,
selfUin: 10001,
groupId: 123456789,
fromUin: 0,
fromUid: 'u_stranger_abc',
subType: 'add',
message: '',
flag: 'add:123456789:u_stranger_abc',
} as QQEventVariant,
await receive(pipeline, requestEvent());
expect(fetchProfileByUid).toHaveBeenCalledWith('u_requester');
expect(captured).toEqual([
expect.objectContaining({
fromUin: 222,
message: 'please',
flag: 'slreq:1:42:12345:1:0',
}),
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
// Let the async hook flush.
await new Promise((r) => setTimeout(r, 10));
expect(fetchProfileByUid).toHaveBeenCalledWith('u_stranger_abc');
expect(captured).toHaveLength(1);
const event = captured[0] as Extract<QQEventVariant, { kind: 'group_invite' }>;
expect(event.fromUin).toBe(950929451); // ← patched in by the async resolve
expect(event.fromUid).toBe('u_stranger_abc');
expect(event.groupId).toBe(123456789);
});
it('still emits the event when fetchProfileByUid returns no uin (fail open)', async () => {
const fetchProfileByUid = vi.fn(async () => makeProfile(0));
const { pipeline, captured } = makePipeline({ fetchProfileByUid });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 12345,
it('keeps an actionable request when its sender cannot be identified', async () => {
const { pipeline, captured } = makePipeline();
await receive(pipeline, requestEvent());
expect(captured).toEqual([
expect.objectContaining({
fromUin: 0,
fromUid: 'u_no_such_user',
subType: 'add',
message: '',
flag: 'add:12345:u_no_such_user',
} as QQEventVariant,
flag: 'slreq:1:42:12345:1:0',
}),
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
expect(fetchProfileByUid).toHaveBeenCalledOnce();
expect(captured).toHaveLength(1);
const event = captured[0] as Extract<QQEventVariant, { kind: 'group_invite' }>;
// fromUin stays 0 — but the event still goes out so the bot can
// see the join request and at least respond using the uid in the
// flag.
expect(event.fromUin).toBe(0);
expect(event.fromUid).toBe('u_no_such_user');
});
it('still emits the event when fetchProfileByUid throws', async () => {
const fetchProfileByUid = vi.fn(async () => {
throw new Error('network down');
});
const { pipeline, captured } = makePipeline({ fetchProfileByUid });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 12345,
fromUin: 0,
fromUid: 'u_x',
subType: 'add',
message: '',
flag: 'add:12345:u_x',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
expect(captured).toHaveLength(1);
expect(captured[0]).toMatchObject({ fromUin: 0, fromUid: 'u_x' });
});
it('still fetches + applies the comment when fromUin is already cached (issue #98)', async () => {
// THE bug: the comment fetch used to piggy-back on the uin-resolve
// gate, so a requester whose uin was already in cache (group member
// roster, prior @-mention, re-application…) silently lost their
// verify text — the OneBot `comment` came out empty. The comment
// and the uin resolve are now decoupled: the comment is fetched
// unconditionally; only the (wasteful) profile lookup is skipped on
// a cache hit.
const fetchProfileByUid = vi.fn(async () => makeProfile(0));
const resolveGroupJoinRequest = vi.fn(async () => ({ comment: '求通过', sequence: 42 }));
const { pipeline, captured } = makePipeline({ fetchProfileByUid, resolveGroupJoinRequest });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 12345,
fromUin: 99999,
fromUid: 'u_known',
subType: 'add',
message: '',
flag: 'add:12345:u_known',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
// Profile lookup is skipped — the uin is already known, no wasted call…
expect(fetchProfileByUid).not.toHaveBeenCalled();
// …but the comment IS still fetched from the pending-request queue
// and applied to the emitted event.
expect(resolveGroupJoinRequest).toHaveBeenCalledWith(12345, 'u_known', 'add');
expect(captured).toHaveLength(1);
const ev = captured[0] as Extract<QQEventVariant, { kind: 'group_invite' }>;
expect(ev.fromUin).toBe(99999);
expect(ev.message).toBe('求通过');
});
it('populates event.message from the pending-request queue (NapCat parity)', async () => {
// User-reported bug: SnowLuma's group_invite event surfaces a
// bare uid + uin but the requester's verify text ("你们好") never
// lands in the OneBot `comment` field. NapCat shows the text via
// `notify.postscript` from `getGroupNotifies`; we mirror that by
// doing an `OIDB 0x10C0 fetchGroupRequests` lookup in the async
// pre-dispatch hook and patching `event.message` from the matching
// row's `comment`.
const fetchProfileByUid = vi.fn(async () =>
makeProfile(1957003260, 'u_UVLHYYqba27WPUzTrdZlCA', 'KitaIkuyo'),
);
const resolveGroupJoinRequest = vi.fn(async () => ({
comment: '你们好',
sequence: 1779543612823906,
}));
const { pipeline, captured } = makePipeline({
fetchProfileByUid,
resolveGroupJoinRequest,
});
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 950929451,
fromUin: 0,
fromUid: 'u_UVLHYYqba27WPUzTrdZlCA',
subType: 'add',
message: '',
flag: 'add:950929451:u_UVLHYYqba27WPUzTrdZlCA',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
it('fetches the comment even when the requester identity is already known', async () => {
const { pipeline, captured, resolveGroupJoinRequest, fetchProfileByUid } = makePipeline();
await receive(pipeline, requestEvent({ fromUin: 222 }));
expect(resolveGroupJoinRequest).toHaveBeenCalledWith(
950929451,
'u_UVLHYYqba27WPUzTrdZlCA',
12345,
'u_requester',
'add',
undefined,
);
expect(captured).toHaveLength(1);
const ev = captured[0] as Extract<QQEventVariant, { kind: 'group_invite' }>;
expect(ev.message).toBe('你们好');
expect(ev.fromUin).toBe(1957003260);
expect(fetchProfileByUid).not.toHaveBeenCalled();
expect(captured).toEqual([expect.objectContaining({ fromUin: 222, message: 'please' })]);
});
it('replaces the legacy identity flag with a canonical approval tuple', async () => {
const resolveGroupJoinRequest = vi.fn(async () => ({
groupId: 950929451,
groupName: 'group',
targetUid: 'u_target',
targetUin: 1957003260,
targetName: 'target',
invitorUid: 'u_inviter',
invitorUin: 0,
invitorName: '',
operatorUid: '',
operatorUin: 0,
operatorName: '',
sequence: 1779543612823906,
state: 1,
eventType: 1,
comment: '你们好',
filtered: true,
}));
const { pipeline, captured } = makePipeline({ resolveGroupJoinRequest });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 950929451,
fromUin: 1957003260,
fromUid: 'u_target',
subType: 'add',
message: '',
flag: 'add:950929451:u_target',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
expect(captured).toHaveLength(1);
expect(captured[0]).toMatchObject({
message: '你们好',
flag: 'slreq:1:1779543612823906:950929451:1:1',
});
});
it('uses the private invite-card msgseq for a self-invite approval tuple', async () => {
const resolveGroupJoinRequest = vi.fn(async () => ({
groupId: 999,
groupName: 'group',
targetUid: 'self',
targetUin: 10001,
targetName: 'self',
invitorUid: 'u_inviter',
invitorUin: 456,
invitorName: 'inviter',
operatorUid: '',
operatorUin: 0,
operatorName: '',
sequence: 111,
state: 1,
eventType: 2,
comment: '',
filtered: false,
}));
const resolveGroupInviteCardSequence = vi.fn(async () => 778899);
const { pipeline, captured } = makePipeline({
resolveGroupJoinRequest,
resolveGroupInviteCardSequence,
});
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 999,
fromUin: 456,
fromUid: 'u_inviter',
subType: 'invite',
message: '',
flag: 'invite:999:u_inviter',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
expect(resolveGroupInviteCardSequence).toHaveBeenCalledWith(999);
expect(captured[0]).toMatchObject({
flag: 'slreq:1:778899:999:2:0',
invitedUin: 10001,
invitedUid: 'self',
});
});
it('fills the invitee from the pending-request row (#394)', async () => {
const resolveGroupJoinRequest = vi.fn(async () => ({
groupId: 1,
groupName: 'group',
targetUid: 'u_invitee',
targetUin: 789,
targetName: 'invitee',
invitorUid: 'u_inviter',
invitorUin: 456,
invitorName: 'inviter',
operatorUid: '',
operatorUin: 0,
operatorName: '',
sequence: 42,
state: 1,
eventType: 2,
comment: '',
filtered: false,
}));
const { pipeline, captured } = makePipeline({ resolveGroupJoinRequest });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 1,
fromUin: 456,
fromUid: 'u_inviter',
subType: 'invite',
message: '',
flag: 'invite:1:u_inviter',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
expect(captured[0]).toMatchObject({
fromUin: 456,
invitedUin: 789,
invitedUid: 'u_invitee',
});
});
it('runs the profile + request lookups in parallel (independent failures)', async () => {
// Profile lookup succeeds, request lookup fails — event should
// still carry the resolved uin but no comment. The dispatch
// doesn't block on the slower / failing path.
const fetchProfileByUid = vi.fn(async () => makeProfile(12345, 'u_x', 'OK'));
const resolveGroupJoinRequest = vi.fn(async () => {
throw new Error('queue lookup failed');
});
const { pipeline, captured } = makePipeline({
fetchProfileByUid,
resolveGroupJoinRequest,
});
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 1,
fromUin: 0,
fromUid: 'u_x',
subType: 'add',
message: '',
flag: 'add:1:u_x',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
const ev = captured[0] as Extract<QQEventVariant, { kind: 'group_invite' }>;
expect(ev.fromUin).toBe(12345); // profile succeeded
expect(ev.message).toBe(''); // request lookup threw, message left empty
});
it('invite subtype matches the request on invitorUid (not targetUid)', async () => {
// For `subType: 'invite'` (group member invited the bot or another
// user), the pending-request row's REQUESTER field is
// `invitorUid` not `targetUid`. The dep contract makes this
// explicit so the bridge facade can route the lookup correctly.
const resolveGroupJoinRequest = vi.fn(async () => ({
comment: 'come join',
sequence: 999,
}));
const { pipeline } = makePipeline({ resolveGroupJoinRequest });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 1,
fromUin: 0,
fromUid: 'u_inviter',
subType: 'invite',
message: '',
flag: 'invite:1:u_inviter',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
expect(resolveGroupJoinRequest).toHaveBeenCalledWith(1, 'u_inviter', 'invite');
});
it('does not treat fromUin === groupId as an unresolved identity', async () => {
const fetchProfileByUid = vi.fn(async () =>
makeProfile(1957003260, 'u_kitaikuyo', 'KitaIkuyo'),
it('uses the sender and approval tuple from the matched filtered request', async () => {
const resolveGroupJoinRequest = vi.fn(async () =>
request({ targetUin: 222, filtered: true }),
);
const { pipeline, captured } = makePipeline({ fetchProfileByUid });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 950929451,
fromUin: 950929451,
fromUid: 'u_kitaikuyo',
subType: 'add',
message: '',
flag: 'add:950929451:u_kitaikuyo',
} as QQEventVariant,
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
const { pipeline, captured, fetchProfileByUid } = makePipeline({ resolveGroupJoinRequest });
await receive(pipeline, requestEvent());
expect(fetchProfileByUid).not.toHaveBeenCalled();
expect(captured).toHaveLength(1);
expect(captured[0]).toMatchObject({
fromUin: 950929451,
fromUid: 'u_kitaikuyo',
groupId: 950929451,
});
expect(captured).toEqual([
expect.objectContaining({ fromUin: 222, flag: 'slreq:1:42:12345:1:1' }),
]);
});
it('skips both lookups when fromUid is empty (no uid to look up by)', async () => {
const fetchProfileByUid = vi.fn(async () => makeProfile(0));
const resolveGroupJoinRequest = vi.fn(async () => null);
const { pipeline, captured } = makePipeline({ fetchProfileByUid, resolveGroupJoinRequest });
pipeline.registerCmd('test.cmd', () => [
{
kind: 'group_invite',
time: 1,
selfUin: 10001,
groupId: 12345,
fromUin: 0,
fromUid: '',
subType: 'add',
message: '',
flag: 'add:12345:',
} as QQEventVariant,
it('matches an invitation approval by inviter and invitee', async () => {
const resolveGroupJoinRequest = vi.fn(async () =>
request({
invitorUid: 'u_inviter',
invitorUin: 456,
targetUid: 'u_invitee',
targetUin: 789,
eventType: 22,
notifyType: 5,
}),
);
const { pipeline, captured } = makePipeline({ resolveGroupJoinRequest });
await receive(
pipeline,
requestEvent({ subType: 'invite', fromUid: 'u_inviter', invitedUid: 'u_invitee' }),
);
expect(resolveGroupJoinRequest).toHaveBeenCalledWith(
12345,
'u_inviter',
'invite',
'u_invitee',
);
expect(captured).toEqual([
expect.objectContaining({
fromUin: 456,
invitedUin: 789,
invitedUid: 'u_invitee',
flag: 'slreq:1:42:12345:22:0',
}),
]);
});
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
await new Promise((r) => setTimeout(r, 10));
it('leaves self invitations to the card without querying or waiting', async () => {
const { pipeline, captured, resolveGroupJoinRequest, fetchProfileByUid } = makePipeline();
await receive(pipeline, requestEvent({ subType: 'invite', invitedUin: 10001 }));
expect(captured).toEqual([]);
expect(resolveGroupJoinRequest).not.toHaveBeenCalled();
expect(fetchProfileByUid).not.toHaveBeenCalled();
});
it('does not substitute a list sequence for the self-invite card sequence', async () => {
const resolveGroupJoinRequest = vi.fn(async () =>
request({ eventType: 2, targetUin: 10001 }),
);
const { pipeline, captured } = makePipeline({ resolveGroupJoinRequest });
await receive(pipeline, requestEvent({ subType: 'invite' }));
expect(captured).toEqual([]);
});
// No uid → nothing to query by → neither lookup runs, event still emits.
it.each(['missing', 'failed'])(
'does not emit an incomplete request after a %s lookup',
async (outcome) => {
const resolveGroupJoinRequest =
outcome === 'missing'
? vi.fn(async () => null)
: vi.fn(async () => {
throw new Error('lookup failed');
});
const { pipeline, captured, fetchProfileByUid } = makePipeline({
resolveGroupJoinRequest,
});
await receive(pipeline, requestEvent());
expect(captured).toEqual([]);
expect(resolveGroupJoinRequest).toHaveBeenCalledOnce();
expect(fetchProfileByUid).not.toHaveBeenCalled();
},
);
it('does not treat a requester UIN equal to the group number as unresolved', async () => {
const { pipeline, captured, identity, fetchProfileByUid } = makePipeline();
await receive(pipeline, requestEvent({ fromUin: 12345 }));
expect(fetchProfileByUid).not.toHaveBeenCalled();
expect(captured[0]).toMatchObject({ fromUin: 12345 });
expect(identity.findUinByUid('u_requester')).toBe(12345);
});
it('rejects a request without its requester UID', async () => {
const { pipeline, captured, resolveGroupJoinRequest } = makePipeline();
await receive(pipeline, requestEvent({ fromUid: '' }));
expect(captured).toEqual([]);
expect(resolveGroupJoinRequest).not.toHaveBeenCalled();
expect(captured).toHaveLength(1);
});
});
@@ -559,6 +279,7 @@ describe('IncomingPacketPipeline / #1 group card self-heal', () => {
fetchGroupList: vi.fn(async () => {}),
fetchGroupMemberList: vi.fn(async () => {}),
resolveGroupJoinRequest: vi.fn(async () => null),
rememberGroupInviteCardSequence: vi.fn(() => true),
});
return { identity, pipeline };
}
@@ -672,6 +393,7 @@ describe('IncomingPacketPipeline / group_card_change from message traffic', () =
fetchGroupList: vi.fn(async () => {}),
fetchGroupMemberList: vi.fn(async () => {}),
resolveGroupJoinRequest: vi.fn(async () => null),
rememberGroupInviteCardSequence: vi.fn(() => true),
});
const captured: QQEventVariant[] = [];
events.onAny((e) => captured.push(e as QQEventVariant));
@@ -734,43 +456,3 @@ describe('IncomingPacketPipeline / group_card_change from message traffic', () =
expect(captured.some((e) => e.kind === 'group_card_change')).toBe(false);
});
});
describe('IncomingPacketPipeline / invite-card pending application', () => {
const friendMessage = (card?: {
inviteCardGroupUin: number;
inviteCardSequence: number;
}): QQEventVariant => ({
kind: 'friend_message',
time: 1,
selfUin: 10001,
senderUin: 222,
senderNick: 'Bob',
msgSeq: 1,
msgId: 1,
elements: [],
...card,
});
it('remembers parsed invite-card facts through the optional port', () => {
const rememberGroupInviteCardSequence = vi.fn();
const pipeline = new IncomingPacketPipeline({
identity: IdentityService.memory('10001'),
events: new BridgeEventBus(),
fetchGroupList: vi.fn(async () => {}),
fetchGroupMemberList: vi.fn(async () => {}),
resolveGroupJoinRequest: vi.fn(async () => null),
rememberGroupInviteCardSequence,
});
pipeline.registerCmd('test.cmd', () => [
friendMessage({ inviteCardGroupUin: 12345, inviteCardSequence: 778899 }),
]);
pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo);
expect(rememberGroupInviteCardSequence).toHaveBeenCalledWith(12345, 778899);
});
it('does not remember when the port is omitted or the event has no card', () => {
const { pipeline } = makePipeline();
pipeline.registerCmd('test.cmd', () => [friendMessage()]);
expect(() => pipeline.process({ serviceCmd: 'test.cmd' } as PacketInfo)).not.toThrow();
});
});
@@ -121,7 +121,26 @@ function makePipeline(
events,
fetchGroupList: opts.fetchGroupList ?? vi.fn(async () => {}),
fetchGroupMemberList: opts.fetchGroupMemberList ?? vi.fn(async () => {}),
resolveGroupJoinRequest: vi.fn(async () => null),
resolveGroupJoinRequest: vi.fn(async () => ({
groupId: 123,
groupName: '',
targetUid: 'u_stranger',
targetUin: 0,
targetName: '',
invitorUid: '',
invitorUin: 0,
invitorName: '',
operatorUid: '',
operatorUin: 0,
operatorName: '',
sequence: 42,
state: 1,
notifyType: 7,
eventType: 1,
comment: '',
filtered: false,
})),
rememberGroupInviteCardSequence: vi.fn(() => true),
});
const dispatched: Array<{ event: QQEventVariant; req: number | undefined }> = [];
events.onAny((event) => {