Simplify Hub channel management
Refactor Hub to treat each UIN as a single active Channel: channelsByUin_ now maps UIN -> Channel (no Set of IDs), primary selection logic and primaryByUin_/pickPrimary were removed, and addChannel replaces any existing channel for a UIN (disposing the previous one). Updated add/remove flows to update channelsById_ and channelsByUin_ accordingly and to bring cores online/offline in the simplified model. Also: removed extensive JSDoc/comment blocks from channel/adapter/channel-ctx/core/hub files to slim the public API surface, reordered/cleaned exports in packages/channel/src/index.ts, added a new hub-events.ts placeholder, and cleaned up a test helper comment. Disposal of replaced channels is wrapped with a try/catch and logged at debug level.
This commit is contained in:
手瓜一十雪
@@ -2,64 +2,14 @@ import type { Logger } from '@snowluma/common/logger';
|
||||
import type { Channel } from './channel';
|
||||
import type { ChannelKind } from './channel-interface';
|
||||
|
||||
/**
|
||||
* Sink the `Hub` (in `@snowluma/core`) exposes to a `ChannelAdapter`
|
||||
* so the adapter can publish discovered channels and tear them down
|
||||
* when the underlying runtime says they're gone.
|
||||
*
|
||||
* The host is intentionally narrow: adapters cannot reach into the
|
||||
* hub's bookkeeping, only emit `addChannel` / `removeChannel` events.
|
||||
* Anything else (Core construction, lifecycle event emission, primary
|
||||
* selection) is the hub's concern.
|
||||
*/
|
||||
export interface ChannelAdapterHost {
|
||||
/** Publish a freshly-online channel. The hub dedupes by
|
||||
* `channel.id`; calling this with an already-registered id is a
|
||||
* no-op (and logs a warning in dev mode). */
|
||||
addChannel(channel: Channel): void;
|
||||
/** Tear down the channel with the given `id`. The hub calls
|
||||
* `channel.dispose()` on its way out and disposes any `Core`
|
||||
* bound to this channel. */
|
||||
removeChannel(channelId: string): void;
|
||||
/** Adapter-scoped logger. Adapters should prefer this over a
|
||||
* module-level logger so the hub can label log lines per
|
||||
* adapter kind. */
|
||||
readonly log: Logger;
|
||||
}
|
||||
|
||||
/**
|
||||
* A `ChannelAdapter` bridges an external login source (NTQQ hook,
|
||||
* future pure-protocol client, …) and the `Hub`'s pool of live
|
||||
* `Channel`s. Each adapter encapsulates:
|
||||
*
|
||||
* - how its runtime is started / stopped,
|
||||
* - how it discovers new logins (events from a watcher, polling,
|
||||
* manual `start` from WebUI, …),
|
||||
* - how it materialises a concrete `Channel` subclass for each login,
|
||||
* - how it cleans up when a login disappears.
|
||||
*
|
||||
* The hub itself remains transport-agnostic: it only knows about
|
||||
* `Channel` instances coming and going through `ChannelAdapterHost`.
|
||||
*
|
||||
* Lifecycle expectations:
|
||||
*
|
||||
* - `start(host)` is called once when the hub initialises. The
|
||||
* adapter retains the `host` and is free to call into it from any
|
||||
* point until `dispose()` is invoked.
|
||||
* - `dispose()` MUST be idempotent and MUST tear down every channel
|
||||
* the adapter previously published (the hub calls
|
||||
* `removeChannel` for them in turn).
|
||||
*/
|
||||
export interface ChannelAdapter {
|
||||
/** Discriminator that mirrors the `kind` of the channels this
|
||||
* adapter produces. Used by the hub for diagnostics and to
|
||||
* refuse duplicate registrations of the same kind. */
|
||||
readonly kind: ChannelKind;
|
||||
|
||||
/** Hook the adapter into a freshly-built hub. Called exactly
|
||||
* once per adapter instance. */
|
||||
start(host: ChannelAdapterHost): Promise<void> | void;
|
||||
|
||||
/** Stop the adapter and tear down every channel it produced. */
|
||||
dispose(): Promise<void> | void;
|
||||
}
|
||||
|
||||
@@ -3,60 +3,18 @@ import type { PacketInfo } from '@snowluma/common/protocol-types';
|
||||
import type { Channel } from './channel';
|
||||
import type { ChannelKind } from './channel-interface';
|
||||
|
||||
/**
|
||||
* `ChannelCtx` — the capability POJO that crosses the `Channel → Core`
|
||||
* boundary in the v3 hexagonal layout.
|
||||
*
|
||||
* `Core` (in `@snowluma/core`) is constructed with one of these and
|
||||
* **never** holds a `Channel` instance directly. That keeps `@snowluma/core`
|
||||
* free of any adapter implementation detail (and even of the abstract
|
||||
* `Channel` class — only this capability surface is consumed).
|
||||
*
|
||||
* The ctx is intentionally a plain object, not an interface backed by a
|
||||
* class:
|
||||
* - `Hub` builds it once per online channel via `makeChannelCtx(ch)`
|
||||
* and emits it on its lifecycle events,
|
||||
* - `Core` captures the closure and treats it as immutable for the
|
||||
* channel's lifetime,
|
||||
* - `dispose` is the channel's tear-down trigger; `Hub` invokes it
|
||||
* when the underlying transport reports gone.
|
||||
*/
|
||||
export interface ChannelCtx {
|
||||
/** QQ number this channel carries. Equivalent to `Channel.uin`. */
|
||||
readonly uin: string;
|
||||
/** Transport family (`'inject'` for hook, `'socket'` for future
|
||||
* pure-socket). Equivalent to `Channel.kind`. */
|
||||
readonly kind: ChannelKind;
|
||||
|
||||
/** Wire-level send. Core's `sendRawPacket` forwards every outbound
|
||||
* request through this single function. */
|
||||
readonly sendRawPacket: (
|
||||
serviceCmd: string,
|
||||
body: Uint8Array,
|
||||
timeoutMs?: number,
|
||||
) => Promise<SendPacketResult>;
|
||||
|
||||
/** Subscribe (or detach with `null`) the single inbound packet
|
||||
* handler. Called once by `Core` in its constructor so its
|
||||
* `IncomingPacketPipeline` receives every frame the channel
|
||||
* delivers. */
|
||||
readonly onPacket: (handler: ((pkt: PacketInfo) => void) | null) => void;
|
||||
|
||||
/** Release channel-side resources. Invoked by `Hub` when the
|
||||
* channel goes offline; idempotent on the channel side. */
|
||||
readonly dispose: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Project a concrete `Channel` instance into its capability POJO. The
|
||||
* returned object captures the channel via closure so consumers see
|
||||
* only the four functions plus `uin` / `kind` — never the class
|
||||
* itself.
|
||||
*
|
||||
* Used exclusively by `Hub` (in `@snowluma/core`); adapters publish
|
||||
* `Channel` instances and `Hub` is the one place that turns each into
|
||||
* a ctx before handing it to `Core`.
|
||||
*/
|
||||
export function makeChannelCtx(channel: Channel): ChannelCtx {
|
||||
return {
|
||||
uin: channel.uin,
|
||||
|
||||
@@ -2,30 +2,6 @@ import type { SendPacketResult } from '@snowluma/common/packet-sender';
|
||||
import type { PacketInfo } from '@snowluma/common/protocol-types';
|
||||
import type { ChannelInterface, ChannelKind } from './channel-interface';
|
||||
|
||||
/**
|
||||
* Abstract base for every concrete channel transport (`HookChannel`,
|
||||
* `SocketChannel`, …). Strictly transport: knows about wire-level
|
||||
* send, packet delivery to a single subscriber, and lifecycle.
|
||||
*
|
||||
* Subclasses provide:
|
||||
*
|
||||
* 1. `kind` / `id` / `uin` — discriminators used by the `Hub` (in
|
||||
* `@snowluma/core`) to index live transports.
|
||||
* 2. `sendRawPacket` — the actual outbound wire path. `HookChannel`
|
||||
* forwards through a `PacketSender` (named-pipe to QQ.exe);
|
||||
* `SocketChannel` will eventually drive a long-lived TCP login
|
||||
* session.
|
||||
*
|
||||
* Inbound packets are routed via the `packetHandler_` slot: `Hub`
|
||||
* wires `ChannelCtx.onPacket` (which calls `setPacketHandler` on
|
||||
* this channel) so each `Core` instance subscribes exactly once, and
|
||||
* the owning `ChannelAdapter` calls `deliverPacket(pkt)` for every
|
||||
* frame the underlying runtime hands it.
|
||||
*
|
||||
* `Channel` knows NOTHING about identity caches, the event bus, the
|
||||
* api hub, or the cmd-dispatch pipeline. That entire account-layer
|
||||
* machinery lives on `Core` (`@snowluma/core`).
|
||||
*/
|
||||
export abstract class Channel implements ChannelInterface {
|
||||
abstract readonly kind: ChannelKind;
|
||||
abstract readonly id: string;
|
||||
|
||||
@@ -1,17 +1,5 @@
|
||||
// `@snowluma/channel` — the abstract transport layer + capability POJO
|
||||
// + adapter port. Concrete transports (HookChannel / SocketChannel)
|
||||
// and their adapters live in dedicated packages
|
||||
// (`@snowluma/channel-hook`, `@snowluma/channel-socket`); the
|
||||
// multi-account host lives in `@snowluma/core` as `Hub`. Nothing in
|
||||
// this package depends on a runtime — it's pure interfaces and the
|
||||
// `Channel` abstract base.
|
||||
|
||||
// ─── Transport abstract base ───────────────────────────────────────
|
||||
export { type ChannelAdapter, type ChannelAdapterHost } from './adapter';
|
||||
export { Channel } from './channel';
|
||||
export { makeChannelCtx, type ChannelCtx } from './channel-ctx';
|
||||
export { type ChannelInterface, type ChannelKind } from './channel-interface';
|
||||
|
||||
// ─── Capability POJO (handed to Core in @snowluma/core) ────────────
|
||||
export { makeChannelCtx, type ChannelCtx } from './channel-ctx';
|
||||
|
||||
// ─── Adapter port ──────────────────────────────────────────────────
|
||||
export { type ChannelAdapter, type ChannelAdapterHost } from './adapter';
|
||||
|
||||
@@ -50,18 +50,6 @@ export class Core implements CoreCtx {
|
||||
private clientSeq_ = 100000000 + (Date.now() % 1000000000);
|
||||
private msgRandom_ = (Date.now() & 0xFFFFFFFF) >>> 0;
|
||||
private disposed_ = false;
|
||||
|
||||
/**
|
||||
* @param channelCtx The capability POJO this core rides on top of.
|
||||
* `Core` calls `channelCtx.sendRawPacket` and
|
||||
* subscribes via `channelCtx.onPacket`. Built by
|
||||
* `Hub` from a concrete `Channel` via `makeChannelCtx`.
|
||||
* @param identity Optional pre-built `IdentityService`. The default
|
||||
* opens the on-disk SQLite store at
|
||||
* `data/<uin>/snowluma_identity.db`; tests usually
|
||||
* pass `IdentityService.memory(uin)` to avoid touching
|
||||
* the filesystem.
|
||||
*/
|
||||
constructor(channelCtx: ChannelCtx, identity: IdentityService = IdentityService.openForUin(channelCtx.uin)) {
|
||||
this.channelCtx_ = channelCtx;
|
||||
this.kind = channelCtx.kind;
|
||||
|
||||
+28
-130
@@ -11,31 +11,15 @@ import type { CoreCtx } from './core-ctx';
|
||||
|
||||
const log = createLogger('Hub');
|
||||
|
||||
/**
|
||||
* Top-level lifecycle events fired by `Hub`. Inbound consumers (OneBot,
|
||||
* WebUI backend, future MCP) subscribe via `hub.events.on(...)`.
|
||||
*/
|
||||
export type HubEventMap = {
|
||||
/** Fired whenever a fresh `Core` is online for a UIN. Either the
|
||||
* account just came up, OR the primary channel was swapped under it
|
||||
* (in which case `'core-offline'` for the same UIN fired first). */
|
||||
'core-online': CoreCtx;
|
||||
/** Fired when an account goes offline OR right before a primary-
|
||||
* channel swap rebuilds its `Core`. */
|
||||
'core-online': CoreCtx;
|
||||
'core-offline': { uin: string };
|
||||
};
|
||||
|
||||
export type HubEventHandler<K extends keyof HubEventMap> = (payload: HubEventMap[K]) => void;
|
||||
export type HubEventDisposer = () => void;
|
||||
|
||||
/**
|
||||
* Tiny typed event emitter scoped to `Hub`'s lifecycle events. Kept
|
||||
* inline rather than reaching for `@snowluma/protocol/event-bus`
|
||||
* (which is hard-typed to QQ event variants) or a generic dependency.
|
||||
*
|
||||
* Subscriber errors are swallowed with a warning so one bad consumer
|
||||
* never blocks the rest.
|
||||
*/
|
||||
|
||||
class HubEvents {
|
||||
private readonly listeners_ = new Map<keyof HubEventMap, Set<(p: unknown) => void>>();
|
||||
|
||||
@@ -67,39 +51,10 @@ class HubEvents {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* `Hub` — the single multi-account entry point and **owner of all
|
||||
* channels and cores**. Replaces the v2 split between
|
||||
* `ChannelManager` (transport pool, lived in `@snowluma/channel`) and
|
||||
* `AccountManager` (account wrapper, lived in `@snowluma/core`).
|
||||
*
|
||||
* Responsibilities (folded from both predecessors):
|
||||
*
|
||||
* 1. Register one `ChannelAdapter` per transport kind (hook,
|
||||
* socket, …). Adapters publish/withdraw `Channel`s through the
|
||||
* `ChannelAdapterHost` interface that Hub implements per-adapter.
|
||||
* 2. Track every live channel (`channelsById_` / `channelsByUin_`)
|
||||
* and pick a primary per UIN (`primaryByUin_`) by adapter rank
|
||||
* (`primaryPriority`: inject > socket today).
|
||||
* 3. Build a `ChannelCtx` capability POJO from each primary channel
|
||||
* and materialise a `Core` on top.
|
||||
* 4. Emit two top-level lifecycle events: `'core-online'` (with the
|
||||
* `CoreCtx` projection) and `'core-offline'` (with `{ uin }`).
|
||||
* Inbound consumers (OneBot, WebUI) only see these two events;
|
||||
* they never touch the underlying `Channel` / `Core` classes.
|
||||
*
|
||||
* Single-owner invariant: every `Channel` and every `Core` instance
|
||||
* is referenced exactly once — from inside `Hub`. External consumers
|
||||
* hold only `CoreCtx` projections, which become stale when `Core`
|
||||
* is disposed (handlers should unsubscribe in their `'core-offline'`
|
||||
* handler).
|
||||
*/
|
||||
export class Hub {
|
||||
// ─── adapter / channel pool (was ChannelManager) ──────────────────
|
||||
private readonly adapters_ = new Map<ChannelKind, ChannelAdapter>();
|
||||
private readonly channelsById_ = new Map<string, Channel>();
|
||||
private readonly channelsByUin_ = new Map<string, Set<string>>();
|
||||
private readonly primaryByUin_ = new Map<string, string>();
|
||||
private readonly channelsByUin_ = new Map<string, Channel>();
|
||||
|
||||
// ─── core pool (was AccountManager) ──────────────────────────────
|
||||
private readonly cores_ = new Map<string, Core>();
|
||||
@@ -170,7 +125,6 @@ export class Hub {
|
||||
this.cores_.clear();
|
||||
this.channelsById_.clear();
|
||||
this.channelsByUin_.clear();
|
||||
this.primaryByUin_.clear();
|
||||
this.events.clear();
|
||||
}
|
||||
|
||||
@@ -187,11 +141,9 @@ export class Hub {
|
||||
return [...this.cores_.values()];
|
||||
}
|
||||
|
||||
/** Primary channel for a UIN (the one driving the current `Core`),
|
||||
* or `null` if the UIN is offline. */
|
||||
/** Channel for a UIN, or `null` if the UIN is offline. */
|
||||
getChannel(uin: string): Channel | null {
|
||||
const id = this.primaryByUin_.get(uin);
|
||||
return id ? this.channelsById_.get(id) ?? null : null;
|
||||
return this.channelsByUin_.get(uin) ?? null;
|
||||
}
|
||||
|
||||
getOnlineUins(): readonly string[] {
|
||||
@@ -215,37 +167,28 @@ export class Hub {
|
||||
adapterLog.warn('addChannel: id=%s already registered, ignoring', channel.id);
|
||||
return;
|
||||
}
|
||||
this.channelsById_.set(channel.id, channel);
|
||||
const existing = this.channelsByUin_.get(channel.uin);
|
||||
if (existing) {
|
||||
adapterLog.warn('addChannel: UIN=%s already has channel=%s, replacing with %s',
|
||||
channel.uin, existing.id, channel.id);
|
||||
this.channelsById_.delete(existing.id);
|
||||
this.channelsById_.set(channel.id, channel);
|
||||
this.channelsByUin_.set(channel.uin, channel);
|
||||
|
||||
let ids = this.channelsByUin_.get(channel.uin);
|
||||
if (!ids) {
|
||||
ids = new Set();
|
||||
this.channelsByUin_.set(channel.uin, ids);
|
||||
}
|
||||
ids.add(channel.id);
|
||||
|
||||
const prevPrimaryId = this.primaryByUin_.get(channel.uin);
|
||||
const newPrimary = this.pickPrimary(channel.uin);
|
||||
if (!newPrimary) return;
|
||||
this.primaryByUin_.set(channel.uin, newPrimary.id);
|
||||
|
||||
const isFirst = prevPrimaryId === undefined;
|
||||
const primaryChanged = !isFirst && prevPrimaryId !== newPrimary.id;
|
||||
|
||||
if (isFirst) {
|
||||
log.debug('account online: UIN=%s via %s', channel.uin, newPrimary.id);
|
||||
this.bringCoreOnline(channel.uin, newPrimary);
|
||||
} else if (primaryChanged) {
|
||||
// A higher-priority transport (e.g. socket overtakes inject)
|
||||
// came online. Tear down the old core and rebuild around the
|
||||
// new primary so apis/identity stay consistent with the active
|
||||
// transport. OneBot already tears down + rebuilds its instance
|
||||
// on offline→online; we faithfully drive that.
|
||||
log.debug('primary switched: UIN=%s %s → %s',
|
||||
channel.uin, prevPrimaryId, newPrimary.id);
|
||||
this.bringCoreOffline(channel.uin);
|
||||
this.bringCoreOnline(channel.uin, newPrimary);
|
||||
this.bringCoreOnline(channel.uin, channel);
|
||||
|
||||
try { existing.dispose(); } catch (err) {
|
||||
adapterLog.debug('channel.dispose() threw: %s',
|
||||
err instanceof Error ? err.message : String(err));
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
this.channelsById_.set(channel.id, channel);
|
||||
this.channelsByUin_.set(channel.uin, channel);
|
||||
log.debug('account online: UIN=%s via %s', channel.uin, channel.id);
|
||||
this.bringCoreOnline(channel.uin, channel);
|
||||
}
|
||||
|
||||
private removeChannel(id: string, adapterLog: Logger): void {
|
||||
@@ -253,25 +196,10 @@ export class Hub {
|
||||
if (!channel) return;
|
||||
this.channelsById_.delete(id);
|
||||
|
||||
const ids = this.channelsByUin_.get(channel.uin);
|
||||
if (ids) {
|
||||
ids.delete(id);
|
||||
if (ids.size === 0) this.channelsByUin_.delete(channel.uin);
|
||||
}
|
||||
|
||||
const wasPrimary = this.primaryByUin_.get(channel.uin) === id;
|
||||
if (wasPrimary) this.primaryByUin_.delete(channel.uin);
|
||||
|
||||
const replacement = this.pickPrimary(channel.uin);
|
||||
if (replacement) {
|
||||
this.primaryByUin_.set(channel.uin, replacement.id);
|
||||
if (wasPrimary) {
|
||||
adapterLog.debug('primary failover: UIN=%s %s → %s',
|
||||
channel.uin, id, replacement.id);
|
||||
this.bringCoreOffline(channel.uin);
|
||||
this.bringCoreOnline(channel.uin, replacement);
|
||||
}
|
||||
} else if (wasPrimary) {
|
||||
const current = this.channelsByUin_.get(channel.uin);
|
||||
const wasCurrent = current?.id === id;
|
||||
if (wasCurrent) {
|
||||
this.channelsByUin_.delete(channel.uin);
|
||||
log.debug('account offline: UIN=%s', channel.uin);
|
||||
this.bringCoreOffline(channel.uin);
|
||||
}
|
||||
@@ -307,34 +235,4 @@ export class Hub {
|
||||
}
|
||||
}
|
||||
|
||||
private pickPrimary(uin: string): Channel | null {
|
||||
const ids = this.channelsByUin_.get(uin);
|
||||
if (!ids || ids.size === 0) return null;
|
||||
|
||||
let best: Channel | null = null;
|
||||
let bestRank = -Infinity;
|
||||
for (const id of ids) {
|
||||
const channel = this.channelsById_.get(id);
|
||||
if (!channel) continue;
|
||||
const rank = primaryPriority(channel.kind);
|
||||
if (rank > bestRank) {
|
||||
best = channel;
|
||||
bestRank = rank;
|
||||
}
|
||||
}
|
||||
return best;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Higher wins. Today we prefer the in-process hook over the future
|
||||
* pure-socket runtime — operators with QQ.exe running trust the hook
|
||||
* path more, and the socket runtime is unproven. Single knob to flip
|
||||
* (or surface as config) once the socket runtime is hardened.
|
||||
*/
|
||||
function primaryPriority(kind: ChannelKind): number {
|
||||
switch (kind) {
|
||||
case 'inject': return 100;
|
||||
case 'socket': return 50;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,13 +36,6 @@ function makeGroup(members: GroupMemberInfo[] = []): QQGroupInfo {
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* `Core` wired up with a member-list interceptor. The interceptor
|
||||
* lives on `core.apis.contacts.fetchGroupMemberList`; direct property
|
||||
* assignment is enough because the ApiHub holds own-property methods,
|
||||
* not prototype ones. The underlying `HookChannel` is exposed so tests
|
||||
* can drive `deliverPacket` directly.
|
||||
*/
|
||||
function buildCore(identity: IdentityService, refreshedMembers: GroupMemberInfo[]) {
|
||||
const channel = new HookChannel(identity.uin);
|
||||
const core = new Core(makeChannelCtx(channel), identity);
|
||||
|
||||
Reference in New Issue
Block a user